ETF holding classification repair
Tracks #1794, #1795, and the bounded zero-fetch recovery in #1810.
The normal operational path is a fresh, read-only --scan-current preview. It discovers the repair scope and cache work from the target database, seals the result in a version-3 report, and binds apply or rollback to its hashes. This document records no approval to stop services, apply, roll back, invoke a model, or perform a historical fetch.
Parser, scanner, and detector behavior
- EZMoney treats live PCF
AssetCodeas the primary contract:GD,ST, andBDmean futures, stock, and bond. AGDrow withMTHhas identityCODE:YYYYMM. Historical rows lackingAssetCodeuse the established code/name fallback. An unsupported nonblankAssetCodeis retained as asset typeotherwith no quantity unit and a warning; only malformed nonblankMTHon a future is a data-contract error. - Fubon recognizes the audited futures suffix and explicit summary rows. A valid Taiwan identifier takes precedence over a name-only classification.
- The holdings delta detector removes a deployment-only EZMoney bare-to-dated change only when a catalog fund has exactly one bare product on one snapshot and exactly one dated contract for that product on the other. It retains real quantity changes and dated-month rollovers. Multiple contracts for the same product remain distinct, as do non-EZMoney assets.
- The dynamic repair scan is limited to EZMoney and Fubon. It does not guess a repair for an unknown source category retained as
other; ambiguous derivative-shaped records require separate review.
The scanner visits every date and fund, including disabled funds, for EZMoney and Fubon. It classifies explicit Fubon footers for deletion and supported stale stock/share futures for reclassification. Ambiguous derivative holdings or a chip source summary that depends on a repaired holding make the preview refuse; resolve those separately before authorizing a repair.
The 2026-09-30 (Asia/Taipei) read-only observation was 9,264 current candidates: 517 EZMoney reclassifications, 3,871 Fubon reclassifications, and 4,876 Fubon footers. This is evidence only. Execution uses the counts, identities, and date range in its own sealed preview.
Archived fixed-scope evidence
The earlier 9,169-row manifest path remains available for archival recovery. Do not mix its six manifest flags or counts with --scan-current. Its pinned preview is invalidated by the reviewed parser and dynamic-cache work. Its 9,169 candidates were 512 EZMoney reclassifications, 3,831 Fubon reclassifications, and 4,826 Fubon footers; its 83 cache rows were 40 chip, 32 integrated, and 11 orchestrator rows. Those values are not acceptance criteria for a fresh dynamic scan.
| Evidence | SHA-256 |
|---|---|
| Holdings manifest | 23b9ca4ea54cdedee77920bf5ed05e1e67d7b64e7332c52bf88243e7ad4168fe |
| Holdings capture | 52041cecce0bdc31a001fe0f86b2c554fd00cd9bef588a7880f3e7a5a8f0bb23 |
| Holdings query | 39e3577202c004cffbacabacc501ed20b627f3a0e2388cbf30e58a5796be8b81 |
| Cache manifest | 748a48910e8975486fc1a11c065a50c29b0aef63158044827e1a22d7995c7bfe |
| Cache capture | de723c6e2dc56a66933adcddeef6029ed5b8729cfdc267ddc6494fc37718d013 |
| Cache query | 5c4f99c0f246acfce7578ea194b3e676642ea4927b03442629463043dda41d5c |
Maintenance order
- Stop and drain API, scheduler, and recap writers before enabling the corrected parser. This prevents a queued retry from replacing a known-good successful sync with newly parsed data. Do not use a blind historical fetch or retry to repair contaminated successful days.
- Enable the reviewed parser and detector while writers remain stopped.
- Run
--scan-current, retain its report and SHA-256, inspect actual targets, counts, cache plan, database identity, empty provider residual-after result, source-summary refusal result, and footer-onlyfreshness_marker. Create the sealed cache execution artifact below, prepare recovery/rollback backup, and obtain authorization for these exact artifacts before any write. - Apply under the five mutation locks. The command also holds
SHARElocks on daily and periodic chip source summaries; it does not modify them. The locked apply recheck repeats scope/residual safety checks; preview never provides that lock. - Perform mandatory holdings/cache read-back. Decide to retain or roll back before regeneration. Regeneration changes derived artifacts, so it removes the possibility of an exact artifact rollback; after that point a new backup and recovery boundary are required.
- Only after choosing retain, take the post-repair recovery backup, perform the separately bounded #1810 backfill if still eligible, then rebuild the sealed cache plan in an isolated maintenance process. Public traffic and scheduler writers remain paused. Resume readers only after mandatory read-back; model regeneration is never assumed successful.
--writers-quiesced is an attestation for apply/rollback; it does not stop or drain anything. The five mutation locks cover etf_funds, etf_holdings, etf_holdings_sync, recap_analysis_cache, and orchestrator_recap.
Evidence, backup, and read-back
Keep live before-images, cache text, SQL exports, backups, and operation reports outside Git in a protected directory (umask 077). Pass DATABASE_URL into the dedicated maintenance process privately through its environment or standard input; do not print it, save it in an artifact, or pass it as a CLI argument. Mount reviewed code read-only. Re-read and approve the actual host, database name, and PostgreSQL system identifier; do not infer them from a historical environment.
Before apply, and again after deciding to retain the repair but before backfill/regeneration, capture a custom-format backup through a read-only connection for exactly taiwan.etf_funds, taiwan.etf_holdings, taiwan.etf_holdings_sync, public.recap_analysis_cache, and public.orchestrator_recap. Record its SHA-256 and verify it with pg_restore --list. The preview's guarded rollback and this independent recovery backup serve different purposes.
Read back the exact preview after-state before leaving the rollback window: zero supported provider residuals, target holdings/sync fingerprints matching the preview, captured caches absent, unrelated controls unchanged, and global MAX(etf_holdings.updated_at) equal to repair_updated_at. Re-running the same approved apply artifact into a new operation-report path must return no_op; this runs the locked state checks without repeating mutations. For rollback, the full before-state and original timestamps must be restored. The marker rotates the holdings response-cache key, including footer-only repairs; a footer-only scope with no surviving holding in its affected snapshots refuses.
Sync fetched_count means the last provider response count. Decrement it only by the footers deleted from that snapshot, never replace it with persisted row count. The six previously audited fetched/persisted discrepancies remain outside this repair: 00983D/2026-06-03, 00920/2026-06-05, 00641R/2026-06-09, 00645/2026-06-10, 006205/2026-06-25, and 00639/2026-06-25.
Locks use a two-second timeout and statements a 120-second timeout. A refusal is not permission to bypass these guards. Exit 2 indicates refusal; exit 3 means commit/report outcome needs investigation. Preserve the original artifact and inspect the report/database state before retrying the same idempotent artifact; do not silently create a different repair scope.
Create and approve the dynamic preview
Use the exact target database identity values approved for this maintenance. The report path must be new and must not be overwritten.
python -m scripts.repair_etf_holding_classifications \
--scan-current \
--expected-database-host "$EXPECTED_DATABASE_HOST" \
--expected-database-name "$EXPECTED_DATABASE_NAME" \
--expected-database-system-identifier "$EXPECTED_DATABASE_SYSTEM_IDENTIFIER" \
--report-path /secure/maintenance/etf-repair-scan-current-v3.json
shasum -a 256 /secure/maintenance/etf-repair-scan-current-v3.json
The read-only preview uses a serializable transaction and returns scope_mode: "scan_current", version: 3, target before-images, action and target counts, active_etf_impact_start_date, cache_rebuild_plan, and a plan SHA-256. Its plan contains typed chip and daily-only integrated identities with market and exact period bounds, typed orchestrator identities, active-ETF dates, and legacy daily *_dates compatibility fields. It refuses rather than inventing a source-summary rebuild. The footer-only survivor is represented by freshness_marker, whose before/after updated_at is restored by rollback; there is no cache row for that marker.
Apply or rollback consumes the exact report and both displayed hashes:
python -m scripts.repair_etf_holding_classifications \
--apply --writers-quiesced \
--expected-database-host "$EXPECTED_DATABASE_HOST" \
--expected-database-name "$EXPECTED_DATABASE_NAME" \
--expected-database-system-identifier "$EXPECTED_DATABASE_SYSTEM_IDENTIFIER" \
--preview-report /secure/maintenance/etf-repair-scan-current-v3.json \
--expected-preview-sha256 "$PREVIEW_SHA256" \
--expected-plan-sha256 "$PLAN_SHA256" \
--report-path /secure/maintenance/etf-repair-apply-v3.json
Replace --apply with --rollback only for the approved rollback decision, using the same preview and plan hashes. A mismatch, source dependency, drift, or freshness marker change is a stop condition.
Seal the cache execution list before writes
This offline planning snippet reads the sealed report plus the latest eligible active-ETF cache key. It uses an explicit read-only transaction and the repository URL normalizer. Before approval, it verifies that every planned identity has its exact source summaries (chip for chip-only rebuilds; technical and chip for integrated/orchestrator): daily sources by date and periodic sources by market, type, start, and end. Retain its output with the preview; the write pass consumes this file and does not compute another date or plan.
python <<'PY'
import asyncio, hashlib, json, os
from datetime import date
from pathlib import Path
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from database.models import (
DailyChipMarketSummary, DailyTechnicalMarketSummary,
PeriodicChipMarketSummary, PeriodicTechnicalMarketSummary,
RecapAnalysisCache,
)
from scripts.etf_holdings import _normalize_async_db_url
from scripts.etf_providers.base import today_taiwan
from services.financial_recap_payload_service import get_financial_recap_payload
from services.active_etf_recap_service import active_etf_session_dates
report_path = Path('/secure/maintenance/etf-repair-scan-current-v3.json')
execution_path = Path('/secure/maintenance/etf-repair-cache-execution-v3.json')
report_bytes = report_path.read_bytes()
report = json.loads(report_bytes)
if report.get('version') != 3 or report.get('scope_mode') != 'scan_current':
raise RuntimeError('expected a sealed scan_current v3 preview')
plan = report.get('cache_rebuild_plan')
if not isinstance(plan, dict):
raise RuntimeError('preview has no cache rebuild plan')
chip, integrated = list(plan.get('chip', ())), list(plan.get('integrated', ()))
orchestrator, active_dates = list(plan.get('orchestrator', ())), set(plan.get('active_etf_dates', ()))
def recap_key(item):
return tuple(item[key] for key in ('market', 'period_type', 'period_start_date', 'period_end_date'))
def require_recap_items(items, *, recap_type, daily_only=False):
if len({recap_key(item) for item in items}) != len(items):
raise RuntimeError(f'duplicate {recap_type} cache identity')
for item in items:
if item['period_type'] not in {'daily', 'weekly', 'monthly'}:
raise RuntimeError(f'unsupported {recap_type} period type')
if daily_only and item['period_type'] != 'daily':
raise RuntimeError('integrated cache must be daily')
if date.fromisoformat(item['period_start_date']) > date.fromisoformat(item['period_end_date']):
raise RuntimeError(f'invalid {recap_type} period bounds')
require_recap_items(chip, recap_type='chip')
require_recap_items(integrated, recap_type='integrated', daily_only=True)
if len({tuple(item[key] for key in ('market', 'period_type', 'period_start_date', 'period_end_date')) for item in orchestrator}) != len(orchestrator):
raise RuntimeError('duplicate orchestrator identity')
for item in orchestrator:
if item['period_type'] not in {'daily', 'weekly', 'monthly'}:
raise RuntimeError('unsupported orchestrator period type')
if date.fromisoformat(item['period_start_date']) > date.fromisoformat(item['period_end_date']):
raise RuntimeError('invalid orchestrator period bounds')
latest = active_etf_session_dates(today_taiwan())[-1]
async def preflight_and_active_cache_exists():
engine = create_async_engine(_normalize_async_db_url(os.environ['DATABASE_URL']))
sessions = async_sessionmaker(engine, expire_on_commit=False)
try:
async with sessions() as session:
async with session.begin():
await session.execute(text('SET TRANSACTION READ ONLY'))
source_items = {
recap_key(item): item for item in chip + integrated + orchestrator
}.values()
needs_technical = {recap_key(item) for item in integrated + orchestrator}
for item in source_items:
start = date.fromisoformat(item['period_start_date'])
end = date.fromisoformat(item['period_end_date'])
if item['period_type'] == 'daily':
sources = [DailyChipMarketSummary]
if recap_key(item) in needs_technical:
sources.append(DailyTechnicalMarketSummary)
filters = lambda model: (
model.market == item['market'], model.trade_date == end,
)
else:
sources = [PeriodicChipMarketSummary]
if recap_key(item) in needs_technical:
sources.append(PeriodicTechnicalMarketSummary)
filters = lambda model: (
model.market == item['market'],
model.period_type == item['period_type'],
model.period_start_date == start, model.period_end_date == end,
)
for model in sources:
if not await session.scalar(select(model).where(*filters(model)).limit(1)):
raise RuntimeError(f'missing exact {model.__tablename__} source: {item}')
for item in integrated:
target = date.fromisoformat(item['period_end_date'])
financial = await get_financial_recap_payload(
session, target, item['market'], pre_resolved_source=target,
)
if financial.get('source_trade_date') != item['period_end_date'] or financial.get('is_fallback'):
raise RuntimeError(f'missing exact financial source: {item}')
return bool(await session.scalar(select(RecapAnalysisCache.id).where(
RecapAnalysisCache.recap_type == 'active_etf',
RecapAnalysisCache.market == 'taiwan',
RecapAnalysisCache.period_type == 'daily',
RecapAnalysisCache.period_start_date == latest,
RecapAnalysisCache.period_end_date == latest,
).limit(1)))
finally:
await engine.dispose()
latest_active_cache_exists = asyncio.run(preflight_and_active_cache_exists())
if plan.get('active_etf_latest_required') and not latest_active_cache_exists:
active_dates.add(latest.isoformat())
execution = {
'preview_report_sha256': hashlib.sha256(report_bytes).hexdigest(),
'chip': sorted(chip, key=recap_key), 'integrated': sorted(integrated, key=recap_key),
'orchestrator': sorted(orchestrator, key=lambda x: (x['period_end_date'], x['period_type'], x['market'])),
'active_etf_dates': sorted(active_dates),
}
execution['counts'] = {key: len(value) for key, value in execution.items() if isinstance(value, list)}
with execution_path.open('x', encoding='utf-8') as output:
json.dump(execution, output, ensure_ascii=False, indent=2, sort_keys=True)
output.write('\n')
print(hashlib.sha256(execution_path.read_bytes()).hexdigest())
PY
The printed SHA-256 is the execution artifact value to approve and set as EXPECTED_CACHE_EXECUTION_SHA256 for the write pass. The typed chip and integrated arrays are authoritative; legacy *_dates are daily compatibility fields only. The count is scheduled logical generation attempts, not an HTTP-request or cost ceiling: provider-internal retries can make more requests. Before cache short circuits, the single-domain chip path can schedule len(chip) + 4*len(integrated) + len(orchestrator) LLM operations. The observed 63 chip, 40 integrated, and 18 orchestrator identities therefore schedule 241 attempts. Active-ETF dates are rebuilt as mechanical templates and make no LLM call. Integrated rebuilds still use their technical/chip/financial dependencies; only the explicit orchestrator list generates orchestrator reports. Empty plan sections are valid; validate exact identities and dynamic counts in the sealed file rather than fixed historical totals.
Rebuild the sealed plan
Run only after retained-repair read-back and any approved #1810 backfill. Daily chip caches use the daily analysis runner with only the chip domain. Weekly/monthly chip caches use the periodic chip payload and MAS analyzer directly, validating exact period bounds before generation and the input hash after persistence. These chip-only paths create no technical, integrated, or orchestrator reports. Do not substitute the bundle endpoint, which also writes technical and orchestrator reports, or the periodic chip aggregation runner, which rewrites source summaries. Integrated cache rebuild remains daily-only. The explicit orchestrator runner checks its current calendar bounds against every sealed identity before any generation and returns scalar dimension statuses; active-ETF generation returns a recap mapping. Any exception, quota result without a persisted fallback, skipped input, failed status, date fallback, wrong bounds, or missing exact cache key stops the maintenance. Active-ETF dates are rebuilt with force=True and no LLM stage, so each persists the mechanical template with fallback_reason=llm_deferred; the scheduled evening and next-morning stages own LLM sends, but they only run for the current session, so a rebuilt historical date keeps the template unless an administrator sends POST /api/etf-holdings/recap/daily/analyze?date=<date>&force=true. Retain the reason in the evidence and verify its persisted exact date.
python <<'PY'
import asyncio, hashlib, json, os
from datetime import date
from pathlib import Path
from collections.abc import Mapping
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from database.models import OrchestratorRecap, RecapAnalysisCache
from core.mas.orchestrator import OrchestratorAgent
from core.utils.config_resolver import resolve_config
from services.mas_service import MASService
from services.periodic_recap_payload_builder import fetch_periodic_recap_payload
from scripts.etf_holdings import _normalize_async_db_url
from scripts.daily_recap_analysis import run_daily_recap_analysis
from scripts.run_periodic_recap import resolve_period_bounds, run_periodic_recap
from scripts.market_providers.snapshot_settings import cached_market_config
from services.active_etf_recap_service import generate_active_etf_recap
execution_path = Path('/secure/maintenance/etf-repair-cache-execution-v3.json')
execution_bytes = execution_path.read_bytes()
if hashlib.sha256(execution_bytes).hexdigest() != os.environ['EXPECTED_CACHE_EXECUTION_SHA256']:
raise RuntimeError('execution artifact SHA-256 does not match approval')
execution = json.loads(execution_bytes)
preview_bytes = Path('/secure/maintenance/etf-repair-scan-current-v3.json').read_bytes()
if execution.get('preview_report_sha256') != hashlib.sha256(preview_bytes).hexdigest():
raise RuntimeError('execution artifact is not bound to this preview')
if not isinstance(execution.get('preview_report_sha256'), str):
raise RuntimeError('unsealed execution plan')
def recap_key(item):
return tuple(item[key] for key in ('market', 'period_type', 'period_start_date', 'period_end_date'))
async def rebuild_periodic_chip(item, sessions):
async with sessions() as session:
await session.execute(text('SET TRANSACTION READ ONLY'))
payload = await fetch_periodic_recap_payload(
session, 'chip', item['period_type'],
date.fromisoformat(item['period_end_date']), item['market'],
)
if recap_key(payload) != recap_key(item) or payload.get('is_fallback'):
raise RuntimeError(f'chip payload resolved a different period: {recap_key(item)}')
service = MASService(
orchestrator=OrchestratorAgent.from_config(resolve_config(), session_factory=sessions),
async_session_factory=sessions,
)
await service.analyze_periodic_chip_recap(
period_type=item['period_type'], period_start_date=item['period_start_date'],
period_end_date=item['period_end_date'], recap_json=payload,
)
cached = await service.get_cached_recap_analysis(
recap_type='chip', trade_date=item['period_end_date'], market=item['market'],
input_hash=service._compute_input_hash(payload), period_type=item['period_type'],
period_start_date=item['period_start_date'], period_end_date=item['period_end_date'],
)
if cached is None:
raise RuntimeError(f'chip cache persistence verification failed: {recap_key(item)}')
def require_daily(stats, names):
bad = {name: str(stats.get(name))[:512] for name in names
if not isinstance(stats.get(name), Mapping) or stats[name].get('status') != 'success'}
if bad:
raise RuntimeError(f'daily recap failed: {bad}')
async def exact_cache_exists(session, recap_type, item):
return bool(await session.scalar(select(RecapAnalysisCache.id).where(
RecapAnalysisCache.recap_type == recap_type,
RecapAnalysisCache.market == item['market'],
RecapAnalysisCache.period_type == item['period_type'],
RecapAnalysisCache.period_start_date == date.fromisoformat(item['period_start_date']),
RecapAnalysisCache.period_end_date == date.fromisoformat(item['period_end_date']),
).limit(1)))
async def main():
for item in execution['orchestrator']:
expected = tuple(date.fromisoformat(item[key]) for key in ('period_start_date', 'period_end_date'))
calendar = cached_market_config().get_exchange_calendar_code(item['market'])
if resolve_period_bounds(item['period_type'], expected[1], calendar) != expected:
raise RuntimeError(f'orchestrator runner would change sealed bounds: {item}')
db_url = _normalize_async_db_url(os.environ['DATABASE_URL'])
engine = create_async_engine(db_url)
sessions = async_sessionmaker(engine, expire_on_commit=False)
try:
for item in execution['chip']:
if item['period_type'] == 'daily':
stats = await run_daily_recap_analysis(
db_url=db_url, target_date=date.fromisoformat(item['period_end_date']),
market=item['market'], domains=('chip',),
)
require_daily(stats, ('chip',))
else:
await rebuild_periodic_chip(item, sessions)
for item in execution['integrated']:
target = date.fromisoformat(item['period_end_date'])
stats = await run_daily_recap_analysis(
db_url=db_url, target_date=target, market=item['market'],
domains=('technical', 'chip', 'financial', 'integrated'),
)
require_daily(stats, ('technical', 'chip', 'financial', 'integrated'))
for item in execution['orchestrator']:
statuses = await run_periodic_recap(
db_url, item['period_type'], date.fromisoformat(item['period_end_date']),
dimensions=('orchestrator',), market=item['market'], skip_llm=False,
)
if statuses.get('orchestrator') != 'success':
raise RuntimeError(f'orchestrator recap failed: {statuses}')
async with sessions() as session:
async with session.begin():
await session.execute(text('SET TRANSACTION READ ONLY'))
for item in execution['chip']:
if not await exact_cache_exists(session, 'chip', item):
raise RuntimeError(f"missing exact chip cache key: {recap_key(item)}")
for item in execution['integrated']:
if not await exact_cache_exists(session, 'integrated', item):
raise RuntimeError(f"missing exact integrated cache key: {recap_key(item)}")
for item in execution['orchestrator']:
row = await session.scalar(select(OrchestratorRecap.period_type).where(
OrchestratorRecap.period_type == item['period_type'],
OrchestratorRecap.period_end_date == date.fromisoformat(item['period_end_date']),
OrchestratorRecap.market == item['market'],
OrchestratorRecap.period_start_date == date.fromisoformat(item['period_start_date']),
).limit(1))
if row is None:
raise RuntimeError(f"missing exact orchestrator key: {item}")
for item in execution['active_etf_dates']:
target = date.fromisoformat(item)
async with sessions() as session:
result = await generate_active_etf_recap(session, target, force=True)
analysis = result.get('analysis')
mode = getattr(analysis, 'generation_mode', None)
reason = getattr(analysis, 'fallback_reason', None)
if result.get('trade_date') != target or (mode, reason) != ('template', 'llm_deferred'):
raise RuntimeError(f'active ETF recap did not persist a valid result for {target}')
print({'active_etf_date': target.isoformat(), 'generation_mode': mode,
'fallback_reason': reason})
async with sessions() as session:
async with session.begin():
await session.execute(text('SET TRANSACTION READ ONLY'))
for item in execution['active_etf_dates']:
target = date.fromisoformat(item)
active_item = {
'market': 'taiwan', 'period_type': 'daily',
'period_start_date': item, 'period_end_date': item,
}
if not await exact_cache_exists(session, 'active_etf', active_item):
raise RuntimeError(f'missing exact active ETF cache key: {target}')
finally:
await engine.dispose()
asyncio.run(main())
PY
Before this pass, verify the sealed prerequisite check passed for every planned identity. The preview refuses source-summary dependencies, so this pass must not be used to regenerate a source summary.
Bounded #1810 zero-fetch recovery
After repair is retained and before report regeneration, separately verify that each row below is still EZMoney, failed or source_date_mismatch, fetched_count = 0, and has no persisted holdings. Only then use the existing bounded capture path for these 12 identities; read back sync state and holdings afterward.
The dates are issuer holdings dates (TranDate) after the #1947 source-date repair. Before #1947 the adapter labelled overseas funds one session late, so this table originally listed 00757 2026-06-17 / 2026-09-16 and 009811 2026-09-17 / 2026-09-18 (the old labels; the rollover fixtures in tests/fixtures/ezmoney_rollover/ keep those asof names). Domestic 00939 dates were already correct. Until the #1947 repair is applied, the database still carries the old labels.
| ETF | Dates |
|---|---|
00757.TW |
2026-06-16, 2026-09-15 |
00939.TW |
2026-04-10, 2026-05-18, 2026-06-11, 2026-07-08, 2026-08-12, 2026-08-13, 2026-08-14, 2026-09-14 |
009811.TW |
2026-09-16, 2026-09-17 |
python -m scripts.etf_holdings fetch --provider ezmoney --etf 00757.TW --date 2026-06-16
For each capture, require the requested ETF-day sync to become success with fetched_count > 0 and holdings on that same source date. A fallback date or remaining failed/empty day does not complete recovery; stop before generating reports from it. Repeat only for independently verified identities above. Runtime reconciliation protects each capture day; it is deployed writer behavior, not a migration to run. Do not extend this to successful contaminated days or use bulk historical retries.