fix(agents): close trace ledger repair receipt gap
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 1m5s
CD Pipeline / build-and-deploy (push) Successful in 4m26s
CD Pipeline / post-deploy-checks (push) Has been cancelled

This commit is contained in:
ogt
2026-07-09 22:48:17 +08:00
parent abd7ec0815
commit 0fe3b4b593
2 changed files with 200 additions and 10 deletions

View File

@@ -1426,6 +1426,55 @@ def test_runtime_receipt_recovery_flags_zero_live_log_events():
assert readback["work_item_progress"]["rollups"]["not_started_count"] == 10
def test_trace_ledger_uses_controlled_apply_chain_for_auto_repair_receipt_gap():
readback = build_runtime_receipt_readback_from_rows(
project_id="awoooi",
db_read_status="ok",
operation_count_rows=[
{"operation_type": "ansible_candidate_matched", "total": 1, "recent": 1},
{"operation_type": "ansible_check_mode_executed", "total": 1, "recent": 1},
{"operation_type": "ansible_apply_executed", "total": 1, "recent": 1},
],
verifier_count_rows=[
{"verification_result": "success", "total": 1, "recent": 1},
],
km_count_rows=[
{"status": "review", "total": 1, "recent": 1},
],
telegram_count_rows=[
{"send_status": "sent", "total": 1, "recent": 1},
],
)
trace = readback["trace_ledger"]
auto_repair_stage = next(
stage
for stage in trace["stages"]
if stage["stage_id"] == "auto_repair_execution_receipt"
)
assert auto_repair_stage["recorded"] is True
assert auto_repair_stage["record_quality"] == (
"controlled_apply_receipt_chain_fallback"
)
assert "auto_repair_execution_receipt" not in trace["missing_required_stage_ids"]
progress_items = {
item["work_item_id"]: item
for item in readback["work_item_progress"]["ordered_items"]
}
assert progress_items["P0-B-trace-ledger"]["status"] == "completed"
auto_repair_source = next(
item
for item in readback["log_integration_taxonomy"]["source_families"]
if item["source_family_id"] == "auto_repair_receipts"
)
assert auto_repair_source["total"] == 1
assert auto_repair_source["record_quality"] == (
"controlled_apply_receipt_chain_fallback"
)
def test_runtime_receipt_recovery_keeps_partial_live_log_events_visible():
readback = build_runtime_receipt_readback_from_rows(
project_id="awoooi",