wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

fix(agent): anchor closure on terminal runs
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 1s
Details
CD Pipeline / cancel-stale-cd (push) Has been skipped
Details
CD Pipeline / tests (push) Successful in 1m16s
Details
CD Pipeline / build-and-deploy (push) Successful in 4m48s
Details
CD Pipeline / post-deploy-checks (push) Successful in 4m48s
Details

Browse Source
This commit is contained in:
ogt
2026-07-10 23:19:51 +08:00
parent fbc647c222
commit 4476389d5a
2 changed files with 94 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
apps/api/tests/test_ai_agent_autonomous_runtime_control.py
View File

@@ -2448,11 +2448,53 @@ def test_runtime_telegram_receipt_queries_include_alert_notifications() -> None:
def test_runtime_operation_latest_queries_prioritize_latest_apply_chain() -> None:
for sql in (_RUNTIME_OPERATION_LATEST_SQL, _RUNTIME_OPERATION_LATEST_DIRECT_SQL):
assert "WITH latest_apply_chain AS" in sql
assert "AND status IN ('success', 'failed')" in sql
assert "operation_row.op_id::text = latest_apply_chain.apply_op_id" in sql
assert "operation_row.op_id::text = latest_apply_chain.check_mode_op_id" in sql
assert "operation_row.op_id::text = latest_apply_chain.candidate_op_id" in sql
def test_runtime_execution_loop_uses_terminal_apply_without_hiding_inflight_apply():
ledger = runtime_control_module._autonomous_execution_loop_ledger(
project_id="awoooi",
operation_latest_rows=[
{
"op_id": "pending-apply",
"operation_type": "ansible_apply_executed",
"status": "pending",
"automation_run_id": "pending-run",
"incident_id": "INC-PENDING",
"created_at": "2026-07-10T15:00:00Z",
},
{
"op_id": "terminal-apply",
"operation_type": "ansible_apply_executed",
"status": "failed",
"automation_run_id": "terminal-run",
"incident_id": "INC-TERMINAL",
"created_at": "2026-07-10T14:59:00Z",
},
],
verifier_latest_rows=[],
km_latest_rows=[],
telegram_latest_rows=[],
auto_repair_latest_rows=[],
latest_flow_closure={},
latest_failure_classification={},
controlled_retry_package={},
)
assert ledger["operation_id"] == "terminal-apply"
assert ledger["completion_anchor"] == "latest_terminal_apply"
assert ledger["latest_inflight_apply"] == {
"op_id": "pending-apply",
"status": "pending",
"automation_run_id": "pending-run",
"incident_id": "INC-PENDING",
"created_at": "2026-07-10T15:00:00Z",
}
def test_runtime_execution_loop_ledger_does_not_mix_unrelated_check_mode_rows():
apply_op_id = "db3f12ce-08fc-4289-8c93-338305d5850c"
readback = build_runtime_receipt_readback_from_rows(