From 9aed328d02e97f6b5dde561d443c47bdf6fb2e98 Mon Sep 17 00:00:00 2001 From: LZH-YS1998 Date: Tue, 21 Jul 2026 11:18:12 +0800 Subject: [PATCH] fix(company): add dispatch attempt ledger to brake infinite re-dispatch loops (#10) Work items whose execution kept dying without a durable verdict (crash mid-dispatch, kill -9, cancel-before-harvest) were re-dispatched forever: every exit path was responsible for remembering to write a terminal phase, and any path that forgot left the card RUNNING and eligible again. Replace that with structural accounting: - claim CAS opens an attempt in the same UPDATE (attempt_seq+1, attempt_settled=false) so no dispatch can start unaccounted (store.py) - transition_work_item becomes the settlement authority: every non-RUNNING transition settles the open attempt in the same write; crashed/interrupted outcomes accumulate streaks, clean outcomes reset them; claim release folds into the same write; settlement still lands when the phase write loses a race (work_item_transition.py) - dispatcher refuses cards over the streak limits (crash>=3, interrupted>=5) in both is_dispatchable and _work_item_is_runnable, and a per-tick reconcile pass back-fills dead attempts as interrupted and terminalizes over-limit cards to FAILED with a visible blocked_reason (dispatch_hold quarantine if even that write fails) (phase.py, company_mode.py) - crash exits now settle: cancellation unwind harvests coroutines that died on a real exception before discarding them, the crashed-item handler releases the claim and settles as crashed with a quarantine fallback, and the timeout path settles as crashed (company_mode.py) Co-Authored-By: Claude Fable 5 --- opc/database/store.py | 10 +- opc/layer2_organization/company_mode.py | 214 ++++++- opc/layer2_organization/metadata_ownership.py | 11 + opc/layer2_organization/phase.py | 53 ++ .../work_item_transition.py | 151 ++++- tests/test_attempt_ledger.py | 555 ++++++++++++++++++ 6 files changed, 968 insertions(+), 26 deletions(-) create mode 100644 tests/test_attempt_ledger.py diff --git a/opc/database/store.py b/opc/database/store.py index 6c4fe1a..1de9d22 100644 --- a/opc/database/store.py +++ b/opc/database/store.py @@ -5497,7 +5497,14 @@ class OPCStore: COALESCE(NULLIF(metadata, ''), '{}'), '$.claimed_by_role_session_id', ?, '$.claimed_task_id', ?, - '$.claimed_work_item_revision', ? + '$.claimed_work_item_revision', ?, + '$.attempt_seq', COALESCE( + CAST(json_extract(metadata, '$.attempt_seq') AS INTEGER), + 0 + ) + 1, + '$.attempt_settled', json('false'), + '$.attempt_outcome', '', + '$.attempt_started_at', ? ), updated_at = ? WHERE work_item_id = ? @@ -5521,6 +5528,7 @@ class OPCStore: claimed_task_id, int(work_item_revision or 0), updated_at.isoformat(), + updated_at.isoformat(), str(work_item_id or "").strip(), phase.value, int(work_item_revision or 0), diff --git a/opc/layer2_organization/company_mode.py b/opc/layer2_organization/company_mode.py index 5237e05..4dc5d67 100644 --- a/opc/layer2_organization/company_mode.py +++ b/opc/layer2_organization/company_mode.py @@ -51,6 +51,9 @@ from opc.layer2_organization.phase import ( IN_PROGRESS_PHASES, IN_REVIEW_PHASES, TODO_PHASES, + InvalidPhaseTransition, + attempt_ledger_dispatch_block_reason, + has_open_attempt, is_dispatchable, is_orphaned, is_report_execution_work_item_metadata, @@ -111,7 +114,9 @@ from opc.layer2_organization.work_item_transition import ( has_pending_settlement_release, normalize_dependency_work_item_ids, refresh_dependents_for_run, + settle_open_attempt_as_interrupted, settled_failure_dependency_ids, + transition_work_item, transition_work_item_from_task, ) from opc.layer2_organization.work_item_identity import ( @@ -2671,6 +2676,13 @@ class CompanyWorkItemExecutor: report_execution_work_item = is_report_execution_work_item_metadata(metadata) if str(metadata.get("dispatch_hold", "") or "").strip(): return False + # Attempt-ledger brake: cards whose recent attempts keep crashing or + # keep getting interrupted must not be re-enqueued — this check also + # covers the review/report `or phase == RUNNING` arms below that + # bypass is_dispatchable. The per-tick ledger reconcile pass + # terminalizes such cards with a visible blocked_reason. + if attempt_ledger_dispatch_block_reason(metadata): + return False # Hidden auxiliary cards (review / report) are still runnable — # they are the kanban-push primitives the dispatcher schedules. # Worker work items marked hidden for any other reason stay @@ -2828,6 +2840,124 @@ class CompanyWorkItemExecutor: return True return True + async def _reconcile_attempt_ledger( + self, + work_items: list[DelegationWorkItem], + active_work_item_tasks: dict[ + "asyncio.Task[TaskResult | None]", + tuple["CompanyMemberSession", Task], + ], + ) -> list[DelegationWorkItem]: + """Per-tick attempt-ledger reconcile: the structural anti-loop pass. + + Two responsibilities, both driven by durable state instead of any + code path "remembering" to behave: + + 1. **Back-fill dead attempts.** A card whose claim opened an attempt + (``attempt_settled == False``) but has no live owner in THIS + dispatcher (not in ``active_work_item_tasks`` / claimed sets) had + its coroutine or process die without a verdict — kill -9, cancel + before harvest, or a settlement write that never landed. Settle it + as ``interrupted`` so the ledger sees it. + 2. **Terminalize over-limit cards.** When the ledger shows a run of + consecutive crashed/interrupted attempts over the limit, transition + the card to FAILED with a visible ``blocked_reason`` (quarantine + via ``dispatch_hold`` if even that write fails). Eligibility checks + (``is_dispatchable`` / ``_work_item_is_runnable``) independently + refuse such cards, so either write path converges the loop. + """ + if not work_items or self.store is None: + return work_items + in_flight_ids: set[str] = { + linked_work_item_id_for_task(claimed_task) + for _member, claimed_task in active_work_item_tasks.values() + } + in_flight_ids.discard("") + in_flight_ids.update( + str(item) for item in getattr(self.runtime, "_claimed_work_item_ids", set()) or set() + ) + reconciled: list[DelegationWorkItem] = [] + for work_item in work_items: + phase = getattr(work_item, "phase", None) + work_item_id = str(getattr(work_item, "work_item_id", "") or "").strip() + if not work_item_id or phase in DONE_PHASES: + reconciled.append(work_item) + continue + metadata = dict(getattr(work_item, "metadata", {}) or {}) + if has_open_attempt(metadata) and work_item_id not in in_flight_ids: + if await settle_open_attempt_as_interrupted(self.store, work_item): + metadata = dict(getattr(work_item, "metadata", {}) or {}) + logger.info( + "[attempt_ledger] settled dead attempt as interrupted " + "work_item={} phase={} interrupted_streak={}", + work_item_id, + getattr(phase, "value", phase), + metadata.get("attempt_interrupted_streak"), + ) + block_reason = attempt_ledger_dispatch_block_reason(metadata) + if not block_reason: + reconciled.append(work_item) + continue + summary_text = ( + f"Work item quarantined by the attempt ledger: {block_reason}. " + "Its recent execution attempts kept dying without a durable verdict; " + "manual triage (rework or new card) is required." + ) + try: + updated = await transition_work_item( + self.store, + work_item_id, + target_phase=Phase.FAILED, + reason="attempt_ledger_limit", + summary=summary_text, + metadata_updates={"attempt_ledger_block_reason": block_reason}, + release_claim=True, + ) + if updated is not None: + work_item = updated + try: + await self.store.update_delegation_work_item( + work_item_id, + blocked_reason=block_reason, + ) + except Exception: + logger.opt(exception=True).debug( + "[attempt_ledger] blocked_reason write failed for {}", work_item_id + ) + await self._emit_progress( + f"[Company:{projection_id_for_work_item(work_item)}] {summary_text}" + ) + except InvalidPhaseTransition: + # A concurrent writer moved the card; re-read and keep going — + # eligibility checks still refuse it while the ledger is over + # the limit. + logger.opt(exception=True).debug( + "[attempt_ledger] FAILED terminalize lost a phase race for {}", + work_item_id, + ) + except Exception: + logger.opt(exception=True).warning( + "[attempt_ledger] FAILED terminalize did not land for {}; " + "quarantining via dispatch_hold", + work_item_id, + ) + try: + await self.store.update_delegation_work_item( + work_item_id, + blocked_reason=block_reason, + metadata_updates={ + "dispatch_hold": "attempt_ledger_quarantine", + "attempt_ledger_block_reason": block_reason, + }, + ) + except Exception: + logger.opt(exception=True).error( + "[attempt_ledger] quarantine hold write also failed for {}", + work_item_id, + ) + reconciled.append(work_item) + return reconciled + def _task_effective_projection_spec(self, task: Task) -> WorkItemProjectionSpec: projection = self._projection_spec_for_task(task) if projection is not None: @@ -4528,6 +4658,10 @@ class CompanyWorkItemExecutor: await self._try_unpark_blocking_comms(parked) await self.runtime.refresh_inbox_state(tasks) work_items = await self._load_delegation_work_items(tasks) + work_items = await self._reconcile_attempt_ledger( + work_items, + active_work_item_tasks, + ) work_items = await self._refresh_ready_work_items(work_items, tasks=tasks) tasks = await self._materialize_work_item_tasks(tasks, work_items) self._active_tasks = tasks @@ -4690,6 +4824,42 @@ class CompanyWorkItemExecutor: self._schedule_kanban_notification() except asyncio.CancelledError: claimed_pairs = list(active_work_item_tasks.values()) + # Coroutines that already died on a real exception must settle + # their attempt BEFORE this turn unwinds. The old gather(..., + # return_exceptions=True) below silently discarded them, leaving + # the card RUNNING with an open attempt — the exact gap that let + # a deterministic crash replay forever across suspend/resume + # cycles (issue #10). Cancellation of live coroutines is still a + # legitimate suspend; only genuine crashes are harvested here. + crashed_pairs: list[tuple[CompanyMemberSession, Task, BaseException]] = [] + for work_item_task, session_task in list(active_work_item_tasks.items()): + if not work_item_task.done() or work_item_task.cancelled(): + continue + crash_exc = work_item_task.exception() + if crash_exc is None or isinstance(crash_exc, asyncio.CancelledError): + continue + crashed_member_session, crashed_task = session_task + crashed_pairs.append((crashed_member_session, crashed_task, crash_exc)) + for crashed_member_session, crashed_task, crash_exc in crashed_pairs: + try: + await asyncio.shield( + self._handle_claimed_work_item_exception( + crashed_member_session, + crashed_task, + crash_exc, + ) + ) + except asyncio.CancelledError: + logger.warning( + "company runtime cancellation: crashed work item settlement " + "for task={} continues shielded in background", + crashed_task.id, + ) + except Exception: + logger.opt(exception=True).error( + "company runtime cancellation: failed to settle crashed work item task={}", + crashed_task.id, + ) for work_item_task in list(active_work_item_tasks.keys()): if not work_item_task.done(): work_item_task.cancel() @@ -4957,13 +5127,42 @@ class CompanyWorkItemExecutor: "artifacts": dict(failure_result.artifacts or {}), } # Phase A: phase write first → hook projects task.status=FAILED - # onto the DB row and syncs our local task.status too. - await transition_work_item_from_task( - self.store, task, - target_status_or_phase=Phase.FAILED, - reason="claimed_work_item_exception", - summary=summary or None, - ) + # onto the DB row and syncs our local task.status too. The + # attempt_outcome="crashed" settlement rides the same write so + # the dispatcher's crash-streak brake has durable accounting. + # If even this write fails, quarantine the card via a plain + # metadata hold — dispatch_hold blocks is_dispatchable without + # needing phase-machine legality — so a store hiccup can never + # convert a crash into an endless re-dispatch loop. + try: + await transition_work_item_from_task( + self.store, task, + target_status_or_phase=Phase.FAILED, + reason="claimed_work_item_exception", + summary=summary or None, + release_claim=True, + attempt_outcome="crashed", + ) + except Exception as transition_exc: + logger.opt(exception=transition_exc).error( + f"[Company:{projection_id}] FAILED transition for crashed work item " + "did not land; quarantining via dispatch_hold" + ) + if work_item_id and self.store is not None and hasattr(self.store, "update_delegation_work_item"): + try: + await self.store.update_delegation_work_item( + work_item_id, + blocked_reason=summary, + metadata_updates={ + "dispatch_hold": "crash_quarantine", + "crash_quarantine_at": datetime.now().isoformat(), + "crash_quarantine_error": str(exc), + }, + ) + except Exception: + logger.opt(exception=True).error( + f"[Company:{projection_id}] crash quarantine write also failed" + ) try: await self.runtime.complete_claim(member_session, task, result=failure_result) except Exception as cleanup_exc: @@ -5428,6 +5627,7 @@ class CompanyWorkItemExecutor: self.store, task, target_status_or_phase=Phase.FAILED, reason="work_item_timeout", + attempt_outcome="crashed", ) await self.save_task(task) return TaskResult(status=TaskStatus.FAILED, content=f"Work item timed out after {self.work_item_timeout}s.") diff --git a/opc/layer2_organization/metadata_ownership.py b/opc/layer2_organization/metadata_ownership.py index 65b7ce1..b5ef7b6 100644 --- a/opc/layer2_organization/metadata_ownership.py +++ b/opc/layer2_organization/metadata_ownership.py @@ -187,6 +187,17 @@ _WORK_ITEM_FIELDS: tuple[MetadataFieldSpec, ...] = ( _spec("self_evolution_patch", MetadataOwner.WORK_ITEM, allowed_locations=("work_item",), migration_policy="work_item_wins"), _spec("self_evolution_completed_at", MetadataOwner.WORK_ITEM, allowed_locations=("work_item",), migration_policy="work_item_wins"), _spec("self_evolution_error", MetadataOwner.WORK_ITEM, allowed_locations=("work_item",), migration_policy="work_item_wins"), + # Attempt ledger (dispatch-attempt accounting; see phase.py). Opened by the + # claim CAS, settled by transition_work_item — the dispatcher's structural + # brake against crash/interrupted re-dispatch loops. + _spec("attempt_seq", MetadataOwner.WORK_ITEM, allowed_locations=("work_item",), migration_policy="work_item_wins"), + _spec("attempt_settled", MetadataOwner.WORK_ITEM, allowed_locations=("work_item",), migration_policy="work_item_wins"), + _spec("attempt_outcome", MetadataOwner.WORK_ITEM, allowed_locations=("work_item",), migration_policy="work_item_wins"), + _spec("attempt_started_at", MetadataOwner.WORK_ITEM, allowed_locations=("work_item",), migration_policy="work_item_wins"), + _spec("attempt_settled_at", MetadataOwner.WORK_ITEM, allowed_locations=("work_item",), migration_policy="work_item_wins"), + _spec("attempt_crash_streak", MetadataOwner.WORK_ITEM, allowed_locations=("work_item",), migration_policy="work_item_wins"), + _spec("attempt_interrupted_streak", MetadataOwner.WORK_ITEM, allowed_locations=("work_item",), migration_policy="work_item_wins"), + _spec("attempt_ledger_block_reason", MetadataOwner.WORK_ITEM, allowed_locations=("work_item",), migration_policy="work_item_wins"), ) _RUNTIME_TASK_FIELDS: tuple[MetadataFieldSpec, ...] = ( diff --git a/opc/layer2_organization/phase.py b/opc/layer2_organization/phase.py index ab47779..b48c270 100644 --- a/opc/layer2_organization/phase.py +++ b/opc/layer2_organization/phase.py @@ -385,6 +385,57 @@ def is_orphaned(item: Any) -> bool: return not claim +# ── Attempt ledger ─────────────────────────────────────────────────────── +# +# Every durable claim opens an "attempt" on the work item (stamped by the +# claim CAS in store.claim_delegation_work_item_if_dispatchable) and every +# turn-boundary transition through work_item_transition.transition_work_item +# settles it with an outcome. The dispatcher refuses to start attempt N+1 +# for cards whose ledger shows a run of consecutive crashed / interrupted +# attempts — this is the structural brake that makes "a crash path forgot +# (or failed) to write FAILED" converge instead of re-dispatching forever +# (issue #10: restart → resume → deterministic crash → re-dispatch loop). + +ATTEMPT_CRASH_STREAK_LIMIT = 3 +ATTEMPT_INTERRUPTED_STREAK_LIMIT = 5 + + +def _attempt_ledger_int(metadata: Mapping[str, Any], key: str) -> int: + try: + return int(metadata.get(key, 0) or 0) + except (TypeError, ValueError): + return 0 + + +def attempt_ledger_dispatch_block_reason(metadata: Mapping[str, Any] | None) -> str: + """Non-empty human-readable reason when the attempt ledger forbids + dispatching this card again; empty string when dispatch is allowed.""" + if not isinstance(metadata, Mapping): + return "" + crash_streak = _attempt_ledger_int(metadata, "attempt_crash_streak") + if crash_streak >= ATTEMPT_CRASH_STREAK_LIMIT: + return ( + f"attempt ledger: {crash_streak} consecutive crashed attempts " + f"(limit {ATTEMPT_CRASH_STREAK_LIMIT})" + ) + interrupted_streak = _attempt_ledger_int(metadata, "attempt_interrupted_streak") + if interrupted_streak >= ATTEMPT_INTERRUPTED_STREAK_LIMIT: + return ( + f"attempt ledger: {interrupted_streak} consecutive interrupted attempts " + f"(limit {ATTEMPT_INTERRUPTED_STREAK_LIMIT})" + ) + return "" + + +def has_open_attempt(metadata: Mapping[str, Any] | None) -> bool: + """True when a claim opened an attempt that was never settled.""" + if not isinstance(metadata, Mapping): + return False + if _attempt_ledger_int(metadata, "attempt_seq") <= 0: + return False + return not bool(metadata.get("attempt_settled", True)) + + def is_dispatchable(item: Any) -> bool: """Combined check used by the dispatcher: pick this card on next tick? @@ -405,6 +456,8 @@ def is_dispatchable(item: Any) -> bool: return False if isinstance(metadata, dict) and str(metadata.get("dispatch_hold", "") or "").strip(): return False + if attempt_ledger_dispatch_block_reason(metadata): + return False return is_runnable(item.phase) or is_orphaned(item) diff --git a/opc/layer2_organization/work_item_transition.py b/opc/layer2_organization/work_item_transition.py index 7689fc0..1a5ee89 100644 --- a/opc/layer2_organization/work_item_transition.py +++ b/opc/layer2_organization/work_item_transition.py @@ -34,6 +34,7 @@ from opc.layer2_organization.phase import ( DONE_PHASES, InvalidPhaseTransition, coerce_phase, + has_open_attempt, phase_for_task_status, task_status_for_phase, validate_transition, @@ -43,6 +44,54 @@ from opc.layer2_organization.work_item_identity import work_item_identity_payloa from opc.layer2_organization.work_item_runtime import is_work_item_runtime_metadata +def build_attempt_settlement_updates( + current_metadata: dict[str, Any] | None, + *, + outcome: str, +) -> dict[str, Any]: + """Compute the metadata delta that settles the currently-open attempt. + + ``outcome`` semantics: + * ``"crashed"`` — the attempt died on an exception/timeout; increments + ``attempt_crash_streak`` (consecutive crashes). + * ``"interrupted"`` — the owning process/coroutine went away without a + verdict (kill, cancel-before-harvest); increments + ``attempt_interrupted_streak``. + * anything else — a clean turn boundary (approved/failed-by-review/ + waiting/park/...); resets both streaks. + + Returns ``{}`` when there is no open attempt to settle, so callers can + merge unconditionally. + """ + metadata = dict(current_metadata or {}) + if not has_open_attempt(metadata): + return {} + + def _streak(key: str) -> int: + try: + return int(metadata.get(key, 0) or 0) + except (TypeError, ValueError): + return 0 + + outcome_clean = str(outcome or "").strip() or "settled" + if outcome_clean == "crashed": + crash_streak = _streak("attempt_crash_streak") + 1 + interrupted_streak = _streak("attempt_interrupted_streak") + elif outcome_clean == "interrupted": + crash_streak = _streak("attempt_crash_streak") + interrupted_streak = _streak("attempt_interrupted_streak") + 1 + else: + crash_streak = 0 + interrupted_streak = 0 + return { + "attempt_settled": True, + "attempt_outcome": outcome_clean, + "attempt_settled_at": datetime.now().isoformat(), + "attempt_crash_streak": crash_streak, + "attempt_interrupted_streak": interrupted_streak, + } + + async def transition_work_item( store: Any, work_item_id: str, @@ -52,6 +101,7 @@ async def transition_work_item( summary: str | None = None, metadata_updates: dict[str, Any] | None = None, release_claim: bool = False, + attempt_outcome: str | None = None, ) -> DelegationWorkItem | None: """Transition a work item to ``target_phase``. @@ -89,41 +139,104 @@ async def transition_work_item( reason_clean = str(reason or "").strip() if reason_clean: merged["last_transition_reason"] = reason_clean + # Attempt settlement: any transition to a non-RUNNING phase is a turn + # boundary for the claim that opened the current attempt (the claim CAS + # stamped attempt_settled=false). Fold the settlement delta into the SAME + # write as the phase change so an attempt can never end without its + # ledger entry — the structural guarantee behind the dispatcher's + # crash/interrupted streak brake (see phase.attempt_ledger_dispatch_ + # block_reason). + settlement: dict[str, Any] = {} + current_item = None + if phase != Phase.RUNNING and hasattr(store, "get_delegation_work_item"): + try: + current_item = await store.get_delegation_work_item(work_item_id) + except Exception: + current_item = None + if current_item is not None: + settlement = build_attempt_settlement_updates( + dict(getattr(current_item, "metadata", {}) or {}), + outcome=attempt_outcome or phase.value, + ) + if settlement: + merged = {**settlement, **merged} kwargs: dict[str, Any] = { "phase": phase, "metadata_updates": merged, } if summary is not None: kwargs["summary"] = summary - # Phase transition first, claim release second (when requested). The - # old rationale for this ordering was "sync_member_session_hook reads - # item.claimed_by_role_runtime_session_id"; Phase B removed that hook - # and moved the unpark to the dispatcher's per-tick rehydrate pass, - # but the two-step ordering is kept so downstream listeners that - # still inspect the claim (audit logs, kanban projections) see a - # consistent before/after. + if release_claim: + # Fold claim release into the same write as the phase change: the + # legacy two-call sequence could commit the phase and then fail the + # release, stranding a dead claim. Terminal phases additionally drop + # the metadata claim mirror keys so the row can never satisfy a + # future claim-CAS predicate by accident. + kwargs["claimed_by_role_runtime_session_id"] = "" + kwargs["claimed_by_seat_id"] = "" + if phase in DONE_PHASES: + merged.setdefault("claimed_by_role_session_id", "") + merged.setdefault("claimed_task_id", "") try: result = await store.update_delegation_work_item(work_item_id, **kwargs) + except InvalidPhaseTransition: + # The phase write lost a race (or the caller is out of date). The + # attempt settlement must still land — losing it is exactly the + # "crash without accounting" gap that produced endless re-dispatch. + if settlement: + try: + await store.update_delegation_work_item( + work_item_id, + metadata_updates=settlement, + ) + except Exception: + logger.opt(exception=True).warning( + f"transition_work_item: settlement fallback failed wid={work_item_id}" + ) + raise except Exception: logger.opt(exception=True).warning( f"transition_work_item failed wid={work_item_id} " f"target={phase.value} reason={reason_clean}" ) raise - if release_claim and result is not None: - try: - result = await store.update_delegation_work_item( - work_item_id, - claimed_by_role_runtime_session_id="", - claimed_by_seat_id="", - ) - except Exception: - logger.opt(exception=True).warning( - f"transition_work_item: claim release failed wid={work_item_id}" - ) return result +async def settle_open_attempt_as_interrupted( + store: Any, + work_item: DelegationWorkItem, +) -> bool: + """Back-fill the ledger for an attempt whose owner died without settling. + + Metadata-only write (no phase change): the card keeps whatever phase the + crash left it in — recovery semantics stay untouched — but the attempt + stops being invisible. Called by the dispatcher's per-tick reconcile pass + for cards with an open attempt and no live in-process owner. Returns True + when a settlement write was issued. + """ + if store is None or not hasattr(store, "update_delegation_work_item"): + return False + metadata = dict(getattr(work_item, "metadata", {}) or {}) + settlement = build_attempt_settlement_updates(metadata, outcome="interrupted") + if not settlement: + return False + try: + updated = await store.update_delegation_work_item( + work_item.work_item_id, + metadata_updates=settlement, + ) + except Exception: + logger.opt(exception=True).warning( + "settle_open_attempt_as_interrupted failed " + f"wid={getattr(work_item, 'work_item_id', '')}" + ) + return False + if updated is not None: + work_item.metadata = dict(updated.metadata or {}) + return True + + def _fallback_status_for( target_status_or_phase: TaskStatus | Phase | str, task: Task, @@ -162,6 +275,7 @@ async def transition_work_item_from_task( metadata_updates: dict[str, Any] | None = None, release_claim: bool = False, require_work_item: bool = False, + attempt_outcome: str | None = None, ) -> bool: """Task bridge helper: transition a work item when the caller holds a Task. @@ -306,6 +420,7 @@ async def transition_work_item_from_task( summary=summary, metadata_updates=back_ref, release_claim=release_claim, + attempt_outcome=attempt_outcome, ) except InvalidPhaseTransition: # Defensive: state-machine validation at the store layer can also diff --git a/tests/test_attempt_ledger.py b/tests/test_attempt_ledger.py new file mode 100644 index 0000000..e64a56b --- /dev/null +++ b/tests/test_attempt_ledger.py @@ -0,0 +1,555 @@ +"""Attempt-ledger settlement tests (issue #10 root fix). + +Covers the structural anti-loop mechanism: +- the claim CAS opens a durable attempt (attempt_seq / attempt_settled) +- transition_work_item settles the attempt on every turn-boundary phase write +- crashed / interrupted streaks accumulate and the dispatcher refuses + over-limit cards (is_dispatchable + _work_item_is_runnable) +- the per-tick reconcile pass back-fills dead attempts and terminalizes + over-limit cards with a visible blocked_reason +- resume availability gate: a work item pinned to a disabled external agent + fails closed at resume-prep instead of crash-looping through dispatch +""" +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock + +from opc.core.config import OPCConfig, RoleConfig +from opc.core.events import EventBus +from opc.core.models import ( + DelegationRoleSession, + DelegationWorkItem, + Phase, + Task, + TaskStatus, +) +from opc.database.store import OPCStore +from opc.engine import OPCEngine +from opc.layer2_organization.communication import CommunicationManager +from opc.layer2_organization.company_mode import ( + CompanyWorkItemExecutor, + serialize_company_work_item_runtime_plan, +) +from opc.layer2_organization.org_engine import OrgEngine +from opc.layer2_organization.org_work_item_planner import ( + CompanyWorkItemRuntimePlan, + WorkItemProjectionSpec, +) +from opc.layer2_organization.phase import ( + ATTEMPT_CRASH_STREAK_LIMIT, + ATTEMPT_INTERRUPTED_STREAK_LIMIT, + attempt_ledger_dispatch_block_reason, + has_open_attempt, + is_dispatchable, +) +from opc.layer2_organization.work_item_links import set_linked_work_item_id +from opc.layer2_organization.work_item_transition import ( + settle_open_attempt_as_interrupted, + transition_work_item, +) + + +class AttemptLedgerStoreTests(unittest.IsolatedAsyncioTestCase): + async def _store(self) -> OPCStore: + tmpdir = tempfile.TemporaryDirectory() + self.addCleanup(tmpdir.cleanup) + store = OPCStore(Path(tmpdir.name) / "tasks.db") + await store.initialize() + self.addAsyncCleanup(store.close) + return store + + async def _seed_item( + self, + store: OPCStore, + *, + work_item_id: str = "wi-1", + phase: Phase = Phase.READY, + metadata: dict | None = None, + ) -> DelegationWorkItem: + item = DelegationWorkItem( + work_item_id=work_item_id, + run_id="run-1", + role_id="executor", + seat_id="seat-1", + title="Execution", + summary="Do the work.", + kind="execute", + projection_id="execution", + phase=phase, + metadata=dict(metadata or {}), + ) + await store.save_delegation_work_item(item) + return item + + async def _claim(self, store: OPCStore, work_item_id: str, phase: Phase) -> DelegationWorkItem: + claimed = await store.claim_delegation_work_item_if_dispatchable( + work_item_id, + expected_phase=phase, + role_runtime_session_id="role-sess-1", + seat_id="seat-1", + task_id="task-1", + ) + assert claimed is not None, "claim CAS unexpectedly failed" + return claimed + + async def test_claim_cas_opens_attempt(self) -> None: + store = await self._store() + await self._seed_item(store) + claimed = await self._claim(store, "wi-1", Phase.READY) + metadata = dict(claimed.metadata or {}) + self.assertEqual(int(metadata.get("attempt_seq")), 1) + self.assertFalse(bool(metadata.get("attempt_settled"))) + self.assertTrue(str(metadata.get("attempt_started_at", "")).strip()) + self.assertTrue(has_open_attempt(metadata)) + + # Settle + release, then re-claim the orphaned RUNNING card: seq += 1. + await transition_work_item( + store, + "wi-1", + target_phase=Phase.READY, + reason="recovery", + release_claim=True, + ) + settled = await store.get_delegation_work_item("wi-1") + assert settled is not None + self.assertTrue(bool(settled.metadata.get("attempt_settled"))) + # transition folded the claim clear into the same write, but the claim + # mirror metadata keys are only dropped on terminal phases — clear them + # the way the resume/suspend paths do before re-claiming. + await store.update_delegation_work_item( + "wi-1", + metadata_updates={"claimed_by_role_session_id": "", "claimed_task_id": ""}, + ) + reclaimed = await self._claim(store, "wi-1", Phase.READY) + self.assertEqual(int(reclaimed.metadata.get("attempt_seq")), 2) + self.assertFalse(bool(reclaimed.metadata.get("attempt_settled"))) + + async def test_transition_settles_attempt_clean(self) -> None: + store = await self._store() + await self._seed_item(store) + await self._claim(store, "wi-1", Phase.READY) + await transition_work_item( + store, + "wi-1", + target_phase=Phase.AWAITING_MANAGER_REVIEW, + reason="turn_done", + ) + item = await store.get_delegation_work_item("wi-1") + assert item is not None + metadata = dict(item.metadata or {}) + self.assertTrue(bool(metadata.get("attempt_settled"))) + self.assertEqual(metadata.get("attempt_outcome"), Phase.AWAITING_MANAGER_REVIEW.value) + self.assertEqual(int(metadata.get("attempt_crash_streak")), 0) + self.assertEqual(int(metadata.get("attempt_interrupted_streak")), 0) + self.assertFalse(has_open_attempt(metadata)) + + async def test_crash_streak_accumulates_and_blocks_dispatch(self) -> None: + store = await self._store() + await self._seed_item(store) + phase = Phase.READY + for round_index in range(ATTEMPT_CRASH_STREAK_LIMIT): + await self._claim(store, "wi-1", phase) + # Crash + recovery-exit back to READY (RUNNING → READY is the + # legal crash-recovery edge) with outcome=crashed. + await transition_work_item( + store, + "wi-1", + target_phase=Phase.READY, + reason="crash_recovery", + release_claim=True, + attempt_outcome="crashed", + ) + await store.update_delegation_work_item( + "wi-1", + metadata_updates={"claimed_by_role_session_id": "", "claimed_task_id": ""}, + ) + item = await store.get_delegation_work_item("wi-1") + assert item is not None + self.assertEqual( + int(item.metadata.get("attempt_crash_streak")), round_index + 1 + ) + phase = item.phase + + item = await store.get_delegation_work_item("wi-1") + assert item is not None + self.assertTrue(attempt_ledger_dispatch_block_reason(item.metadata)) + self.assertFalse(is_dispatchable(item)) + # A clean settle resets the streak and dispatch reopens. + await store.update_delegation_work_item( + "wi-1", + metadata_updates={ + "attempt_crash_streak": 0, + }, + ) + item = await store.get_delegation_work_item("wi-1") + assert item is not None + self.assertTrue(is_dispatchable(item)) + + async def test_settle_interrupted_is_idempotent_and_blocks_at_limit(self) -> None: + store = await self._store() + await self._seed_item(store) + for round_index in range(ATTEMPT_INTERRUPTED_STREAK_LIMIT): + claimed = await self._claim( + store, "wi-1", Phase.READY if round_index == 0 else Phase.RUNNING + ) + settled = await settle_open_attempt_as_interrupted(store, claimed) + self.assertTrue(settled) + # Second settle on the same attempt is a no-op. + self.assertFalse(await settle_open_attempt_as_interrupted(store, claimed)) + item = await store.get_delegation_work_item("wi-1") + assert item is not None + self.assertEqual( + int(item.metadata.get("attempt_interrupted_streak")), round_index + 1 + ) + # Free the claim like the suspend/startup sweeps do. + await store.update_delegation_work_item( + "wi-1", + claimed_by_role_runtime_session_id="", + claimed_by_seat_id="", + metadata_updates={"claimed_by_role_session_id": "", "claimed_task_id": ""}, + ) + item = await store.get_delegation_work_item("wi-1") + assert item is not None + self.assertTrue(attempt_ledger_dispatch_block_reason(item.metadata)) + self.assertFalse(is_dispatchable(item)) + + +class AttemptLedgerReconcileTests(unittest.IsolatedAsyncioTestCase): + async def _executor(self) -> tuple[CompanyWorkItemExecutor, OPCStore]: + tmpdir = tempfile.TemporaryDirectory() + self.addCleanup(tmpdir.cleanup) + root = Path(tmpdir.name) + store = OPCStore(root / "tasks.db") + await store.initialize() + self.addAsyncCleanup(store.close) + config = OPCConfig() + config.org.company_profile = "custom" + config.org.final_decider_role_id = "executor" + config.org.roles = [ + RoleConfig(id="executor", name="Executor", responsibility="Do work.", reports_to="owner"), + ] + org_engine = OrgEngine(config, root) + communication = CommunicationManager(store, EventBus(), llm=None, org_engine=org_engine) + executor = CompanyWorkItemExecutor( + org_engine=org_engine, + communication=communication, + approval_engine=SimpleNamespace(), + memory=None, + execute_task=AsyncMock(), + save_task=store.save_task, + store=store, + llm=None, + ) + return executor, store + + async def test_reconcile_backfills_dead_attempt_and_terminalizes(self) -> None: + executor, store = await self._executor() + # A card whose owner died: open attempt, no claim, one interruption + # short of the limit — the reconcile pass must settle (streak hits the + # limit) and terminalize with a visible blocked_reason. + item = DelegationWorkItem( + work_item_id="wi-dead", + run_id="run-1", + role_id="executor", + seat_id="seat-1", + title="Doomed", + summary="Crash loops forever.", + kind="execute", + projection_id="doomed", + phase=Phase.RUNNING, + metadata={ + "attempt_seq": ATTEMPT_INTERRUPTED_STREAK_LIMIT, + "attempt_settled": False, + "attempt_interrupted_streak": ATTEMPT_INTERRUPTED_STREAK_LIMIT - 1, + }, + ) + await store.save_delegation_work_item(item) + + reconciled = await executor._reconcile_attempt_ledger([item], {}) + + refreshed = await store.get_delegation_work_item("wi-dead") + assert refreshed is not None + self.assertEqual(refreshed.phase, Phase.FAILED) + self.assertTrue(bool(refreshed.metadata.get("attempt_settled"))) + self.assertEqual(refreshed.metadata.get("attempt_outcome"), "interrupted") + self.assertEqual( + int(refreshed.metadata.get("attempt_interrupted_streak")), + ATTEMPT_INTERRUPTED_STREAK_LIMIT, + ) + self.assertIn("attempt ledger", str(refreshed.blocked_reason or "")) + self.assertEqual(len(reconciled), 1) + + async def test_reconcile_leaves_live_and_healthy_items_alone(self) -> None: + executor, store = await self._executor() + healthy = DelegationWorkItem( + work_item_id="wi-healthy", + run_id="run-1", + role_id="executor", + seat_id="seat-1", + title="Healthy", + summary="Fine.", + kind="execute", + projection_id="healthy", + phase=Phase.READY, + metadata={}, + ) + await store.save_delegation_work_item(healthy) + # Live item: open attempt but currently claimed by this runtime. + live = DelegationWorkItem( + work_item_id="wi-live", + run_id="run-1", + role_id="executor", + seat_id="seat-1", + title="Live", + summary="Running now.", + kind="execute", + projection_id="live", + phase=Phase.RUNNING, + metadata={"attempt_seq": 1, "attempt_settled": False}, + ) + await store.save_delegation_work_item(live) + executor.runtime._claimed_work_item_ids.add("wi-live") + + await executor._reconcile_attempt_ledger([healthy, live], {}) + + refreshed_live = await store.get_delegation_work_item("wi-live") + assert refreshed_live is not None + self.assertEqual(refreshed_live.phase, Phase.RUNNING) + self.assertFalse(bool(refreshed_live.metadata.get("attempt_settled"))) + refreshed_healthy = await store.get_delegation_work_item("wi-healthy") + assert refreshed_healthy is not None + self.assertEqual(refreshed_healthy.phase, Phase.READY) + self.assertNotIn("attempt_settled", dict(refreshed_healthy.metadata or {})) + + +class ResumeAvailabilityGateTests(unittest.IsolatedAsyncioTestCase): + """Fix-1: a resume pin to a disabled external agent fails closed.""" + + async def _store(self) -> OPCStore: + tmpdir = tempfile.TemporaryDirectory() + self.addCleanup(tmpdir.cleanup) + store = OPCStore(Path(tmpdir.name) / "tasks.db") + await store.initialize() + self.addAsyncCleanup(store.close) + return store + + def _plan(self) -> CompanyWorkItemRuntimePlan: + return CompanyWorkItemRuntimePlan( + profile="corporate", + projections=[ + WorkItemProjectionSpec( + projection_id="execution", + turn_type="execute", + title="Execution", + summary="Produce the main execution output.", + role_id="executor", + ) + ], + metadata={ + "execution_model": "multi_team_org", + "runtime_model": "multi_team_org", + "final_decider_role_id": "executor", + "top_level_role_ids": ["executor"], + }, + ) + + async def _seed(self, store: OPCStore, *, external_agent: str) -> Task: + plan = self._plan() + await store.save_delegation_role_session( + DelegationRoleSession( + role_session_id="role-runtime-1", + run_id="run-1", + project_id="proj1", + role_id="executor", + seat_id="seat-1", + ) + ) + await store.save_delegation_work_item( + DelegationWorkItem( + work_item_id="work-item-1", + run_id="run-1", + role_id="executor", + seat_id="seat-1", + title="Execution", + summary="Execute the project.", + kind="execute", + projection_id="execution", + phase=Phase.RUNNING, + claimed_by_role_runtime_session_id="role-runtime-1", + claimed_by_seat_id="seat-1", + metadata={"work_item_projection_id": "execution"}, + ) + ) + task = Task( + id="execution-task", + title="Execution", + session_id="sess-child", + parent_session_id="sess-parent", + status=TaskStatus.RUNNING, + project_id="proj1", + assigned_to="executor", + assigned_external_agent=external_agent, + execution_lock=True, + metadata={ + "company_profile": "corporate", + "execution_model": "multi_team_org", + "runtime_model": "multi_team_org", + "work_item_runtime": True, + "work_item_projection_id": "execution", + "delegation_run_id": "run-1", + "delegation_role_session_id": "role-runtime-1", + "selected_execution_agent": external_agent, + "company_work_item_plan": serialize_company_work_item_runtime_plan(plan), + }, + ) + set_linked_work_item_id(task, "work-item-1") + await store.save_task(task) + await store.link_work_item_runtime_task("work-item-1", "execution-task") + return task + + def _engine(self, store: OPCStore, *, available: list[str]) -> OPCEngine: + engine = OPCEngine() + engine.project_id = "proj1" + engine.store = store + engine.adapter_registry = SimpleNamespace( + list_available=lambda: list(available), + ) + return engine + + async def test_resume_fails_closed_when_pinned_agent_unavailable(self) -> None: + store = await self._store() + task = await self._seed(store, external_agent="codex") + engine = self._engine(store, available=["opencode"]) # codex disabled + suspended = await engine.suspend_company_runtime( + origin_task_id=task.id, + session_id="sess-parent", + reason="user_stop", + ) + self.assertIsNotNone(suspended) + + executed: dict[str, list[Task]] = {} + + class DummyCompanyExecutor: + async def execute(self, _plan, tasks: list[Task]) -> str: + executed["tasks"] = tasks + return "runtime resumed" + + engine.company_executor = DummyCompanyExecutor() + + response = await engine._maybe_resume_checkpoint( + "continue", + "sess-parent", + reply_metadata={"ui_force_resume": True}, + ) + + self.assertIsNotNone(response) + refreshed_item = await store.get_delegation_work_item("work-item-1") + assert refreshed_item is not None + self.assertEqual(refreshed_item.phase, Phase.FAILED) + self.assertIn("codex", str(refreshed_item.blocked_reason or "")) + refreshed_task = await store.get_task(task.id) + assert refreshed_task is not None + self.assertEqual(refreshed_task.status, TaskStatus.FAILED) + self.assertEqual( + refreshed_task.metadata.get("resume_unavailable_external_agent"), + "codex", + ) + # The runtime still executed (the rest of the org resumes normally). + self.assertIn("tasks", executed) + + async def test_plain_message_after_gate_failure_converges_without_revival(self) -> None: + """A plain text follow-up (final-decider routing path) on a run whose + decider card failed terminally must drain the checkpoint and must not + clobber the FAILED task back to PENDING (InvalidPhaseTransition crash + found by the issue #10 end-to-end reproduction).""" + store = await self._store() + task = await self._seed(store, external_agent="codex") + engine = self._engine(store, available=["opencode"]) # codex disabled + await engine.suspend_company_runtime( + origin_task_id=task.id, + session_id="sess-parent", + reason="user_stop", + ) + + class DummyCompanyExecutor: + async def execute(self, _plan, tasks: list[Task]) -> str: + return "runtime resumed" + + engine.company_executor = DummyCompanyExecutor() + # First resume: gate fails the codex-pinned decider card closed. + await engine._maybe_resume_checkpoint( + "continue", + "sess-parent", + reply_metadata={"ui_force_resume": True}, + ) + refreshed_item = await store.get_delegation_work_item("work-item-1") + assert refreshed_item is not None + self.assertEqual(refreshed_item.phase, Phase.FAILED) + # Re-suspend cannot happen (run is terminal) — but if a pending + # suspend checkpoint still exists, a plain message must converge + # instead of raising InvalidPhaseTransition. Route a plain message + # through the checkpoint machinery when present, else assert the + # terminal state simply holds. + response = await engine._maybe_resume_checkpoint("重跑", "sess-parent") + self.assertNotIsInstance(response, Exception) + refreshed_item = await store.get_delegation_work_item("work-item-1") + assert refreshed_item is not None + self.assertEqual(refreshed_item.phase, Phase.FAILED) + refreshed_task = await store.get_task(task.id) + assert refreshed_task is not None + self.assertEqual( + refreshed_task.status, + TaskStatus.FAILED, + "follow-up routing must not clobber the FAILED task projection", + ) + remaining = await store.get_pending_checkpoints( + project_id="proj1", + session_id="sess-parent", + ) + self.assertEqual( + [c.checkpoint_type for c in remaining], + [], + "suspend checkpoint must drain instead of bouncing back to pending", + ) + + async def test_resume_proceeds_when_pinned_agent_available(self) -> None: + store = await self._store() + task = await self._seed(store, external_agent="codex") + engine = self._engine(store, available=["codex", "opencode"]) + suspended = await engine.suspend_company_runtime( + origin_task_id=task.id, + session_id="sess-parent", + reason="user_stop", + ) + self.assertIsNotNone(suspended) + + class DummyCompanyExecutor: + async def execute(self, _plan, tasks: list[Task]) -> str: + return "runtime resumed" + + engine.company_executor = DummyCompanyExecutor() + await engine._maybe_resume_checkpoint( + "continue", + "sess-parent", + reply_metadata={"ui_force_resume": True}, + ) + refreshed_item = await store.get_delegation_work_item("work-item-1") + assert refreshed_item is not None + self.assertEqual(refreshed_item.phase, Phase.RUNNING) + refreshed_task = await store.get_task(task.id) + assert refreshed_task is not None + self.assertEqual(refreshed_task.status, TaskStatus.RUNNING) + self.assertEqual( + refreshed_task.metadata.get("_company_runtime_resume_execution_agent_pin", {}).get( + "selected_execution_agent" + ), + "codex", + ) + + +if __name__ == "__main__": + unittest.main()