From b36f4adc58dc15022bef1c40bbb48826fe5e25a3 Mon Sep 17 00:00:00 2001 From: host Date: Sat, 31 Oct 2026 08:05:24 +0000 Subject: [PATCH] audit: wire loop driver to server endpoints and release gate (track 2) Server-side execution for the track-1 pure driver (#260), no driver semantics changed: audit verdict / fix completion / correction / rebuild confirmation endpoints persist LoopState in audit_loop_state, write every step through the #241/#247 ledger path (ledger first, state second, fail-closed), derive fix/rebuild subtasks, and block merge/release via the existing #247 chain reader. POST /tasks/{id}/release follows the #259 contract (global-admin write, fresh hard-gate evidence, merged status, multi-source 409 with no audit write on block); deploy.py stays outside task gates. --- cloud/audit_loop_wiring.py | 954 ++++++++++++++++++++++++++ cloud/server.py | 524 ++++++++++++++ cloud/tests/test_audit_loop_wiring.py | 439 ++++++++++++ cloud/tests/test_server_audit_loop.py | 457 ++++++++++++ 4 files changed, 2374 insertions(+) create mode 100644 cloud/audit_loop_wiring.py create mode 100644 cloud/tests/test_audit_loop_wiring.py create mode 100644 cloud/tests/test_server_audit_loop.py diff --git a/cloud/audit_loop_wiring.py b/cloud/audit_loop_wiring.py new file mode 100644 index 0000000..08924f5 --- /dev/null +++ b/cloud/audit_loop_wiring.py @@ -0,0 +1,954 @@ +"""Server-side wiring for the project-level audit loop (track 2). + +This module is the ONLY place that connects the track-1 pure driver +(``cloud/audit_loop_driver.py``, PR #260) to durable server state. +It changes no driver semantics and it does not re-implement the +driver, the #247 gate reader, or the #241 hash chain: + +- Decisions come from the driver, verbatim: + ``decide_audit`` / ``after_fix`` / ``apply_correction`` / + ``confirm_rebuild``. This module never recomputes a pushback + round, a failure count, or a rebuild budget; every number it + persists comes from ``Action.next_state``. +- Merge blocking stays exactly the #247 reader + (``audit_wiring.pushback_block_for_task``) ANDed with the five + hard-gate checks in ``server._merge_task``. This module does NOT + add a second gate reader for merge: the loop-state table below + is driver state, not a gate. If chain and loop state ever + diverged (only possible if a caller wrote state without going + through this module), merge follows the chain, because the + chain is the auditable record. The normal flow cannot diverge: + ``process_*`` writes the ledger FIRST and only then saves the + state, so a ledger failure leaves the old state untouched. +- Ledger writes go through the #241/#247 layer only + (``audit_wiring.record_*`` over a ``LedgerStore``-shaped store). + No second write path is invented here. +- Quota and cost are caller-injected, as the driver requires. + The quota adapter below follows the existing server quota + model (``server._quota_status``: a USD-micros budget per user + per month, ``used >= quota`` blocks, ``None`` means uncapped); + it does not invent a second quota table. #212's design口径 is + the server's existing ``user_quotas`` table, reused as-is. +- ``evaluate_feedback`` / the #238 metrics loop is untouched: + it stays ``auto_apply: False``. Nothing in this module feeds + loop outcomes back into the standard documents, and no + threshold in this module loosens or tightens a standard. +- ``deploy.py`` is NOT in any task gate here. It deploys the + platform itself, has no task id, and does not go through + server HTTP auth; binding it to a task chain would lock + deployment exactly when a fix or rollback needs deploying. + The release endpoint below is a gate + authorization for a + release, not a deployment runner. + +Module mapping (task -> loop key), stated once and tested: + +- The loop key is ``(module_id, contract_version)``, exactly as + the driver counts it. +- ``module_id``: a task's explicit ``module_id`` field when it + is a non-empty string, else its ``module`` field, else the + task id itself. Trade-off, stated not hidden: an explicit + module field lets several tasks share one loop (one module, + one counter), which is what a module-level loop means; the + task-id fallback isolates a task that names no module, so two + unrelated tasks can never silently share a counter. Callers + must not mix the two for the same module: once a module is + named, every verdict/fix for it must carry that name. +- ``contract_version``: the verdict's (or re-audit verdict's) + ``contract_version`` when given, else the task's + ``contract_version`` field. When neither exists the call + fails closed (ValueError) instead of inventing a version: + an invented version would merge counters across versions. +- Loop state is persisted in table ``audit_loop_state`` keyed + by ``(module_id, contract_version)`` with the owning + ``task_id`` recorded for traceability. It is NOT memory-only: + a server restart reloads the same counters, rounds, rebuild + budget, targeted budget, and pending confirmation from this + table. The table is created idempotently by + ``ensure_loop_state_table``. + +Automatic chain (what "automatic" means here, precisely): + +- "Task completed -> independent audit -> fail auto-derives a + fix subtask and blocks merge/release (original task stays + ``completed``, it is NOT rolled back to a re-run)" is + ``process_audit_verdict``: the driver decides, the ledger is + written, the state is saved, and a fix/rebuild subtask + descriptor is returned. The server handler turns the + descriptor into a real queued task row; this module itself + creates no task rows (it has no HTTP/task-row layer). +- "Fix completed -> automatic full re-audit" is + ``process_fix_completion``: the fix-completion event itself + advances the loop through the driver's ``after_fix``, which + requires a FULL re-audit verdict. The automated checks run + here (hard gate five + directory/contract + demo/runnable, + regression, security, signer-separation, QA-link re-runs) + are EVIDENCE for that re-audit, never the verdict: if any + automated check fails, a claimed pass is forced to fail + before it reaches the driver; if they all pass, the loop + still waits for the auditor-submitted verdict, and a + missing verdict or ``diff_only`` is refused with ValueError + by the driver. Automated-checks-passing is NEVER reported + as audit-passing. +- Subjective audit conclusions still come from the auditor: + this module accepts verdicts, it does not invent them. + No verdict is fabricated anywhere in this file. + +Release gate (``evaluate_release_gate``) follows the #259 +contract: re-read the SAME #247 reader at release time, re-run +the hard gate on caller-supplied FRESH evidence (a merge-time +200 is never an input), and require task status ``merged``. +All three are ANDed. ``blocked_sources`` lists every blocking +source when several block at once, except a chain-read failure, +which short-circuits per #259 (no other reading is trustworthy +then). A 409 from the endpoint writes NO audit record, because +a check writing to the chain would move the chain head and +invalidate an anchor taken before the check. + +Confirmation rule (acceptance-triggered rebuild): + +- ``process_rebuild_confirmation`` never auto-confirms. It + requires an explicit ``confirmed_by`` that is a real user id: + the reserved markers ``"system"`` and ``"coordinator"`` are + refused with ValueError, and the server endpoint additionally + requires the request body to carry ``user_confirmed: true`` + AND ``confirmed_by`` equal to the authenticated caller. + No code path in this module or in the server handlers calls + confirmation on behalf of a user. A rejection escalates to + the coordinator and consumes no rebuild budget (driver). + Pending and confirmed are two chain rows, never an edit. + Confirmation is NOT G6 acceptance: the user still holds + final acceptance afterwards, and there is no + ``tick_acceptance`` anywhere in this wiring. +""" + +import json +import os +import time + +from cloud import audit_loop_driver as _driver +from cloud import audit_wiring as _wiring +from cloud.merge_gate_hard import evaluate_hard_merge_gate as _hard_gate + +# ------------------------------------------------------------------ +# Mapping: task / verdict -> loop key +# ------------------------------------------------------------------ + +_RESERVED_CONFIRMERS = ("system", "coordinator", "") + +# Automated re-audit evidence: extra checks beyond the hard five. +# Each must be exactly True; missing or non-True is a failure, +# never a skip. These names are the driver's FULL_REAUDIT_CRITERIA +# minus the five hard-gate items, plus the QA re-run check. +_AUTOMATED_EXTRA_CHECKS = ( + "security_unconditional", + "regression_exact_artifact", + "demo_runnable", + "known_deviation_due_todo", + "audit_signer_separation", + "qa_failed_links_rerun", +) + + +def resolve_module_id(task, explicit=None): + """Task -> module_id. Explicit argument wins, then the task's + own module fields, then the task id (isolated fallback). + See the module docstring for the trade-off.""" + if isinstance(explicit, str) and explicit.strip(): + return explicit.strip() + if isinstance(task, dict): + for key in ("module_id", "module"): + value = task.get(key) + if isinstance(value, str) and value.strip(): + return value.strip() + task_id = task.get("id") + if isinstance(task_id, str) and task_id.strip(): + return task_id.strip() + raise ValueError("cannot resolve module_id: no module field " + "and no task id") + + +def resolve_contract_version(task, verdict=None): + """Task/verdict -> contract_version. Verdict wins when given + (a version change is how the driver resets counters), else + the task's field. Neither -> ValueError, never an invented + version.""" + if isinstance(verdict, dict): + version = verdict.get("contract_version") + if isinstance(version, str) and version.strip(): + return version.strip() + if isinstance(task, dict): + version = task.get("contract_version") + if isinstance(version, str) and version.strip(): + return version.strip() + raise ValueError("cannot resolve contract_version: neither the " + "verdict nor the task carries one") + + +def task_project_id(task): + """Project id for ledger rows (empty string when unknown; + the ledger layer, not this module, decides whether that is + acceptable for a given row type).""" + if isinstance(task, dict): + project_id = task.get("project_id") + if isinstance(project_id, str): + return project_id + return "" + + +def task_id_of(task): + if isinstance(task, dict): + return task.get("id") + return None + + +# ------------------------------------------------------------------ +# Loop-state persistence (table audit_loop_state) +# ------------------------------------------------------------------ + +def ensure_loop_state_table(conn): + conn.execute( + "CREATE TABLE IF NOT EXISTS audit_loop_state(" + "module_id TEXT NOT NULL, " + "contract_version TEXT NOT NULL, " + "task_id TEXT, " + "state_json TEXT NOT NULL, " + "updated_at REAL, " + "PRIMARY KEY(module_id, contract_version))") + try: + conn.commit() + except Exception: + pass + + +def state_to_dict(state): + return { + "module_id": state.module_id, + "contract_version": state.contract_version, + "pushback_round": state.pushback_round, + "failure_count": state.failure_count, + "audit_fail_count": state.audit_fail_count, + "acceptance_fail_count": state.acceptance_fail_count, + "rebuild_no": state.rebuild_no, + "targeted_used": state.targeted_used, + "status": state.status, + "pending_confirmation": state.pending_confirmation, + "retained_tests": list(state.retained_tests or []), + "retained_passed_items": list(state.retained_passed_items or []), + "previous_artifacts": list(state.previous_artifacts or []), + "history": list(state.history or []), + } + + +def state_from_dict(data): + if not isinstance(data, dict): + raise ValueError("loop state must be a dict") + return _driver.LoopState( + module_id=data["module_id"], + contract_version=data["contract_version"], + pushback_round=int(data.get("pushback_round") or 0), + failure_count=int(data.get("failure_count") or 0), + audit_fail_count=int(data.get("audit_fail_count") or 0), + acceptance_fail_count=int(data.get("acceptance_fail_count") or 0), + rebuild_no=int(data.get("rebuild_no") or 0), + targeted_used=bool(data.get("targeted_used")), + status=data.get("status") or "auditing", + pending_confirmation=bool(data.get("pending_confirmation")), + retained_tests=list(data.get("retained_tests") or []), + retained_passed_items=list( + data.get("retained_passed_items") or []), + previous_artifacts=list(data.get("previous_artifacts") or []), + history=list(data.get("history") or []), + ) + + +def load_loop_state(conn, module_id, contract_version): + """Load persisted state for (module_id, contract_version), + or None when the loop has never started. Numbers are never + taken from a caller: they come from this row, or from a + fresh driver state.""" + ensure_loop_state_table(conn) + row = conn.execute( + "SELECT state_json FROM audit_loop_state" + " WHERE module_id = ? AND contract_version = ?", + (module_id, contract_version)).fetchone() + if row is None: + return None + raw = row[0] + return state_from_dict(json.loads(raw)) + + +def save_loop_state(conn, task_id, state): + """Upsert by (module_id, contract_version) using a portable + select-then-write (no dialect-specific upsert syntax, so the + same code works on the SQLite and Postgres adapters).""" + ensure_loop_state_table(conn) + payload = json.dumps(state_to_dict(state), sort_keys=True, + ensure_ascii=False) + existing = conn.execute( + "SELECT module_id FROM audit_loop_state" + " WHERE module_id = ? AND contract_version = ?", + (state.module_id, state.contract_version)).fetchone() + if existing is None: + conn.execute( + "INSERT INTO audit_loop_state(module_id," + " contract_version, task_id, state_json, updated_at)" + " VALUES(?, ?, ?, ?, ?)", + (state.module_id, state.contract_version, task_id, + payload, time.time())) + else: + conn.execute( + "UPDATE audit_loop_state SET task_id = ?," + " state_json = ?, updated_at = ?" + " WHERE module_id = ? AND contract_version = ?", + (task_id, payload, time.time(), + state.module_id, state.contract_version)) + conn.commit() + + +# ------------------------------------------------------------------ +# Quota / cost injection (server quota model, no new table) +# ------------------------------------------------------------------ + +def build_quota_check(quota_usd_micros, used_usd_micros): + """Server quota -> driver quota_check callable. + + ``quota_usd_micros`` is None for uncapped users (always + allowed). Otherwise an estimate (USD float, as the driver + carries it) blocks when used + estimate would exceed the + budget; with no estimate the existing server rule applies + (used >= quota blocks). Any lookup problem must surface as + a raising callable: the driver fail-closes a raising check + to quota_blocked, it never treats it as allowed. + """ + def _check(estimate): + if quota_usd_micros is None: + return True + if estimate is None: + return used_usd_micros < quota_usd_micros + try: + estimate_micros = int(float(estimate) * 1000000) + except (TypeError, ValueError): + return False + return used_usd_micros + estimate_micros <= quota_usd_micros + return _check + + +def cost_threshold_from_env(env=None): + """User-set rebuild cost-escalation threshold in USD, from + CLOUD_AUDIT_LOOP_COST_THRESHOLD_USD. Unset / unparsable + means None: no cost escalation, only quota checks. The + threshold is never hard-coded here; an operator sets it.""" + raw = (env if env is not None else os.environ).get( + "CLOUD_AUDIT_LOOP_COST_THRESHOLD_USD", "") + raw = (raw or "").strip() + if not raw: + return None + try: + value = float(raw) + except ValueError: + return None + return value if value >= 0 else None + + +# ------------------------------------------------------------------ +# Automated re-audit evidence (never a verdict by itself) +# ------------------------------------------------------------------ + +def evaluate_automated_reaudit(hard_gate_kwargs, extra_checks): + """Run the automatable part of a full re-audit. + + ``hard_gate_kwargs`` are the kwargs for + ``evaluate_hard_merge_gate`` (fresh evidence, supplied by + the caller at fix-completion time). ``extra_checks`` maps + each _AUTOMATED_EXTRA_CHECKS name to a boolean supplied by + the caller (demo run result, regression artifact comparison, + and so on). Missing evidence fails, it never passes by + default. Returns {automated_checks_pass, failed_checks, + hard_gate}. This result is evidence only: see + process_fix_completion for how a failed automated check + forces a claimed pass to fail, and how passing automated + checks still require an auditor verdict. + """ + failed = [] + gate = None + if not isinstance(hard_gate_kwargs, dict): + failed.append("hard_gate_evidence_missing") + else: + # Vocabulary translation at the boundary: merge bodies + # carry "branch_protection", the hard-gate function + # takes "protection". Anything else is passed through + # untouched (unknown kwargs still fail loudly, which + # is the honest outcome for malformed evidence). + kwargs = dict(hard_gate_kwargs) + if "branch_protection" in kwargs: + kwargs.setdefault("protection", kwargs.pop("branch_protection")) + gate = _hard_gate(**kwargs) + if not gate.get("allowed"): + failed.extend("hard_gate:%s" % code + for code in (gate.get("blocks") or [])) + checks = extra_checks if isinstance(extra_checks, dict) else {} + for name in _AUTOMATED_EXTRA_CHECKS: + if checks.get(name) is not True: + failed.append(name) + return {"automated_checks_pass": not failed, + "failed_checks": failed, "hard_gate": gate} + + +# ------------------------------------------------------------------ +# Ledger sink: driver events -> #241/#247 write path, fail-closed +# ------------------------------------------------------------------ + +def _prevalidate_ledger_events(action, old_impl_ref): + """Refuse, BEFORE writing anything, the two driver-event + shapes the #241 layer cannot store: a rebuild event with + no old-implementation reference, and a meta-review event + with no chain entry id. Writing the verdict row first and + then failing on the rebuild row would leave a half-written + round on the chain; fail-closed means refusing the whole + step up front. The old_impl_ref itself is caller-supplied + (a real commit/tag when known, else a task-scoped archive + reference); this module never invents a commit hash.""" + for event in (action.ledger_events or []): + payload = event.get("payload") or {} + if event.get("event_type") == "audit_rebuild": + ref = payload.get("old_impl_ref") or old_impl_ref + if not isinstance(ref, dict) or not any( + isinstance(ref.get(key), str) and ref[key] + for key in ("commit", "tag", "archive_ref")): + raise ValueError( + "rebuild ledger event needs old_impl_ref " + "(commit/tag/archive_ref) from the caller; " + "refusing to write a partial round") + if event.get("event_type") == "audit_meta_review": + entry_id = payload.get("original_entry_id") + if (isinstance(entry_id, bool) + or not isinstance(entry_id, int) + or entry_id < 1): + raise ValueError( + "meta-review ledger event needs " + "original_entry_id (chain id >= 1)") + + +def write_loop_ledger_events(ledger_store, task_id, project_id, + action, old_impl_ref=None, + user_id="system"): + """Write one Action's ledger events + compat hint through + the #247 record_* functions, in driver order. Any failure + propagates (fail-closed): the caller must NOT save the new + loop state after this raises. Unknown event types raise + instead of being silently skipped.""" + _prevalidate_ledger_events(action, old_impl_ref) + written = [] + for event in (action.ledger_events or []): + event_type = event.get("event_type") + payload = event.get("payload") or {} + # The driver's meta-review event carries no module key + # (its field table does not require one); the loop key + # for the chain row then comes from next_state, which + # is the same loop the correction acted on. Other + # events carry their own key and are left untouched. + module_id = payload.get("module_id") \ + or action.next_state.module_id + contract_version = payload.get("contract_version") \ + or action.next_state.contract_version + if event_type == "audit_verdict": + written.append(_wiring.record_audit_verdict( + ledger_store, task_id, project_id, module_id, + contract_version, payload.get("verdict"), + payload.get("source"), + failures=payload.get("failures"), + user_id=user_id, extra=payload.get("extra"))) + elif event_type == "audit_pushback_round": + trigger = _wiring._trigger_check_names( + payload.get("trigger_items")) + if not trigger: + # A fail verdict with no itemised failures still + # happened; the generic code names that fact + # instead of inventing a specific check. + trigger = ["audit_verdict_not_pass"] + if payload.get("round_status") == "re_reviewed": + written.append(_wiring.record_reaudit( + ledger_store, task_id, project_id, module_id, + contract_version, payload.get("round_no"), + payload.get("re_review_result"), + package=payload.get("pushback_package"), + failure_source=payload.get("failure_source"), + failure_counts=payload.get("failure_counts"), + user_id=user_id)) + else: + written.append(_wiring.record_pushback( + ledger_store, task_id, project_id, module_id, + contract_version, payload.get("round_no"), + trigger, payload.get("pushback_package"), + failure_source=payload.get("failure_source"), + failure_counts=payload.get("failure_counts"), + user_id=user_id)) + elif event_type == "audit_rebuild": + root_cause = payload.get("root_cause") + # Driver emits "implementation"; the #247 layer maps + # contract/design to contract_design, so a + # contract_design payload is handed over as + # "contract" (the layer's own vocabulary). + if root_cause == "contract_design": + root_cause = "contract" + written.append(_wiring.record_rebuild( + ledger_store, task_id, project_id, module_id, + contract_version, payload.get("rebuild_no"), + payload.get("old_impl_ref") or old_impl_ref, + root_cause=root_cause or "implementation", + status=payload.get("status") or "decided", + retained_assets=payload.get("retained_assets"), + trigger_source=payload.get("trigger_source"), + user_id=user_id)) + elif event_type == "audit_third_failure_assessment": + # Translation at the boundary (driver field vs #247 + # API): the driver emits the #241 determination + # ("continuation_round" / "early_rebuild"), while + # #247's record_* takes the triage verdict + # ("targeted_fix" / "early_rebuild") and maps it to + # the same determination on the chain. Translating + # here is this layer's job; the driver is untouched. + triage_verdict = { + "continuation_round": "targeted_fix", + "early_rebuild": "early_rebuild", + }.get(payload.get("determination")) + if triage_verdict is None: + raise ValueError( + "unknown third-failure determination: %r" + % (payload.get("determination"),)) + written.append(_wiring.record_third_failure_assessment( + ledger_store, task_id, project_id, module_id, + contract_version, triage_verdict, + payload.get("basis"), + failure_total=payload.get("failure_total") or 3, + max_extra_rounds=payload.get("max_extra_rounds"), + user_id=user_id)) + elif event_type == "audit_test_change": + changes = payload.get("test_changes") or {} + written.append(_wiring.record_test_change( + ledger_store, task_id, project_id, module_id, + contract_version, changes, + payload.get("auditor") or user_id, + bool(payload.get("contract_refrozen")), + refreeze_ref=payload.get("refreeze_ref"), + reason=payload.get("reason") or "", + user_id=user_id)) + elif event_type == "audit_meta_review": + written.append(_wiring.record_meta_review( + ledger_store, task_id, project_id, module_id, + contract_version, payload.get("original_entry_id"), + payload.get("original_verdict"), + payload.get("corrected_verdict"), + bool(payload.get("misjudgment_found")), + payload.get("reviewer"), + findings=payload.get("findings"), + triggered_process=payload.get("triggered_process"), + user_id=user_id)) + elif event_type == "audit_rebuild_notice": + written.append(_wiring.record_rebuild_notice( + ledger_store, task_id, project_id, module_id, + contract_version, payload.get("rebuild_no"), + payload.get("notification_content"), + payload.get("notified_party"), + payload.get("confirmation_status"), + confirmed_by=payload.get("confirmed_by"), + user_id=user_id)) + else: + raise ValueError( + "unknown loop ledger event type: %r" % (event_type,)) + hint = getattr(action, "compat_hint", None) + if hint == "audit_wiring_event:blocked_quota": + written.append(_wiring.record_escalation( + ledger_store, task_id, project_id, + action.next_state.module_id, + action.next_state.contract_version, + "blocked_quota", reason=action.reason, + user_id=user_id)) + elif hint == "audit_wiring_event:escalate": + written.append(_wiring.record_escalation( + ledger_store, task_id, project_id, + action.next_state.module_id, + action.next_state.contract_version, + "escalate", target=action.escalation_target, + reason=action.reason, user_id=user_id)) + elif hint == "audit_wiring_event:escalate_refreeze": + written.append(_wiring.record_escalation( + ledger_store, task_id, project_id, + action.next_state.module_id, + action.next_state.contract_version, + "escalate_refreeze", + target=action.escalation_target, + reason=action.reason, user_id=user_id)) + return written + + +def derive_subtask(action, parent_task): + """Driver Action -> fix/rebuild subtask descriptor, or None. + + This is a descriptor only: the server handler turns it into + a queued task row (same project, created_by = the parent + task's creator for cost attribution). No subtask is derived + for pass / escalate / refreeze / pending confirmation / + quota_blocked: a pending acceptance rebuild must not create + a rebuild task before the user confirms, and an escalation + goes to a human, not to an auto-created task. + A correction to pass is the one exception inside kind + "pushback": it schedules a re-audit only, so its + descriptor is kind "reaudit", never a fix task. + """ + if getattr(action, "correction_action", None) == "schedule_reaudit": + return { + "kind": "reaudit", + "parent_task_id": task_id_of(parent_task), + "module_id": action.next_state.module_id, + "contract_version": action.next_state.contract_version, + "scope": "reaudit_only", + "pushback_package": None, + "requires_reaudit": True, + } + if action.kind == "pushback": + return { + "kind": "fix", + "parent_task_id": task_id_of(parent_task), + "module_id": action.next_state.module_id, + "contract_version": action.next_state.contract_version, + "scope": action.scope, + "targeted_items": action.targeted_items, + "pushback_package": action.pushback_package, + "requires_reaudit": True, + } + if action.kind == "rebuild": + return { + "kind": "rebuild", + "parent_task_id": task_id_of(parent_task), + "module_id": action.next_state.module_id, + "contract_version": action.next_state.contract_version, + "scope": "rebuild", + "pushback_package": action.pushback_package, + "requires_reaudit": True, + } + return None + + +def derive_notification(action): + """Notification/escalation descriptor for an Action, or + None. Escalations carry the FULL round history and the + caller-supplied cost estimate (never just the last round); + a pending acceptance rebuild carries the driver's notice + content (failure reasons, rebuild scope, retained assets, + first gate).""" + if action.kind == "pending_user_confirmation": + content = None + for event in (action.ledger_events or []): + if event.get("event_type") == "audit_rebuild_notice": + content = event.get("payload", {}).get( + "notification_content") + return {"party": "user", "content": content, + "rebuild_no": action.rebuild_no, + "confirmation_status": "pending"} + if action.kind == "escalate": + return {"party": action.escalation_target, + "reason": action.reason, + "escalation_history": action.escalation_history, + "cost_estimate": action.cost_estimate, + "grants_user_acceptance": False} + return None + + +def _finish_process(conn, ledger_store, task, action, + old_impl_ref=None, user_id="system", + ledger_task_id=None): + """Ledger rows are attributed to ``ledger_task_id`` (the + loop's root task) when given, else to the task passed in. + This matters for fix completions: the #247 reader replays + ONE task's chain, so a passing re-audit written under a + transient fix-subtask id would never unblock the parent + task's merge. The server handler therefore passes the root + task id (following forked_from) for every loop call; the + loop-state row itself still records the calling task id + for traceability. + + Shared tail of the process_* functions: ledger first + (fail-closed, raises before any state save), then persist, + then derive the subtask/notification descriptors. A + quota_blocked action is the exception that proves the + ledger-first rule: its state is returned unchanged, so it + is deliberately NOT saved (saving an identical state would + still be harmless, but not saving is the auditable proof + that no round and no rebuild was consumed).""" + ledger_id = ledger_task_id or task_id_of(task) + if action.kind == "quota_blocked": + # Only the compat wiring row is written; the loop state + # is returned exactly as the driver returned it. + write_loop_ledger_events(ledger_store, ledger_id, + task_project_id(task), action, + old_impl_ref=old_impl_ref, + user_id=user_id) + return { + "action_kind": action.kind, + "reason": action.reason, + "next_state": action.next_state, + "merge_blocked": action.merge_blocked, + "release_blocked": action.release_blocked, + "subtask": None, + "notification": None, + "requires_reaudit": action.requires_reaudit, + "escalation_grants_user_acceptance": False, + "state_saved": False, + } + write_loop_ledger_events(ledger_store, ledger_id, + task_project_id(task), action, + old_impl_ref=old_impl_ref, + user_id=user_id) + save_loop_state(conn, task_id_of(task), action.next_state) + return { + "action_kind": action.kind, + "reason": action.reason, + "next_state": action.next_state, + "merge_blocked": action.merge_blocked, + "release_blocked": action.release_blocked, + "subtask": derive_subtask(action, task), + "notification": derive_notification(action), + "requires_reaudit": action.requires_reaudit, + "escalation_grants_user_acceptance": False, + "state_saved": True, + } + + +def process_audit_verdict(conn, ledger_store, task, verdict, *, + quota_check=None, cost_threshold=None, + old_impl_ref=None, user_id="system", + ledger_task_id=None): + """One independent audit/acceptance verdict -> loop step. + + The original task's status is never touched here (the + server handler likewise does not change it): a fail blocks + merge/release via the chain and derives a fix subtask, it + does not roll the completed task back into a re-run.""" + if not isinstance(verdict, dict): + raise TypeError("verdict must be a dict") + module_id = resolve_module_id(task) + contract_version = resolve_contract_version(task, verdict) + state = load_loop_state(conn, module_id, contract_version) + if state is None: + state = _driver.new_loop_state(module_id, contract_version) + action = _driver.decide_audit( + state, verdict, quota_check=quota_check, + cost_threshold=cost_threshold) + return _finish_process(conn, ledger_store, task, action, + old_impl_ref=old_impl_ref, + user_id=user_id, + ledger_task_id=ledger_task_id) + + +def process_fix_completion(conn, ledger_store, task, fix_result, *, + automated=None, quota_check=None, + cost_threshold=None, old_impl_ref=None, + user_id="system", ledger_task_id=None): + """A fix landing -> automatic FULL re-audit. + + ``automated`` is {"hard_gate": {...}, "extra_checks": {...}} + or None (no automated evidence supplied: the re-audit then + rests entirely on the auditor verdict, and the caller must + say so in its own records -- this function never claims an + automated pass it did not run). When automated evidence IS + supplied and any check fails, a claimed pass verdict is + forced to fail with the automated failures appended, before + the driver ever sees it. The driver then enforces the rest: + a missing reaudit verdict or diff_only raises ValueError, + and only the re-audit verdict counts (the fix itself never + counts). The loop state must already exist: a fix with no + prior loop is refused rather than inventing a fresh loop + that a pass could silently clear. + """ + if not isinstance(fix_result, dict): + raise TypeError("fix_result must be a dict") + reaudit = (fix_result.get("reaudit_verdict") + or fix_result.get("reaudit") + or fix_result.get("verdict")) + module_id = resolve_module_id(task) + # Load by the task's own contract version first: that is + # the loop's current version, and the driver itself applies + # a version change carried by the re-audit verdict (reset). + # Loading by the verdict version first would mistake a + # version-change re-audit for a loop that never started. + contract_version = None + state = None + task_version = None + if isinstance(task, dict): + task_version = task.get("contract_version") + if isinstance(task_version, str) and task_version.strip(): + state = load_loop_state(conn, module_id, + task_version.strip()) + contract_version = task_version.strip() + if state is None: + contract_version = resolve_contract_version(task, reaudit) + state = load_loop_state(conn, module_id, contract_version) + if state is None: + raise ValueError( + "no audit loop state for module %s@%s: a fix cannot " + "re-audit a loop that never started" % ( + module_id, contract_version)) + automated_result = None + if automated is not None: + automated_result = evaluate_automated_reaudit( + automated.get("hard_gate"), + automated.get("extra_checks")) + if not automated_result["automated_checks_pass"]: + forced = dict(fix_result) + verdict = dict(reaudit) if isinstance(reaudit, dict) \ + else {} + verdict["verdict"] = "fail" + failures = list(verdict.get("failures") or []) + failures.extend( + {"check": check, + "detail": "automated re-audit check failed"} + for check in automated_result["failed_checks"]) + verdict["failures"] = failures + forced["reaudit_verdict"] = verdict + fix_result = forced + action = _driver.after_fix( + state, fix_result, quota_check=quota_check, + cost_threshold=cost_threshold) + result = _finish_process(conn, ledger_store, task, action, + old_impl_ref=old_impl_ref, + user_id=user_id, + ledger_task_id=ledger_task_id) + result["automated_checks"] = automated_result + return result + + +def process_correction(conn, ledger_store, task, correction, *, + user_id="system", ledger_task_id=None): + """An audit-of-audit correction. Semantics are the driver's + (#247, unchanged): corrected-to-fail re-blocks at once and + counts a failure; corrected-to-pass does NOT unblock (it + schedules a re-audit, keeps merge/release blocked, zeroes + the effective failure count). The original chain row is + never edited: only the new meta-review row is written.""" + if not isinstance(correction, dict): + raise TypeError("correction must be a dict") + module_id = resolve_module_id(task) + contract_version = resolve_contract_version( + task, {"contract_version": correction.get("contract_version")} + if correction.get("contract_version") else None) + state = load_loop_state(conn, module_id, contract_version) + if state is None: + state = _driver.new_loop_state(module_id, contract_version) + action = _driver.apply_correction(state, correction) + result = _finish_process(conn, ledger_store, task, action, + user_id=user_id, + ledger_task_id=ledger_task_id) + result["correction_action"] = action.correction_action + return result + + +def process_rebuild_confirmation(conn, ledger_store, task, *, + user_confirmed, confirmed_by, + quota_check=None, + cost_threshold=None, + cost_estimate=None, + old_impl_ref=None, + user_id="system", + ledger_task_id=None): + """Resolve a pending acceptance-triggered rebuild. + + ``confirmed_by`` must be a real user id: "system" and + "coordinator" are refused here, and the server endpoint + additionally requires it to equal the authenticated caller + with an explicit user_confirmed body flag. Nothing in this + module calls this on a user's behalf.""" + if (not isinstance(confirmed_by, str) + or confirmed_by.strip() in _RESERVED_CONFIRMERS): + raise ValueError( + "rebuild confirmation requires a real user id as " + "confirmed_by, not a system/coordinator marker") + module_id = resolve_module_id(task) + contract_version = resolve_contract_version(task) + state = load_loop_state(conn, module_id, contract_version) + if state is None: + raise ValueError( + "no audit loop state for module %s@%s" % ( + module_id, contract_version)) + action = _driver.confirm_rebuild( + state, user_confirmed=bool(user_confirmed), + quota_check=quota_check, cost_threshold=cost_threshold, + cost_estimate=cost_estimate) + return _finish_process(conn, ledger_store, task, action, + old_impl_ref=old_impl_ref, + user_id=user_id, + ledger_task_id=ledger_task_id) + + +# ------------------------------------------------------------------ +# Release gate (#259 contract, pure evaluator) +# ------------------------------------------------------------------ + +def evaluate_release_gate(task_id, conn, *, task_status=None, + hard_gate=None): + """Pure release decision. Inputs, in #259 evaluation order: + + 1. Chain read via the SAME #247 reader merge uses. Any + read failure (other than a never-written chain, which + the reader reports as unblocked) short-circuits to + release_blocked_state_unavailable. + 2. Task status: must be "merged" (release_requires_merged). + 3. Fresh hard gate: None means evidence missing + (release_hard_gate_evidence_missing); allowed=False + means release_blocked_hard_gate with blocks verbatim. + 4. Audit pushback: reader blocked means + release_blocked_audit_pushback with failed_checks / + round / modules verbatim. + + Returns {allowed, http_status, code, blocked_by, + blocked_sources, task_status, hard_gate, audit_pushback}. + ``blocked_sources`` lists every blocking source when + several block (chain failure excepted, see above). This + evaluator never writes anything: no audit, no chain row, + no status change. The endpoint's 200/409 handling must + preserve that read-only property for the check itself; + any success record is an explicit post-check action, not + part of this evaluation. + """ + try: + pushback = _wiring.pushback_block_for_task(conn, task_id) + except Exception: + return { + "allowed": False, "http_status": 409, + "code": "release_blocked_state_unavailable", + "blocked_by": "audit_pushback_state_unavailable", + "blocked_sources": ["audit_pushback_state_unavailable"], + "task_status": task_status, "hard_gate": hard_gate, + "audit_pushback": None, + } + sources = [] + first_code = None + first_by = None + + def _add(code, by): + nonlocal first_code, first_by + sources.append(by) + if first_code is None: + first_code, first_by = code, by + + if task_status != "merged": + _add("release_requires_merged", "task_status") + if hard_gate is None: + _add("release_hard_gate_evidence_missing", "hard_gate") + elif not hard_gate.get("allowed"): + _add("release_blocked_hard_gate", "hard_gate") + if pushback.get("merge_blocked") or pushback.get("release_blocked"): + _add("release_blocked_audit_pushback", "audit_pushback") + if not sources: + return { + "allowed": True, "http_status": 200, + "code": "release_allowed", "blocked_by": None, + "blocked_sources": [], "task_status": task_status, + "hard_gate": hard_gate, "audit_pushback": pushback, + } + return { + "allowed": False, "http_status": 409, + "code": first_code, "blocked_by": first_by, + "blocked_sources": sources, "task_status": task_status, + "hard_gate": hard_gate, "audit_pushback": pushback, + } diff --git a/cloud/server.py b/cloud/server.py index ef75e6f..017f3b1 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -106,6 +106,16 @@ from cloud.marker_tools import parse_spawn_defaults as _parse_spawn_defaults from cloud.merge_gate import evaluate_merge_gate as _merge_gate_evaluate from cloud.merge_gate import protection_from_env as _merge_gate_protection from cloud.audit_wiring import pushback_block_for_task as _pushback_block +from cloud.audit_loop_wiring import build_quota_check as _loop_quota_check +from cloud.audit_loop_wiring import cost_threshold_from_env as _loop_cost_threshold +from cloud.audit_loop_wiring import evaluate_release_gate as _release_gate_evaluate +from cloud.audit_loop_wiring import process_audit_verdict as _loop_verdict +from cloud.audit_loop_wiring import process_correction as _loop_correction +from cloud.audit_loop_wiring import process_fix_completion as _loop_fix +from cloud.audit_loop_wiring import process_rebuild_confirmation as _loop_confirm +from cloud.audit_loop_wiring import state_to_dict as _loop_state_to_dict +from cloud.audit_loop_wiring import load_loop_state as _loop_load_state +from cloud.ledger import LedgerStore as _LedgerStore from cloud.merge_gate_hard import blocked_audit_detail as _gate_blocked_detail from cloud.merge_gate_hard import evaluate_hard_merge_gate as _merge_gate_hard from cloud.merge_gate_hard import hard_mode_from_env as _merge_gate_hard_mode @@ -2781,6 +2791,22 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._pause_task(user_id, rest[: -len("/pause")]) elif rest.endswith("/merge") and method == "POST": self._merge_task(user_id, rest[: -len("/merge")]) + elif rest.endswith("/release") and method == "POST": + self._release_task(user_id, rest[: -len("/release")]) + elif rest.endswith("/audit-verdict") and method == "POST": + self._audit_loop_verdict(user_id, + rest[: -len("/audit-verdict")]) + elif rest.endswith("/audit-fix") and method == "POST": + self._audit_loop_fix(user_id, rest[: -len("/audit-fix")]) + elif rest.endswith("/audit-correction") and method == "POST": + self._audit_loop_correction( + user_id, rest[: -len("/audit-correction")]) + elif rest.endswith("/audit-rebuild-confirm") and method == "POST": + self._audit_loop_confirm( + user_id, rest[: -len("/audit-rebuild-confirm")]) + elif rest.endswith("/audit-loop") and method == "GET": + self._audit_loop_get(user_id, rest[: -len("/audit-loop")], + query) elif rest.endswith("/pr") and method == "POST": self._task_pr(user_id, rest[: -len("/pr")]) elif rest.endswith("/retry") and method == "POST": @@ -8734,6 +8760,504 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): task["merge_gate"] = gate self._send_json(200, task) + # -- 项目级审计循环接线（#260 驱动的服务端执行层） -------------- + # 决策一律来自 cloud/audit_loop_driver，本节只做执行：持久化 + # 循环状态、经 #241/#247 写链、派生修复/重做子任务、拦合并与 + # 发布。合并端点本身不改：它继续用 #247 reader + 五项硬门禁 + # 做 AND，本节不给合并造第二套 reader。原任务状态在本节的 + # 任何一步都不被回退：fail 只阻塞 + 派生子任务，completed + # 保持 completed，直到合并端点按门禁把它标 merged。 + def _loop_task_for_body(self, task, body): + """把请求体里的模块/合同声明并入任务 dict，供接线层按 + (module_id, contract_version) 解析循环键。任务行本身没 + 有 module 列，所以模块归属必须由提交方在每次调用里显 + 式声明（或沿用任务 id 回落）；接线层绝不悄悄猜模块。""" + merged = dict(task) + if isinstance(body.get("module_id"), str) \ + and body.get("module_id").strip(): + merged["module_id"] = body["module_id"].strip() + if isinstance(body.get("contract_version"), str) \ + and body.get("contract_version").strip(): + merged["contract_version"] = body["contract_version"].strip() + return merged + + def _loop_quota_check(self, task, caller_id): + """服务端配额 -> 驱动 quota_check：配额归属取原任务 + 创建人（成本归属口径与子任务一致），无创建人时取调用 + 人。_quota_status 抛错时闭包同样抛错，驱动 fail-closed + 为 quota_blocked，绝不把查不到配额当成放行。""" + quota_user = task.get("created_by") or caller_id + + def _check(estimate): + quota, used, _month = self._quota_status(quota_user) + return _loop_quota_check(quota, used)(estimate) + return _check + + def _loop_root_task_id(self, task): + """循环根任务 id：沿 forked_from 回溯到链头（最多 20 + 步防环）。链行必须归属根任务，否则 #247 的按任务读链 + 在修复子任务上写下的通过结论永远解不开父任务的合并 + ——这是本节最容易踩错的归属点，单列一处 helper。""" + current = task + for _step in range(20): + parent_id = current.get("forked_from") + if not parent_id: + return current["id"] + parent_row = self._fetch_task(parent_id) + if parent_row is None: + return current["id"] + current = _task_to_dict(parent_row) + return current["id"] + + def _loop_ledger_store(self): + """#241/#247 写路径的 store：LedgerStore 自带连接与 + ensure_schema，每条 record_* 调用内提交。调用方用完 + 必须 close（各 handler 的 finally 里做）。""" + return _LedgerStore(self.server.db_path) + + def _loop_create_subtask(self, descriptor, parent_task, caller_id): + """把接线层返回的子任务描述符落成一条 queued 任务行： + 同项目、prompt 写明模块/合同/失败项与完整重审标准、 + created_by 取原任务创建人（成本归属）、forked_from 记 + 父任务。返回新任务 dict；INSERT 失败时抛错，由调用方 + 按 503 回执（循环已推进、描述符在响应里，调用方可按 + 描述符经 POST /tasks 手工补建，绝不悄悄重跑决策）。""" + package = descriptor.get("pushback_package") or {} + failures = package.get("failures") or [] + criteria = package.get("reaudit_criteria") or [] + lines = [ + "Audit %s subtask for module %s@%s (parent task %s)." + % (descriptor.get("kind"), descriptor.get("module_id"), + descriptor.get("contract_version"), + descriptor.get("parent_task_id")), + "Scope: %s." % descriptor.get("scope"), + "Failures: %s" % json.dumps(failures, ensure_ascii=False)[:2000], + "Full re-audit criteria (all must pass, not only the " + "failed items): %s" % ", ".join(str(c) for c in criteria), + ] + if descriptor.get("targeted_items"): + lines.append("Targeted items only: %s" + % json.dumps(descriptor["targeted_items"], + ensure_ascii=False)[:1000]) + if package.get("retained_assets"): + lines.append("Retained assets (do not rebuild these): %s" + % json.dumps(package["retained_assets"], + ensure_ascii=False)[:2000]) + prompt = "\n".join(lines) + subtask_id = uuid.uuid4().hex + now = time.time() + conn = self._db() + try: + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " created_by, created_at, updated_at, forked_from)" + " VALUES(?, ?, ?, 'queued', ?, ?, ?, ?)", + (subtask_id, parent_task["project_id"], prompt, + parent_task.get("created_by") or caller_id, now, now, + parent_task["id"])) + conn.commit() + row = conn.execute( + "SELECT * FROM tasks WHERE id = ?", (subtask_id,) + ).fetchone() + return _task_to_dict(row) + finally: + conn.close() + + def _loop_run(self, user_id, task_id, body, kind): + """循环端点的公共骨架：404 / approve_merge 403 / 体校验 / + 接线调用 / 子任务落行。ValueError/TypeError（驱动拒绝、 + 映射缺失、写链预校验）回 400；其它异常（账本写失败等） + 回 503 且状态未推进（接线层先写链后存状态的顺序保证 + 了这一点）；子任务落行失败同样 503，但循环已推进的事 + 实在响应里明说，不谎称“未推进”。""" + row = self._fetch_task(task_id) + if row is None: + self._send_json(404, {"error": "task not found"}) + return + task = _task_to_dict(row) + with _auth_store(self.server.db_path) as store: + if not store.check_permission( + user_id, task["project_id"], "approve_merge"): + self._send_json(403, {"error": "forbidden"}) + return + loop_task = self._loop_task_for_body(task, body) + store = self._loop_ledger_store() + loop_conn = self._db() + try: + if kind == "verdict": + verdict = body.get("verdict") + if not isinstance(verdict, dict) \ + or not verdict.get("verdict"): + self._send_json(400, { + "error": "verdict dict with a verdict " + "field is required"}) + return + result = _loop_verdict( + loop_conn, store, loop_task, verdict, + quota_check=self._loop_quota_check(task, user_id), + cost_threshold=_loop_cost_threshold(), + old_impl_ref=body.get("old_impl_ref"), + user_id=user_id, + ledger_task_id=self._loop_root_task_id(task)) + elif kind == "fix": + fix_result = body.get("fix_result") + if not isinstance(fix_result, dict): + self._send_json(400, { + "error": "fix_result dict is required"}) + return + result = _loop_fix( + loop_conn, store, loop_task, fix_result, + automated=body.get("automated"), + quota_check=self._loop_quota_check(task, user_id), + cost_threshold=_loop_cost_threshold(), + old_impl_ref=body.get("old_impl_ref"), + user_id=user_id, + ledger_task_id=self._loop_root_task_id(task)) + elif kind == "correction": + correction = body.get("correction") + if not isinstance(correction, dict): + self._send_json(400, { + "error": "correction dict is required"}) + return + if isinstance(body.get("contract_version"), str): + correction = dict( + correction, + contract_version=body["contract_version"]) + result = _loop_correction( + loop_conn, store, loop_task, correction, + user_id=user_id, + ledger_task_id=self._loop_root_task_id(task)) + else: + raise ValueError("unknown loop kind: %r" % (kind,)) + except (ValueError, TypeError) as exc: + self._send_json(400, {"error": str(exc)}) + return + except Exception as exc: + self._send_json(503, { + "error": "audit_loop_unavailable", + "detail": "ledger or loop state write failed; " + "loop state was not advanced: %s" % exc}) + return + finally: + try: + store.close() + except Exception: + pass + try: + loop_conn.close() + except Exception: + pass + subtask_task = None + subtask_error = None + descriptor = result.get("subtask") + if descriptor is not None and descriptor.get("kind") != "reaudit": + try: + subtask_task = self._loop_create_subtask( + descriptor, task, user_id) + except Exception as exc: + subtask_error = str(exc) + payload = { + "task": task, + "loop_state": _loop_state_to_dict(result["next_state"]), + "action_kind": result["action_kind"], + "reason": result["reason"], + "merge_blocked": result["merge_blocked"], + "release_blocked": result["release_blocked"], + "requires_reaudit": result["requires_reaudit"], + "escalation_grants_user_acceptance": + result["escalation_grants_user_acceptance"], + "subtask": descriptor, + "subtask_task": subtask_task, + "notification": result["notification"], + } + if "automated_checks" in result: + payload["automated_checks"] = result["automated_checks"] + if "correction_action" in result: + payload["correction_action"] = result["correction_action"] + if subtask_error is not None: + payload["subtask_error"] = subtask_error + payload["subtask_note"] = ( + "loop state advanced and ledger written, but the " + "subtask row was not created; create it from the " + "subtask descriptor via POST /tasks, do not " + "re-submit the verdict") + self._send_json(503, payload) + return + self._send_json(200, payload) + + def _audit_loop_verdict(self, user_id, task_id): + """POST /tasks/{id}/audit-verdict：提交一次独立审计或 + 验收结论。权限：项目 approve_merge（审计提交方须是 + maintainer 以上，与合并权同级；普通 contributor 不能 + 凭空给循环喂结论）。体：{module_id, contract_version, + verdict: {verdict, source, failures, ...}, + old_impl_ref?}。原任务状态不动；fail 自动派生修复子 + 任务并经链阻塞合并/发布。""" + body, jerr = self._read_json() + if jerr: + self._send_json(400, {"error": jerr}) + return + self._loop_run(user_id, task_id, body, "verdict") + + def _audit_loop_fix(self, user_id, task_id): + """POST /tasks/{id}/audit-fix：修复完成事件。体： + {module_id, contract_version, fix_result: + {reaudit_verdict: {...}}, automated?: {hard_gate: {...}, + extra_checks: {...}}, old_impl_ref?}。注意“自动完整 + 重审”的准确含义：本端点把修复完成推进到重审（跑自动 + 化检查 + 要求完整重审结论），而不是自动判通过—— + reaudit_verdict 仍须审计方提交，缺失或 diff_only 由 + 驱动以 400 拒绝；自动化检查全过也绝不等于审计通过。 + 本端点调用的任务应为修复子任务（其 module 与父任务 + 相同，体里显式声明），原任务与修复子任务的状态本端 + 点都不改（修复子任务的 completed 由 worker/调用方写 + 回，本端点只推进循环）。""" + body, jerr = self._read_json() + if jerr: + self._send_json(400, {"error": jerr}) + return + self._loop_run(user_id, task_id, body, "fix") + + def _audit_loop_correction(self, user_id, task_id): + """POST /tasks/{id}/audit-correction：审计的审计纠错。 + 体：{module_id, contract_version, correction: + {original_verdict, corrected_verdict, reviewer, + original_entry_id, ...}}。纠正为 fail 立即重拦；纠正 + 为 pass 不解封、只安排重审（原链行永不改动，只追加 + meta-review 行）。""" + body, jerr = self._read_json() + if jerr: + self._send_json(400, {"error": jerr}) + return + self._loop_run(user_id, task_id, body, "correction") + + def _audit_loop_confirm(self, user_id, task_id): + """POST /tasks/{id}/audit-rebuild-confirm：验收触发推 + 倒的用户确认。体必须显式带 user_confirmed（true/false） + 与 confirmed_by，且 confirmed_by 必须等于认证调用方 + 本人：协调员或系统没有任何路径可以代用户确认（本端 + 点之外没有任何代码调用确认流程；pending 状态下不建 + 重做任务、不耗推倒次数）。确认后才进 rebuilding，拒 + 绝则 escalate 且不耗预算；确认不等于 G6 验收。""" + row = self._fetch_task(task_id) + if row is None: + self._send_json(404, {"error": "task not found"}) + return + task = _task_to_dict(row) + with _auth_store(self.server.db_path) as store: + if not store.check_permission( + user_id, task["project_id"], "approve_merge"): + self._send_json(403, {"error": "forbidden"}) + return + body, jerr = self._read_json() + if jerr: + self._send_json(400, {"error": jerr}) + return + if "user_confirmed" not in body \ + or not isinstance(body.get("user_confirmed"), bool): + self._send_json(400, { + "error": "user_confirmed boolean is required; " + "confirmation must be explicit"}) + return + if body.get("confirmed_by") != user_id: + self._send_json(403, { + "error": "confirmed_by must equal the authenticated " + "caller; no coordinator or system may " + "confirm on a user's behalf"}) + return + loop_task = self._loop_task_for_body(task, body) + store = self._loop_ledger_store() + loop_conn = self._db() + try: + result = _loop_confirm( + loop_conn, store, loop_task, + user_confirmed=body["user_confirmed"], + confirmed_by=body["confirmed_by"], + quota_check=self._loop_quota_check(task, user_id), + cost_threshold=_loop_cost_threshold(), + cost_estimate=body.get("cost_estimate"), + old_impl_ref=body.get("old_impl_ref"), + user_id=user_id, + ledger_task_id=self._loop_root_task_id(task)) + except (ValueError, TypeError) as exc: + self._send_json(400, {"error": str(exc)}) + return + except Exception as exc: + self._send_json(503, { + "error": "audit_loop_unavailable", + "detail": "ledger or loop state write failed; " + "loop state was not advanced: %s" % exc}) + return + finally: + try: + store.close() + except Exception: + pass + try: + loop_conn.close() + except Exception: + pass + subtask_task = None + subtask_error = None + descriptor = result.get("subtask") + if descriptor is not None: + try: + subtask_task = self._loop_create_subtask( + descriptor, task, user_id) + except Exception as exc: + subtask_error = str(exc) + payload = { + "task": task, + "loop_state": _loop_state_to_dict(result["next_state"]), + "action_kind": result["action_kind"], + "reason": result["reason"], + "merge_blocked": result["merge_blocked"], + "release_blocked": result["release_blocked"], + "subtask": descriptor, + "subtask_task": subtask_task, + "notification": result["notification"], + } + if subtask_error is not None: + payload["subtask_error"] = subtask_error + self._send_json(503, payload) + return + self._send_json(200, payload) + + def _audit_loop_get(self, user_id, task_id, query): + """GET /tasks/{id}/audit-loop?module_id=&contract_version=： + 读循环持久化状态 + 同一 #247 reader 的链阻塞视图。 + 权限：项目 read。读链失败（表不存在除外）回 503、不 + 谎称无阻塞。""" + row = self._fetch_task(task_id) + if row is None: + self._send_json(404, {"error": "task not found"}) + return + task = _task_to_dict(row) + with _auth_store(self.server.db_path) as store: + if not store.check_permission( + user_id, task["project_id"], "read"): + self._send_json(403, {"error": "forbidden"}) + return + module_id = (query.get("module_id") or [None])[0] + contract_version = (query.get("contract_version") or [None])[0] + if not module_id or not contract_version: + self._send_json(400, { + "error": "module_id and contract_version query " + "parameters are required"}) + return + conn = self._db() + try: + state = _loop_load_state(conn, module_id, contract_version) + finally: + conn.close() + if state is None: + self._send_json(404, {"error": "no audit loop state"}) + return + try: + conn = self._db() + try: + pushback = _pushback_block(conn, task_id) + finally: + conn.close() + except Exception: + self._send_json(503, { + "error": "audit_pushback_state_unavailable"}) + return + self._send_json(200, { + "task": task, + "loop_state": _loop_state_to_dict(state), + "audit_pushback": pushback, + }) + + def _release_task(self, user_id, task_id): + """POST /tasks/{id}/release：发布门禁 + 发布授权（#259 + 契约正式建路由）。权限：只认 CODEX_GLOBAL_ADMINS 的 + 全局管理员写（与 verify/export 同身份来源，但 verify/ + export 是只读、发布是写；项目 approve_merge 不能放行 + 发布）。三者 AND：① 发布时刻重读 #247 同一 reader 的 + 链；② 发布时刻调用方新提交的硬门禁证据重算五项（合 + 并时的 200 永远不作输入；硬门禁一律按 hard 算，模式环 + 境变量不能把发布降级）；③ 任务状态须为 merged。 + 阻塞源区分见 release_gate：blocked_sources 列出全部， + code/blocked_by 取求值顺序第一项。409 阻塞不写任何审 + 计（检查写链会移动链头、使锚点失效）；403 被拒沿用 + _audit_admin 惯例记审计。200 不改任务状态（本站没有 + released 状态）：200 是“发布已授权”，真正的部署动作 + 在平台部署通道（deploy.py）里，它部署平台本身、无任 + 务 id、不走服务端 HTTP 鉴权，因此不在任务门禁内—— + 把它绑到任务链上会在最需要部署修复与回滚时反锁部署。 + 200 时写一条显式的 release_task 审计作为发布后动作记 + 录（检查本身是只读的，这条是检查完成后的显式动作， + 不是检查的副产品；锚点须在这条之后重取）。""" + row = self._fetch_task(task_id) + if row is None: + self._send_json(404, {"error": "task not found"}) + return + task = _task_to_dict(row) + if not self._is_global_admin(user_id): + self._audit_admin(user_id, "release_denied", task_id) + self._send_json(403, {"error": "forbidden"}) + return + body, jerr = self._read_json() + if jerr: + self._send_json(400, {"error": jerr}) + return + if body.get("files_changed") is not None and not isinstance( + body.get("files_changed"), list): + self._send_json(400, {"error": "files_changed must be a list"}) + return + hard_err = _validate_hard_gate_body(body) + if hard_err: + self._send_json(400, {"error": hard_err}) + return + # 发布时刻的新证据：体里没给任何硬门禁字段时，hard_gate + # 为 None -> 证据缺失 409，而不是用合并时的旧结论补。 + has_evidence = any( + body.get(field) is not None + for field in ("files_changed", "ci_status", "ci_checks", + "required_checks", "contract_version", + "contract_frozen", "coordinator_approved", + "unresolved_review_comments", + "branch_protection")) + hard_gate = None + if has_evidence: + protection = body.get("branch_protection") \ + or _merge_gate_protection() + hard_gate = _merge_gate_hard( + files_changed=body.get("files_changed"), + owned_dir=body.get("owned_dir"), + approved_exceptions=body.get("approved_exceptions") or (), + protection=protection, + ci_status=body.get("ci_status"), + ci_checks=body.get("ci_checks"), + required_checks=body.get("required_checks"), + contract_version=body.get("contract_version"), + contract_frozen=body.get("contract_frozen", False), + contract_expected_version=body.get( + "contract_expected_version"), + coordinator_approved=body.get("coordinator_approved"), + unresolved_review_comments=body.get( + "unresolved_review_comments")) + conn = self._db() + try: + decision = _release_gate_evaluate( + task_id, conn, task_status=task["status"], + hard_gate=hard_gate) + finally: + conn.close() + if not decision["allowed"]: + # 409 不写审计：见本方法 docstring。 + self._send_json(409, { + "error": "release blocked", + "release_gate": decision, + }) + return + with _auth_store(self.server.db_path) as store: + store.audit(user_id, task["project_id"], "release_task", + task_id) + task["release_gate"] = decision + self._send_json(200, task) + # -- 运行时审批队列 ------------------------------------------------ def _list_approvals(self, user_id, task_id): """列出某任务的审批项（含已决定的，回执一并带出）。 diff --git a/cloud/tests/test_audit_loop_wiring.py b/cloud/tests/test_audit_loop_wiring.py new file mode 100644 index 0000000..1f8e812 --- /dev/null +++ b/cloud/tests/test_audit_loop_wiring.py @@ -0,0 +1,439 @@ +"""Server wiring for the #260 loop driver (track 2, module level). + +Uses a real #241/#247 chain (LedgerStore on a temp file) and a +real sqlite loop-state connection -- no production tasks, no +GitHub, no production DB. The one fake is FailingLedgerStore, +used only to prove ledger-failure fail-closed behaviour. +""" + +import os +import sqlite3 +import sys +import tempfile +import unittest +from pathlib import Path + +try: + from cloud import audit_loop_wiring as wiring + from cloud.audit_wiring import pushback_block_for_task + from cloud.ledger import LedgerStore + from cloud.ledger import verify_chain +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud import audit_loop_wiring as wiring + from cloud.audit_wiring import pushback_block_for_task + from cloud.ledger import LedgerStore + from cloud.ledger import verify_chain + +MODULE = "mod-a" +VERSION = "v1" +OLD_REF = {"archive_ref": "task:parent:pre-rebuild"} +EXTRA_PASS = { + "security_unconditional": True, + "regression_exact_artifact": True, + "demo_runnable": True, + "known_deviation_due_todo": True, + "audit_signer_separation": True, + "qa_failed_links_rerun": True, +} +HARD_PASS = { + "files_changed": ["apps/web/a.py"], + "owned_dir": "apps/web", + "protection": "protected", + "ci_status": "success", + "ci_checks": [{"name": "ci", "status": "success"}], + "required_checks": ["ci"], + "contract_version": "v1", + "contract_frozen": True, + "coordinator_approved": True, + "unresolved_review_comments": 0, +} + + +def fail_verdict(text="check x failed", source="audit", **extra): + verdict = { + "verdict": "fail", + "source": source, + "verdict_text": text, + "failures": [{"check": "ci_green", "detail": text}], + "passed_items": ["a", "b", "c"], + "contract_version": VERSION, + } + verdict.update(extra) + return verdict + + +def pass_verdict(**extra): + verdict = { + "verdict": "pass", + "source": "audit", + "verdict_text": "all green", + "failures": [], + "passed_items": ["a", "b", "c"], + "contract_version": VERSION, + } + verdict.update(extra) + return verdict + + +class FailingLedgerStore: + """Every append raises: proves the wiring never saves state + after a ledger failure.""" + + def __getattr__(self, name): + if name.startswith("append_"): + def _raise(*args, **kwargs): + raise RuntimeError("ledger down") + return _raise + raise AttributeError(name) + + +class WiringTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.ledger_path = os.path.join(self.tmp.name, "ledger.db") + self.store = LedgerStore(self.ledger_path) + self.conn = sqlite3.connect(":memory:") + self.task = {"id": "task-parent", "project_id": "proj-1", + "module_id": MODULE, "contract_version": VERSION} + + def tearDown(self): + try: + self.store.close() + except Exception: + pass + self.conn.close() + self.tmp.cleanup() + + def _block(self, task_id="task-parent"): + return pushback_block_for_task(self.store.conn, task_id) + + def _verdict(self, verdict, **kwargs): + kwargs.setdefault("user_id", "auditor-1") + return wiring.process_audit_verdict( + self.conn, self.store, self.task, verdict, **kwargs) + + def test_module_mapping_explicit_wins_and_fallback_is_task_id(self): + self.assertEqual( + wiring.resolve_module_id( + {"id": "t1", "module_id": "m-explicit"}), "m-explicit") + self.assertEqual( + wiring.resolve_module_id({"id": "t1"}), "t1") + self.assertEqual( + wiring.resolve_module_id({"id": "t1", "module": "m2"}), "m2") + with self.assertRaises(ValueError): + wiring.resolve_contract_version({"id": "t1"}, None) + self.assertEqual( + wiring.resolve_contract_version( + {"id": "t1"}, {"contract_version": "v9"}), "v9") + + def test_state_persists_and_reloads_from_table_not_memory(self): + result = self._verdict(fail_verdict()) + self.assertEqual(result["action_kind"], "pushback") + self.assertEqual(result["subtask"]["kind"], "fix") + self.assertTrue(result["merge_blocked"]) + reloaded = wiring.load_loop_state(self.conn, MODULE, VERSION) + self.assertEqual(reloaded.pushback_round, 1) + self.assertEqual(reloaded.failure_count, 1) + self.assertEqual(reloaded.status, "pushback") + self.assertTrue(self._block()["merge_blocked"]) + # The chain itself verifies: the wiring wrote real rows. + self.assertTrue(verify_chain(self.store.conn)["ok"]) + + def test_full_chain_fail_fix_reaudit_pass_unblocks(self): + first = self._verdict(fail_verdict()) + fix_task = {"id": "task-fix-1", "project_id": "proj-1", + "module_id": MODULE, "contract_version": VERSION} + result = wiring.process_fix_completion( + self.conn, self.store, fix_task, + {"reaudit_verdict": pass_verdict()}, + automated={"hard_gate": HARD_PASS, "extra_checks": EXTRA_PASS}, + user_id="auditor-1", ledger_task_id="task-parent") + self.assertEqual(result["action_kind"], "pass") + self.assertIsNone(result["subtask"]) + self.assertFalse(result["merge_blocked"]) + self.assertTrue( + result["automated_checks"]["automated_checks_pass"]) + # Parent task dict untouched by the wiring layer: the + # original task is never rolled back by this module. + self.assertNotIn("status", self.task) + self.assertFalse(self._block()["merge_blocked"]) + state = wiring.load_loop_state(self.conn, MODULE, VERSION) + self.assertEqual(state.status, "passed") + self.assertEqual(state.failure_count, 0) + + def test_automated_failure_forces_claimed_pass_to_fail(self): + self._verdict(fail_verdict()) + fix_task = {"id": "task-fix-1", "project_id": "proj-1", + "module_id": MODULE, "contract_version": VERSION} + broken_extra = dict(EXTRA_PASS, demo_runnable=False) + result = wiring.process_fix_completion( + self.conn, self.store, fix_task, + {"reaudit_verdict": pass_verdict()}, + automated={"hard_gate": HARD_PASS, + "extra_checks": broken_extra}, + user_id="auditor-1", ledger_task_id="task-parent") + # Automated checks passing is never audit passing, and + # here they did not pass: the claimed pass is forced fail. + self.assertEqual(result["action_kind"], "pushback") + self.assertTrue(result["merge_blocked"]) + self.assertIn("demo_runnable", + result["automated_checks"]["failed_checks"]) + self.assertTrue(self._block()["merge_blocked"]) + + def test_fix_without_reaudit_verdict_and_diff_only_refused(self): + self._verdict(fail_verdict()) + fix_task = {"id": "task-fix-1", "project_id": "proj-1", + "module_id": MODULE, "contract_version": VERSION} + with self.assertRaises(ValueError): + wiring.process_fix_completion( + self.conn, self.store, fix_task, {"fix_diff": {}}, + user_id="auditor-1", ledger_task_id="task-parent") + with self.assertRaises(ValueError): + wiring.process_fix_completion( + self.conn, self.store, fix_task, + {"diff_only": True, + "reaudit_verdict": pass_verdict()}, + user_id="auditor-1", ledger_task_id="task-parent") + # Fix with no prior loop at all is refused, not invented. + fresh_conn = sqlite3.connect(":memory:") + try: + with self.assertRaises(ValueError): + wiring.process_fix_completion( + fresh_conn, self.store, fix_task, + {"reaudit_verdict": pass_verdict()}, + user_id="auditor-1", ledger_task_id="task-parent") + finally: + fresh_conn.close() + + def test_exactly_three_pushes_back_fourth_rebuilds(self): + for _ in range(3): + result = self._verdict(fail_verdict()) + self.assertEqual(result["action_kind"], "pushback") + # Fourth failure without a caller-supplied old impl ref + # is refused BEFORE any ledger row: state stays at 3. + with self.assertRaises(ValueError): + self._verdict(fail_verdict()) + state = wiring.load_loop_state(self.conn, MODULE, VERSION) + self.assertEqual(state.pushback_round, 3) + self.assertEqual(state.failure_count, 3) + result = self._verdict(fail_verdict(), old_impl_ref=OLD_REF) + self.assertEqual(result["action_kind"], "rebuild") + self.assertEqual(result["subtask"]["kind"], "rebuild") + state = wiring.load_loop_state(self.conn, MODULE, VERSION) + self.assertEqual(state.rebuild_no, 1) + self.assertEqual(state.status, "rebuilding") + + def test_targeted_third_round_scope(self): + self._verdict(fail_verdict()) + self._verdict(fail_verdict()) + result = self._verdict(fail_verdict( + failures=[{"check": "ci_green", "detail": "one left"}], + passed_items=["a", "b", "c"], + broken_passed_items=[])) + self.assertEqual(result["action_kind"], "pushback") + self.assertEqual(result["subtask"]["scope"], "targeted") + self.assertEqual(result["subtask"]["targeted_items"], + ["ci_green"]) + + def test_correction_to_pass_does_not_unblock(self): + self._verdict(fail_verdict()) + verdict_rows = self.store.conn.execute( + "SELECT id FROM audit_ledger WHERE action = ?" + " ORDER BY id ASC", ("audit_verdict",)).fetchall() + entry_id = verdict_rows[0][0] + result = wiring.process_correction( + self.conn, self.store, self.task, + {"original_verdict": "fail", "corrected_verdict": "pass", + "reviewer": "reviewer-1", "original_entry_id": entry_id, + "findings": ["misjudged"], "contract_version": VERSION}, + user_id="reviewer-1", ledger_task_id="task-parent") + self.assertEqual(result["correction_action"], "schedule_reaudit") + self.assertTrue(result["merge_blocked"]) + self.assertEqual(result["next_state"].failure_count, 0) + self.assertEqual(result["next_state"].status, "blocked") + self.assertEqual(result["subtask"]["kind"], "reaudit") + # Counting view (0) vs gate view (still blocked) is the + # #247 intended divergence, pinned here, not papered over. + self.assertTrue(self._block()["merge_blocked"]) + # Only a fresh passing verdict unblocks. + self._verdict(pass_verdict()) + self.assertFalse(self._block()["merge_blocked"]) + + def test_correction_to_fail_reblocks_at_once(self): + self._verdict(pass_verdict()) + self.assertFalse(self._block()["merge_blocked"]) + verdict_rows = self.store.conn.execute( + "SELECT id FROM audit_ledger WHERE action = ?" + " ORDER BY id ASC", ("audit_verdict",)).fetchall() + entry_id = verdict_rows[-1][0] + result = wiring.process_correction( + self.conn, self.store, self.task, + {"original_verdict": "pass", "corrected_verdict": "fail", + "reviewer": "reviewer-1", "original_entry_id": entry_id, + "findings": ["missed failure"], + "contract_version": VERSION}, + user_id="reviewer-1", ledger_task_id="task-parent") + self.assertEqual( + result["correction_action"], "retrigger_pushback_and_reblock") + self.assertTrue(result["merge_blocked"]) + self.assertEqual(result["next_state"].failure_count, 1) + self.assertTrue(self._block()["merge_blocked"]) + + def test_acceptance_rebuild_pending_confirm_reject_no_budget(self): + for _ in range(4): + result = self._verdict( + fail_verdict(source="acceptance")) + self.assertEqual(result["action_kind"], + "pending_user_confirmation") + self.assertIsNone(result["subtask"]) + self.assertIsNotNone(result["notification"]) + state = wiring.load_loop_state(self.conn, MODULE, VERSION) + self.assertTrue(state.pending_confirmation) + self.assertEqual(state.rebuild_no, 0) + # System/coordinator markers can never confirm. + with self.assertRaises(ValueError): + wiring.process_rebuild_confirmation( + self.conn, self.store, self.task, + user_confirmed=True, confirmed_by="system", + user_id="user-1", ledger_task_id="task-parent") + with self.assertRaises(ValueError): + wiring.process_rebuild_confirmation( + self.conn, self.store, self.task, + user_confirmed=True, confirmed_by="coordinator", + user_id="user-1", ledger_task_id="task-parent") + # Rejection escalates and consumes no rebuild budget. + rejected = wiring.process_rebuild_confirmation( + self.conn, self.store, self.task, + user_confirmed=False, confirmed_by="user-1", + user_id="user-1", ledger_task_id="task-parent") + self.assertEqual(rejected["action_kind"], "escalate") + state = wiring.load_loop_state(self.conn, MODULE, VERSION) + self.assertEqual(state.rebuild_no, 0) + + def test_acceptance_rebuild_confirm_then_rebuilds(self): + for _ in range(4): + self._verdict(fail_verdict(source="acceptance")) + confirmed = wiring.process_rebuild_confirmation( + self.conn, self.store, self.task, + user_confirmed=True, confirmed_by="user-1", + old_impl_ref=OLD_REF, + user_id="user-1", ledger_task_id="task-parent") + self.assertEqual(confirmed["action_kind"], "rebuild") + self.assertEqual(confirmed["subtask"]["kind"], "rebuild") + state = wiring.load_loop_state(self.conn, MODULE, VERSION) + self.assertEqual(state.rebuild_no, 1) + self.assertEqual(state.status, "rebuilding") + # Confirming again with no pending confirmation fails. + with self.assertRaises(ValueError): + wiring.process_rebuild_confirmation( + self.conn, self.store, self.task, + user_confirmed=True, confirmed_by="user-1", + user_id="user-1", ledger_task_id="task-parent") + + def test_quota_blocked_consumes_nothing(self): + self._verdict(fail_verdict()) + blocked = self._verdict( + fail_verdict(), quota_check=lambda estimate: False) + self.assertEqual(blocked["action_kind"], "quota_blocked") + self.assertFalse(blocked["state_saved"]) + state = wiring.load_loop_state(self.conn, MODULE, VERSION) + self.assertEqual(state.pushback_round, 1) + self.assertEqual(state.failure_count, 1) + + def _raising(estimate): + raise RuntimeError("quota service down") + blocked2 = self._verdict( + fail_verdict(), quota_check=_raising) + self.assertEqual(blocked2["action_kind"], "quota_blocked") + state = wiring.load_loop_state(self.conn, MODULE, VERSION) + self.assertEqual(state.pushback_round, 1) + + def test_quota_adapter_matches_server_rule(self): + check = wiring.build_quota_check(1000000, 999999) + self.assertTrue(check(None)) + self.assertFalse(check(0.01)) + self.assertTrue(wiring.build_quota_check(None, 10 ** 9)(100.0)) + + def test_cost_threshold_escalates_rebuild_with_full_history(self): + for _ in range(3): + self._verdict(fail_verdict()) + result = self._verdict( + fail_verdict(cost_estimate=100.0), cost_threshold=10.0, + old_impl_ref=OLD_REF) + self.assertEqual(result["action_kind"], "escalate") + self.assertEqual(result["notification"]["party"], "user") + self.assertEqual( + len(result["notification"]["escalation_history"]), 4) + self.assertFalse( + result["escalation_grants_user_acceptance"]) + + def test_ledger_failure_fail_closed_state_not_saved(self): + failing = FailingLedgerStore() + with self.assertRaises(RuntimeError): + wiring.process_audit_verdict( + self.conn, failing, self.task, fail_verdict(), + user_id="auditor-1") + self.assertIsNone( + wiring.load_loop_state(self.conn, MODULE, VERSION)) + + def test_automated_reaudit_evidence_missing_fails(self): + result = wiring.evaluate_automated_reaudit(None, {}) + self.assertFalse(result["automated_checks_pass"]) + self.assertIn("hard_gate_evidence_missing", + result["failed_checks"]) + result = wiring.evaluate_automated_reaudit( + HARD_PASS, EXTRA_PASS) + self.assertTrue(result["automated_checks_pass"]) + # Merge-body vocabulary is translated, not rejected: + # branch_protection -> the hard gate's protection. + body_style = dict(HARD_PASS) + body_style["branch_protection"] = body_style.pop("protection") + result = wiring.evaluate_automated_reaudit( + body_style, EXTRA_PASS) + self.assertTrue(result["automated_checks_pass"]) + + def test_release_gate_multi_source_and_read_only(self): + self._verdict(fail_verdict()) + before = self.store.conn.execute( + "SELECT COUNT(*) FROM audit_ledger").fetchone()[0] + decision = wiring.evaluate_release_gate( + "task-parent", self.store.conn, + task_status="completed", hard_gate={"allowed": False, + "blocks": ["ci_not_green"]}) + self.assertFalse(decision["allowed"]) + self.assertEqual(decision["http_status"], 409) + self.assertEqual(decision["blocked_sources"], + ["task_status", "hard_gate", "audit_pushback"]) + self.assertEqual(decision["code"], "release_requires_merged") + after = self.store.conn.execute( + "SELECT COUNT(*) FROM audit_ledger").fetchone()[0] + self.assertEqual(before, after) + self.assertTrue(verify_chain(self.store.conn)["ok"]) + + def test_release_gate_evidence_missing_and_allowed(self): + decision = wiring.evaluate_release_gate( + "task-parent", self.store.conn, + task_status="merged", hard_gate=None) + self.assertEqual(decision["code"], + "release_hard_gate_evidence_missing") + decision = wiring.evaluate_release_gate( + "task-parent", self.store.conn, + task_status="merged", + hard_gate={"allowed": True, "blocks": []}) + self.assertTrue(decision["allowed"]) + self.assertEqual(decision["code"], "release_allowed") + + def test_release_gate_chain_read_failure_fail_closed(self): + closed = sqlite3.connect(":memory:") + closed.close() + decision = wiring.evaluate_release_gate( + "task-parent", closed, task_status="merged", + hard_gate={"allowed": True, "blocks": []}) + self.assertFalse(decision["allowed"]) + self.assertEqual( + decision["code"], "release_blocked_state_unavailable") + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_server_audit_loop.py b/cloud/tests/test_server_audit_loop.py new file mode 100644 index 0000000..9ab5b10 --- /dev/null +++ b/cloud/tests/test_server_audit_loop.py @@ -0,0 +1,457 @@ +"""HTTP end-to-end tests for the audit-loop wiring (track 2). + +Same harness shape as test_server_merge.py: a real server on a +temp DB, no production tasks, no GitHub. Covers the automatic +chain through HTTP (fail -> fix subtask -> merge 409 -> fix +completion -> full re-audit pass -> merge 200 -> release), +plus the release contract, acceptance-confirmation flow, and +the quota / correction boundaries. +""" + +import http.client +import json +import os +import sqlite3 +import sys +import tempfile +import threading +import unittest +from pathlib import Path + +try: + from cloud.auth import AuthStore + from cloud.server import create_server +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore + from cloud.server import create_server + +MODULE = "mod-http" +VERSION = "v1" +EXTRA_PASS = { + "security_unconditional": True, + "regression_exact_artifact": True, + "demo_runnable": True, + "known_deviation_due_todo": True, + "audit_signer_separation": True, + "qa_failed_links_rerun": True, +} + + +def hard_body(): + return { + "files_changed": ["apps/web/a.py"], + "owned_dir": "apps/web", + "branch_protection": "protected", + "ci_status": "success", + "ci_checks": [{"name": "ci", "status": "success"}], + "required_checks": ["ci"], + "contract_version": VERSION, + "contract_frozen": True, + "coordinator_approved": True, + "unresolved_review_comments": 0, + } + + +def fail_body(source="audit"): + return { + "module_id": MODULE, + "contract_version": VERSION, + "verdict": { + "verdict": "fail", + "source": source, + "verdict_text": "check x failed", + "failures": [{"check": "ci_green", "detail": "x"}], + "passed_items": ["a", "b", "c"], + "contract_version": VERSION, + }, + } + + +def pass_body(): + return { + "module_id": MODULE, + "contract_version": VERSION, + "verdict": { + "verdict": "pass", + "source": "audit", + "verdict_text": "all green", + "failures": [], + "passed_items": ["a", "b", "c"], + "contract_version": VERSION, + }, + } + + +class AuditLoopServerTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self._saved_env = { + k: os.environ.get(k) for k in ( + "CLOUD_MERGE_GATE_MODE", "CLOUD_BRANCH_PROTECTION", + "CODEX_GLOBAL_ADMINS", + "CLOUD_AUDIT_LOOP_COST_THRESHOLD_USD")} + os.environ.pop("CLOUD_MERGE_GATE_MODE", None) + os.environ.pop("CLOUD_BRANCH_PROTECTION", None) + os.environ.pop("CLOUD_AUDIT_LOOP_COST_THRESHOLD_USD", None) + self.auth = AuthStore(self.db_path) + self.owner_id, self.owner_key = self.auth.create_user("owner") + self.maint_id, self.maint_key = self.auth.create_user("maint") + self.contrib_id, self.contrib_key = self.auth.create_user( + "contrib") + self.outsider_id, self.outsider_key = self.auth.create_user( + "outsider") + os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id + self.server = create_server(self.db_path, "127.0.0.1", 0) + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + status, self.project = self._req( + "POST", "/projects", {"name": "demo"}, self.owner_key) + self.assertEqual(status, 201) + self.auth.set_member( + self.project["id"], self.owner_id, self.maint_id, + "maintainer") + self.auth.set_member( + self.project["id"], self.owner_id, self.contrib_id, + "contributor") + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + for key, value in self._saved_env.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + try: + self.auth.conn.close() + except Exception: + pass + self.tmp.cleanup() + + def _req(self, method, path, body=None, key=None): + conn = http.client.HTTPConnection( + "127.0.0.1", self.port, timeout=10) + hdrs = {} + payload = None + if body is not None: + payload = json.dumps(body).encode("utf-8") + hdrs["Content-Type"] = "application/json" + if key is not None: + hdrs["X-API-Key"] = key + try: + conn.request(method, path, body=payload, headers=hdrs) + resp = conn.getresponse() + data = resp.read() + finally: + conn.close() + try: + parsed = json.loads(data.decode("utf-8")) if data else None + except ValueError: + parsed = None + return resp.status, parsed + + def _completed_task(self): + status, task = self._req( + "POST", "/tasks", + {"project_id": self.project["id"], "prompt": "x"}, + self.owner_key) + self.assertEqual(status, 201) + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "UPDATE tasks SET status = 'completed' WHERE id = ?", + (task["id"],)) + conn.commit() + finally: + conn.close() + return task + + def _task_status(self, task_id): + status, got = self._req( + "GET", "/tasks/%s" % task_id, key=self.owner_key) + self.assertEqual(status, 200) + return got["status"] + + def _chain_count(self): + conn = sqlite3.connect(self.db_path) + try: + return conn.execute( + "SELECT COUNT(*) FROM audit_ledger").fetchone()[0] + finally: + conn.close() + + def test_full_chain_fail_fix_reaudit_pass_merge_release(self): + task = self._completed_task() + status, first = self._req( + "POST", "/tasks/%s/audit-verdict" % task["id"], + fail_body(), self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(first["action_kind"], "pushback") + self.assertEqual(first["subtask"]["kind"], "fix") + fix_task = first["subtask_task"] + self.assertIsNotNone(fix_task) + self.assertEqual(fix_task["status"], "queued") + # Original task stays completed: fail blocks, it does + # not roll the task back into a re-run. + self.assertEqual(self._task_status(task["id"]), "completed") + # Merge is blocked by the chain pushback even with a + # fully green hard gate. + status, blocked = self._req( + "POST", "/tasks/%s/merge" % task["id"], + hard_body(), self.owner_key) + self.assertEqual(status, 409) + self.assertIn("audit_pushback_blocked", + blocked["merge_gate"]["blocks"]) + # Fix completion on the SUBTASK id, with automated + # evidence + a real auditor re-audit verdict, unblocks + # the PARENT task: chain rows are attributed to the root + # task (forked_from walk), not to the transient subtask. + fix_body = { + "module_id": MODULE, + "contract_version": VERSION, + "fix_result": {"reaudit_verdict": pass_body()["verdict"]}, + "automated": {"hard_gate": hard_body(), + "extra_checks": EXTRA_PASS}, + } + status, fixed = self._req( + "POST", "/tasks/%s/audit-fix" % fix_task["id"], + fix_body, self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(fixed["action_kind"], "pass") + self.assertTrue( + fixed["automated_checks"]["automated_checks_pass"]) + # Loop state is durable and readable back. + status, loop = self._req( + "GET", "/tasks/%s/audit-loop?module_id=%s" + "&contract_version=%s" + % (task["id"], MODULE, VERSION), + key=self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(loop["loop_state"]["status"], "passed") + self.assertFalse(loop["audit_pushback"]["merge_blocked"]) + # Now merge passes, and a merge 200 is never a release + # input: release with no fresh evidence is 409. + status, merged = self._req( + "POST", "/tasks/%s/merge" % task["id"], + hard_body(), self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(merged["status"], "merged") + status, rel = self._req( + "POST", "/tasks/%s/release" % task["id"], + {}, self.owner_key) + self.assertEqual(status, 409) + self.assertEqual(rel["release_gate"]["code"], + "release_hard_gate_evidence_missing") + # Release with fresh evidence passes for a global admin. + status, rel = self._req( + "POST", "/tasks/%s/release" % task["id"], + hard_body(), self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(rel["release_gate"]["code"], "release_allowed") + + def test_release_409_writes_no_audit_and_multi_source(self): + task = self._completed_task() + self._req("POST", "/tasks/%s/audit-verdict" % task["id"], + fail_body(), self.owner_key) + before = self._chain_count() + # Not merged + hard gate failing + pushback blocked: + # all three sources listed, first code is task_status. + bad_hard = hard_body() + bad_hard["ci_status"] = "failure" + status, rel = self._req( + "POST", "/tasks/%s/release" % task["id"], + bad_hard, self.owner_key) + self.assertEqual(status, 409) + self.assertEqual(rel["release_gate"]["blocked_sources"], + ["task_status", "hard_gate", "audit_pushback"]) + self.assertEqual(self._chain_count(), before) + + def test_release_requires_global_admin_write(self): + task = self._completed_task() + # Maintainer with merge rights cannot release: release + # is a global-admin WRITE, not a project permission. + status, _ = self._req( + "POST", "/tasks/%s/release" % task["id"], + hard_body(), self.maint_key) + self.assertEqual(status, 403) + # A global admin who is not a project member CAN release + # (identity is CODEX_GLOBAL_ADMINS only, per #259). + os.environ["CODEX_GLOBAL_ADMINS"] = self.outsider_id + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "UPDATE tasks SET status = 'merged' WHERE id = ?", + (task["id"],)) + conn.commit() + finally: + conn.close() + status, rel = self._req( + "POST", "/tasks/%s/release" % task["id"], + hard_body(), self.outsider_key) + self.assertEqual(status, 200) + self.assertEqual(rel["release_gate"]["code"], "release_allowed") + + def test_acceptance_pending_no_subtask_confirm_flow(self): + task = self._completed_task() + for _ in range(4): + status, result = self._req( + "POST", "/tasks/%s/audit-verdict" % task["id"], + fail_body(source="acceptance"), self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(result["action_kind"], + "pending_user_confirmation") + self.assertIsNone(result["subtask"]) + self.assertIsNone(result["subtask_task"]) + self.assertEqual(result["notification"]["party"], "user") + # A coordinator-style call (confirmed_by != caller) is + # refused: nobody confirms on a user's behalf. + status, _ = self._req( + "POST", "/tasks/%s/audit-rebuild-confirm" % task["id"], + {"module_id": MODULE, "contract_version": VERSION, + "user_confirmed": True, "confirmed_by": "someone-else"}, + self.owner_key) + self.assertEqual(status, 403) + # Missing explicit flag is refused, not defaulted. + status, _ = self._req( + "POST", "/tasks/%s/audit-rebuild-confirm" % task["id"], + {"module_id": MODULE, "contract_version": VERSION, + "confirmed_by": self.owner_id}, + self.owner_key) + self.assertEqual(status, 400) + # Real user confirmation creates the rebuild subtask. + status, confirmed = self._req( + "POST", "/tasks/%s/audit-rebuild-confirm" % task["id"], + {"module_id": MODULE, "contract_version": VERSION, + "user_confirmed": True, + "confirmed_by": self.owner_id, + "old_impl_ref": {"archive_ref": "task:%s:pre-rebuild" + % task["id"]}}, + self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(confirmed["action_kind"], "rebuild") + self.assertEqual(confirmed["subtask"]["kind"], "rebuild") + self.assertIsNotNone(confirmed["subtask_task"]) + + def test_exactly_three_pushback_fourth_rebuild_http(self): + task = self._completed_task() + for index in range(3): + status, result = self._req( + "POST", "/tasks/%s/audit-verdict" % task["id"], + fail_body(), self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(result["action_kind"], "pushback") + body = fail_body() + body["old_impl_ref"] = {"archive_ref": "task:%s:pre-rebuild" + % task["id"]} + status, result = self._req( + "POST", "/tasks/%s/audit-verdict" % task["id"], + body, self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(result["action_kind"], "rebuild") + self.assertEqual(result["loop_state"]["rebuild_no"], 1) + + def test_correction_to_pass_still_blocks_until_fresh_pass(self): + task = self._completed_task() + self._req("POST", "/tasks/%s/audit-verdict" % task["id"], + fail_body(), self.owner_key) + conn = sqlite3.connect(self.db_path) + try: + entry_id = conn.execute( + "SELECT id FROM audit_ledger WHERE action = ?" + " ORDER BY id ASC", ("audit_verdict",)).fetchone()[0] + finally: + conn.close() + status, corrected = self._req( + "POST", "/tasks/%s/audit-correction" % task["id"], + {"module_id": MODULE, "contract_version": VERSION, + "correction": { + "original_verdict": "fail", + "corrected_verdict": "pass", + "reviewer": "reviewer-1", + "original_entry_id": entry_id, + "findings": ["misjudged"]}}, + self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(corrected["correction_action"], + "schedule_reaudit") + self.assertTrue(corrected["merge_blocked"]) + status, _ = self._req( + "POST", "/tasks/%s/merge" % task["id"], + hard_body(), self.owner_key) + self.assertEqual(status, 409) + # A fresh passing verdict (not the correction) unblocks. + status, _ = self._req( + "POST", "/tasks/%s/audit-verdict" % task["id"], + pass_body(), self.owner_key) + self.assertEqual(status, 200) + status, merged = self._req( + "POST", "/tasks/%s/merge" % task["id"], + hard_body(), self.owner_key) + self.assertEqual(status, 200) + + def test_fix_without_reaudit_is_refused_and_automation_not_pass(self): + task = self._completed_task() + status, first = self._req( + "POST", "/tasks/%s/audit-verdict" % task["id"], + fail_body(), self.owner_key) + self.assertEqual(status, 200) + fix_task = first["subtask_task"] + # No reaudit verdict at all: 400, loop still blocked. + status, _ = self._req( + "POST", "/tasks/%s/audit-fix" % fix_task["id"], + {"module_id": MODULE, "contract_version": VERSION, + "fix_result": {"fix_diff": {"summary": "changed"}}}, + self.owner_key) + self.assertEqual(status, 400) + # Automated checks all pass but the auditor verdict is a + # fail: the loop stays blocked (automation != audit). + status, fixed = self._req( + "POST", "/tasks/%s/audit-fix" % fix_task["id"], + {"module_id": MODULE, "contract_version": VERSION, + "fix_result": {"reaudit_verdict": + fail_body()["verdict"]}, + "automated": {"hard_gate": hard_body(), + "extra_checks": EXTRA_PASS}}, + self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(fixed["action_kind"], "pushback") + self.assertTrue(fixed["merge_blocked"]) + + def test_quota_zero_blocks_without_consuming_round(self): + task = self._completed_task() + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "UPDATE user_quotas SET quota_usd_micros = 0" + " WHERE user_id = ?", (self.owner_id,)) + conn.commit() + finally: + conn.close() + status, result = self._req( + "POST", "/tasks/%s/audit-verdict" % task["id"], + fail_body(), self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(result["action_kind"], "quota_blocked") + # No loop state was saved: nothing was consumed. + status, _ = self._req( + "GET", "/tasks/%s/audit-loop?module_id=%s" + "&contract_version=%s" + % (task["id"], MODULE, VERSION), + key=self.owner_key) + self.assertEqual(status, 404) + + def test_verdict_forbidden_for_contributor(self): + task = self._completed_task() + status, _ = self._req( + "POST", "/tasks/%s/audit-verdict" % task["id"], + fail_body(), self.contrib_key) + self.assertEqual(status, 403) + + +if __name__ == "__main__": + unittest.main()