diff --git a/opc/engine.py b/opc/engine.py index 574273b..48e3d49 100644 --- a/opc/engine.py +++ b/opc/engine.py @@ -5520,6 +5520,10 @@ class OPCEngine: work_item: DelegationWorkItem, work_item_by_id: dict[str, DelegationWorkItem], ) -> bool: + from opc.layer2_organization.work_item_transition import ( + settled_failure_dependency_ids, + ) + metadata = dict(getattr(work_item, "metadata", {}) or {}) dependency_ids = [ str(item).strip() @@ -5529,6 +5533,7 @@ class OPCEngine: if not dependency_ids: return True dependency_classes = dict(metadata.get("dependency_classes", {}) or {}) + settled_failure_ids = settled_failure_dependency_ids(metadata) for dep_id in dependency_ids: dependency = work_item_by_id.get(dep_id) if dependency is None: @@ -5539,9 +5544,18 @@ class OPCEngine: continue if dep_class == "soft": if dep_phase not in DONE_PHASES and dep_phase not in IN_PROGRESS_PHASES: + if dep_id in settled_failure_ids: + continue return False continue if dep_phase != Phase.APPROVED: + # Failure-triage release: the frontier pass released this + # card over the dep (dependency_settlement stamp). Stop/ + # resume must not regress a released triage card back into + # WAITING_* — with the failed dep already terminal, no + # later event would ever wake it again. + if dep_id in settled_failure_ids: + continue return False return True diff --git a/opc/layer2_organization/company_mode.py b/opc/layer2_organization/company_mode.py index df72200..508d917 100644 --- a/opc/layer2_organization/company_mode.py +++ b/opc/layer2_organization/company_mode.py @@ -7,6 +7,7 @@ import copy import hashlib import inspect import json +import re import uuid from contextvars import ContextVar, Token from dataclasses import asdict, dataclass, field @@ -101,8 +102,12 @@ from opc.layer2_organization.recruiter import ( ) from opc.layer2_organization.seat_executor import SeatExecutor from opc.layer2_organization.work_item_transition import ( + DEPENDENCY_CLASS_DEFAULT, + compute_doomed_work_item_ids, + has_pending_settlement_release, normalize_dependency_work_item_ids, refresh_dependents_for_run, + settled_failure_dependency_ids, transition_work_item_from_task, ) from opc.layer2_organization.work_item_identity import ( @@ -152,6 +157,23 @@ from opc.llm.retry import LLMRetryError, call_llm_json_with_retry # exhausted the turn exits with a parked summary instead of spinning forever. _HUMAN_WAIT_MAX_STALL_TICKS = 24 +# Matches the manager dispatch guard's escape line while tolerating the +# markdown decoration models routinely wrap protocol tokens in — bold +# (`**NO_DELEGATION_JUSTIFICATION**:`), headings, quotes, list markers, +# `_`/`-`/space separator variants, and fullwidth colons. A bare +# ``startswith("NO_DELEGATION_JUSTIFICATION:")`` here cost project 4444 its +# CTO card: the justification was written but bold-wrapped, went unparsed, +# and the guard drove the work item to FAILED. +_NO_DELEGATION_JUSTIFICATION_LINE = re.compile( + r"^\s*(?:>+\s*)*" # blockquote markers + r"(?:[-*+•]\s+|\d+[.)]\s+)?" # list markers + r"[\s#*_`~\"']*" # heading/bold/quote decoration + r"NO[_\s-]?DELEGATION[_\s-]?JUSTIFICATION" + r"[\s*_`~\"']*" # decoration between token and colon + r"[:：]\s*(?P.*?)\s*$", + re.IGNORECASE, +) + def review_work_item_id_for_attempt(worker_work_item_id: str, attempt: int) -> str: """Compute a per-attempt review work-item ID for a given worker. @@ -2700,19 +2722,31 @@ class CompanyWorkItemExecutor: owner_work_item_id=str(getattr(work_item, "work_item_id", "") or "").strip(), ) dependency_classes = dict(metadata.get("dependency_classes", {}) or {}) + settled_failure_ids = settled_failure_dependency_ids(metadata) for dep_id in dependency_ids: dependency = work_item_by_id.get(dep_id) if dependency is None: continue dep_phase = dependency.phase - dep_class = str(dependency_classes.get(dep_id, "hard") or "hard").strip().lower() + dep_class = str( + dependency_classes.get(dep_id, DEPENDENCY_CLASS_DEFAULT) + or DEPENDENCY_CLASS_DEFAULT + ).strip().lower() if dep_class == "info": continue if dep_class == "soft": if dep_phase not in DONE_PHASES and dep_phase not in IN_PROGRESS_PHASES: + if dep_id in settled_failure_ids: + continue return False continue if dep_phase != Phase.APPROVED: + # Failure-triage release: the frontier pass stamped this + # card's dependency_settlement over the dep (terminal + # failure, or stuck behind one), so it is settled + # context here, not a blocker. + if dep_id in settled_failure_ids: + continue return False return True adaptive = cls._normalize_adaptive_metadata(metadata.get("adaptive", {})) @@ -2730,15 +2764,24 @@ class CompanyWorkItemExecutor: work_item_by_id, owner_work_item_id=str(getattr(work_item, "work_item_id", "") or "").strip(), ) + settled_failure_ids = settled_failure_dependency_ids(metadata) for dep_id in all_dep_ids: - dep_class = dep_classes_map.get(dep_id, "soft") + # Default must match _dependency_release_state and the doomed + # computation (both DEPENDENCY_CLASS_DEFAULT="hard"): a "soft" + # default here let a card count as runnable while the frontier + # counted it doomed — divergent verdicts on the same dep. + dep_class = dep_classes_map.get(dep_id, DEPENDENCY_CLASS_DEFAULT) dependency = work_item_by_id.get(dep_id) if dependency is None: continue dep_phase = dependency.phase if dep_class == "hard" and dep_phase != Phase.APPROVED: + if dep_id in settled_failure_ids: + continue return False if dep_class == "soft" and dep_phase not in DONE_PHASES and dep_phase not in IN_PROGRESS_PHASES: + if dep_id in settled_failure_ids: + continue return False if str(adaptive.get("normalized_state", "") or "").strip().lower() == "invalidated": return False @@ -3366,6 +3409,7 @@ class CompanyWorkItemExecutor: if not self.store or not work_items: return work_items work_items = await self._repair_stuck_aggregate_review_items(work_items) + work_items = await self._reconcile_missing_review_chain(work_items) work_item_by_id = {item.work_item_id: item for item in work_items} changed = False for work_item in work_items: @@ -3408,6 +3452,22 @@ class CompanyWorkItemExecutor: metadata_updates=dependency_state["metadata_updates"], ) changed = True + # Failure frontier for late-created cards: a delivery/aggregate card + # created AFTER its dependency already failed never sees a failure + # transition hook, and the per-item pass above only releases on + # all-approved. Detection is cheap and idempotent — released cards + # (stamp present, phase moved) stop matching. + if has_pending_settlement_release(work_item_by_id): + run_id = str(work_items[0].run_id or "").strip() + if run_id: + try: + if await refresh_dependents_for_run(self.store, run_id=run_id): + changed = True + except Exception: + logger.opt(exception=True).debug( + "Best-effort settlement frontier refresh failed for run " + f"{run_id}" + ) if not changed: return work_items try: @@ -3549,6 +3609,99 @@ class CompanyWorkItemExecutor: return await self.store.list_delegation_work_items(run_id) return work_items + @staticmethod + def _task_carrier_for_work_item(item: DelegationWorkItem) -> Task: + """Minimal Task stand-in for lifecycle helpers when the original + runtime task row is unavailable (legacy DB rows, crash recovery). + Metadata keys mirror what _ensure_report_work_item_for_work_item + reads off a real runtime task.""" + item_metadata = dict(getattr(item, "metadata", {}) or {}) + task = Task( + id=f"reconcile::{item.work_item_id}", + title=str(item.title or "").strip() or item.work_item_id, + description=str(item.summary or "").strip(), + project_id="default", + assigned_to=str(item.role_id or "").strip(), + status=TaskStatus.DONE, + metadata={ + "execution_mode": "company_mode", + "runtime_model": str(item_metadata.get("runtime_model", "") or "multi_team_org"), + **build_work_item_owner_execution_copy(item), + }, + ) + set_linked_work_item_id(task, item.work_item_id) + return task + + async def _reconcile_missing_review_chain( + self, + work_items: list[DelegationWorkItem], + ) -> list[DelegationWorkItem]: + """Idempotently rebuild the report card for reviewable work items + parked in AWAITING_MANAGER_REVIEW with no live report/review card. + + A card in that phase is only ever advanced by its report→review + auxiliary chain; if the chain is missing (legacy DBs where the + report spawn refused self-produced dispatch parents before + ``turn_output_kind`` existed, or a crash between the phase write + and the spawn), nothing will ever consume the review and the card + waits forever. Detection is cheap (set lookup over the run + snapshot), and _ensure_report_work_item_for_work_item is + idempotent per attempt, so re-running every tick is safe. + """ + if not self.store or not work_items or not hasattr(self.store, "save_delegation_work_item"): + return work_items + waiting = [ + item for item in work_items + if item.phase == Phase.AWAITING_MANAGER_REVIEW and is_manager_reviewable_turn(item) + ] + if not waiting: + return work_items + live_aux_targets: set[str] = set() + for item in work_items: + if item.phase in DONE_PHASES: + continue + item_metadata = dict(item.metadata or {}) + for key in ("report_target_work_item_id", "review_target_work_item_id"): + target = str(item_metadata.get(key, "") or "").strip() + if target: + live_aux_targets.add(target) + rebuilt_ids: list[str] = [] + for item in waiting: + if item.work_item_id in live_aux_targets: + continue + worker_task: Task | None = None + claimed_task_id = str((item.metadata or {}).get("claimed_task_id", "") or "").strip() + if claimed_task_id and hasattr(self.store, "get_task"): + try: + worker_task = await self.store.get_task(claimed_task_id) + except Exception: + worker_task = None + if worker_task is None: + worker_task = self._task_carrier_for_work_item(item) + try: + spawned = await self._ensure_report_work_item_for_work_item( + item.work_item_id, + worker_task=worker_task, + ) + except Exception: + logger.opt(exception=True).warning( + "Failed to rebuild missing report card for reviewable work item " + f"work_item_id={item.work_item_id}" + ) + continue + if spawned is not None: + rebuilt_ids.append(item.work_item_id) + if not rebuilt_ids: + return work_items + logger.info( + "Rebuilt missing report cards for work items awaiting manager review: " + + ", ".join(rebuilt_ids) + ) + run_id = str(work_items[0].run_id or "").strip() + if run_id and hasattr(self.store, "list_delegation_work_items"): + return await self.store.list_delegation_work_items(run_id) + return work_items + async def _reconcile_role_serial_queues( self, work_items: list[DelegationWorkItem], @@ -5079,6 +5232,12 @@ class CompanyWorkItemExecutor: while True: task.metadata.pop("_retry_contract_enforcement", None) + # Turn-scoped: the dispatch guard sets this when THIS turn + # mutated the board. Left over from a previous turn it would + # let a later non-delegating turn skip the guard entirely (and + # clear the justification markers), so a manager that delegated + # once could self-produce unreviewed forever after. + task.metadata.pop("manager_board_mutation_performed", None) manager_dispatch_retry_count = int( task.metadata.get("_manager_dispatch_retry_count", 0) or 0 ) @@ -5273,38 +5432,24 @@ class CompanyWorkItemExecutor: task_id=task.id, ) continue - # Retries exhausted. Preserve whatever the agent produced - # (content + artifacts) so the user can inspect the work - # even though the dispatch policy was never satisfied — - # historically this content was overwritten with the - # violation message and the turn output was lost. + # Retries exhausted. Dispatch is a soft constraint: the org + # chart fixes who *can* delegate, but not every task needs + # every seat, so accept the turn output as normal completion + # instead of failing the work item. Record the unresolved + # guard note so reviewers and the delivery report can weigh + # the output accordingly. task.metadata = dict(task.metadata or {}) - preserved_content = str(getattr(result, "content", "") or "") - if preserved_content: - task.metadata["last_turn_preserved_content"] = preserved_content - task.metadata["manager_dispatch_guard_terminal_violation"] = violation_text - await transition_work_item_from_task( - self.store, task, - target_status_or_phase=Phase.FAILED, - reason="manager_dispatch_guard_violation", + task.metadata["manager_dispatch_guard_unresolved"] = violation_text + await self._append_progress( + task, + "Dispatch guard reminders exhausted; accepting the turn output " + "without delegation (noted for review).", ) - await self.save_task(task) await self._emit_progress( - f"[Company:{projection_id}] failed manager dispatch guard", + f"[Company:{projection_id}] accepted manager turn without delegation " + f"after dispatch guard reminders were exhausted", task_id=task.id, ) - failure_content = violation_text - if preserved_content: - failure_content = ( - f"{violation_text}\n\n---\n" - f"Preserved agent output (not accepted as work-item output):\n" - f"{preserved_content}" - ) - return TaskResult( - status=TaskStatus.FAILED, - content=failure_content, - artifacts=dict(result.artifacts or {}), - ) task.metadata.pop("_manager_dispatch_retry_count", None) task.metadata.pop("manager_dispatch_guard_terminal_violation", None) task.context_snapshot = dict(task.context_snapshot or {}) @@ -5415,6 +5560,74 @@ class CompanyWorkItemExecutor: ) return result + # Turn types whose completion is review-exempt ONLY because their + # deliverable is normally a delegated child card set (each child gets + # its own manager review). When such a turn instead completes with the + # manager's own work product — no live children — that output must go + # through manager review like any execute turn. + _DELEGATION_OUTPUT_TURN_TYPES = frozenset({"dispatch", "intake", "plan"}) + + async def _classify_delegation_turn_output( + self, + task: Task, + work_item_id: str, + ) -> tuple[str, str]: + """Classify what a delegation-kind turn actually delivered. + + Ground truth is the store, not transient task markers: a card that + completes with live (non-deleted, non-auxiliary) children delivered + a delegated board; one without any delivered its own work product. + + Returns ``(output_kind, output_source)`` where output_kind is + ``"delegated"`` or ``"self_produced"`` and output_source records how + the dispatch guard was satisfied for self-produced output + (``justified`` / ``dispatch_guard_exhausted`` / ``no_child_work``). + """ + run_id = str((task.metadata or {}).get("delegation_run_id", "") or "").strip() + if run_id and self.store and hasattr(self.store, "list_delegation_work_items"): + try: + run_items = await self.store.list_delegation_work_items(run_id) + except Exception: + run_items = [] + for item in run_items: + if str(getattr(item, "parent_work_item_id", "") or "").strip() != work_item_id: + continue + item_metadata = dict(getattr(item, "metadata", {}) or {}) + if bool(item_metadata.get("deleted_by_manager_tool", False)): + continue + # Hidden report/review cards are runtime plumbing spawned + # under the worker card, not delegated business work. + if bool(item_metadata.get("report_execution_work_item", False)) or bool( + item_metadata.get("review_execution_work_item", False) + ): + continue + return ("delegated", "") + if str((task.metadata or {}).get("manager_no_delegation_justification", "") or "").strip(): + return ("self_produced", "justified") + if str((task.metadata or {}).get("manager_dispatch_guard_unresolved", "") or "").strip(): + return ("self_produced", "dispatch_guard_exhausted") + return ("self_produced", "no_child_work") + + def _has_agent_manager_above(self, task: Task, linked_work_item: Any) -> bool: + """True when the card's manager is a real agent role that can run a + review turn. Top seats report to the human ``owner`` — a review card + assigned there is unclaimable (the a7846729 stuck-review case), so + self-produced output at the top auto-approves and is covered by the + final delivery's human acceptance instead.""" + manager_role_id = ( + str((task.metadata or {}).get("manager_role_id", "") or "").strip() + or str(getattr(linked_work_item, "manager_role_id", "") or "").strip() + ) + if not manager_role_id or manager_role_id == "owner": + return False + get_agent = getattr(getattr(self, "org_engine", None), "get_agent", None) + if callable(get_agent): + try: + return get_agent(manager_role_id) is not None + except Exception: + return True + return True + async def _apply_done_transition( self, task: Task, @@ -5529,6 +5742,21 @@ class CompanyWorkItemExecutor: is_delivery_turn(task.metadata) or str(task.metadata.get("review_owner_kind", "") or "").strip().lower() == "human" ) + turn_output_kind = "" + turn_output_source = "" + if ( + not manager_reviewable + and not is_attention_work_item + and not is_delivery_card + and work_kind in self._DELEGATION_OUTPUT_TURN_TYPES + ): + turn_output_kind, turn_output_source = await self._classify_delegation_turn_output( + task, work_item_id, + ) + if turn_output_kind == "self_produced" and self._has_agent_manager_above( + task, linked_work_item, + ): + manager_reviewable = True if is_attention_work_item: # Attention work items are wake-up wrappers that let a parked # manager consume inbox/board state and call orchestration tools. @@ -5556,6 +5784,16 @@ class CompanyWorkItemExecutor: **work_item_identity_payload_for_task(task), "adaptive": dict(task.metadata.get("adaptive", {}) or {}), } + if turn_output_kind: + # Persist the classification on the WorkItem so every downstream + # consumer of is_manager_reviewable_turn (report spawn, report + # completion, recovery scans) reads the same fact instead of + # re-deriving it from the static turn type. + metadata_updates["turn_output_kind"] = turn_output_kind + task.metadata["turn_output_kind"] = turn_output_kind + if turn_output_source: + metadata_updates["turn_output_source"] = turn_output_source + task.metadata["turn_output_source"] = turn_output_source if is_attention_work_item: metadata_updates["attention_work_item_outcome"] = "completed" if target_phase in {Phase.AWAITING_MANAGER_REVIEW, Phase.AWAITING_HUMAN}: @@ -7997,6 +8235,57 @@ class CompanyWorkItemExecutor: if counts: counts_text = ", ".join(f"{phase}={count}" for phase, count in sorted(counts.items())) lines.append(f"Children by phase: {counts_text}") + settlement = {} + if current_item is not None: + settlement = dict( + (current_item.metadata or {}).get("dependency_settlement", {}) or {} + ) + failed_board_items = [ + item for item in board_items + if getattr(item, "phase", None) in (Phase.FAILED, Phase.CANCELLED) + ] + if failed_board_items or settlement: + lines.append("### Failed or cancelled children (triage needed)") + for item in failed_board_items[:8]: + item_meta = dict(item.metadata or {}) + reason = str(item_meta.get("last_transition_reason", "") or "").strip() + summary_text = str(item.summary or "").strip() + entry = ( + f"- `{str(item.work_item_id or '').strip()}` [{item.phase.value}] " + f"{str(item.role_id or '').strip()}" + ) + if reason: + entry += f" reason={reason}" + if summary_text: + entry += ": " + clip_text( + summary_text, limit=240, marker="failed child summary truncated" + ).text + lines.append(entry) + preserved = str(item_meta.get("last_turn_preserved_content", "") or "").strip() + if preserved: + lines.append( + " Output preserved from its final turn (not lost): " + + clip_text(preserved, limit=700, marker="preserved output truncated").text + ) + stuck_ids = [ + str(item).strip() + for item in list(settlement.get("stuck", []) or []) + if str(item).strip() + ] + if stuck_ids: + lines.append( + "Downstream children blocked by these failures: " + + ", ".join(f"`{sid}`" for sid in stuck_ids[:8]) + + (" ..." if len(stuck_ids) > 8 else "") + + ". They are cancelled automatically when this turn completes " + "unless you rebuild or rewire them." + ) + lines.append( + "You can rebuild the failed work with `delegate_work` (pair it with " + "`delete_work_item` + `replacement_dependency_work_item_ids` to rewire " + "dependents), continue with the successful results only, or record the " + "gap in your handoff so the upper role or user can decide." + ) lines.append( "Use `manager_board_read` without `parent_work_item_id` to inspect this business board. " "Do not call `delegate_work` again for any existing `scope_key`; use `modify_work_item` or `delete_work_item` " @@ -8229,6 +8518,25 @@ class CompanyWorkItemExecutor: "child_mutation_state": child_mutation_state, } + @staticmethod + def _genuine_no_delegation_justification(text: str) -> str: + """Cleaned justification text, or "" for empty input or an echo of + the instruction template's `` placeholder — with + any markdown decoration, quoting, or trailing punctuation around + the placeholder stripped before the check.""" + reason = re.sub(r"[\s*_`~\"']+$", "", str(text or "")).strip() + if not reason: + return "" + core = reason + previous = None + while previous != core: + previous = core + core = core.strip().strip("*_~`\"'") + core = re.sub(r"[\s.。!！,，;；:：]+$", "", core) + if re.fullmatch(r"<[^<>]*>", core): + return "" + return reason + @staticmethod def _extract_no_delegation_justification(task: Task, result: TaskResult | None) -> str: artifact_candidates = [] @@ -8249,15 +8557,19 @@ class CompanyWorkItemExecutor: ] ) for candidate in artifact_candidates: - if candidate: - return candidate + cleaned = CompanyWorkItemExecutor._genuine_no_delegation_justification(candidate) + if cleaned: + return cleaned content = str(getattr(result, "content", "") or "").strip() for line in content.splitlines(): - stripped = str(line).strip() - if not stripped: + match = _NO_DELEGATION_JUSTIFICATION_LINE.match(str(line)) + if not match: continue - if stripped.upper().startswith("NO_DELEGATION_JUSTIFICATION:"): - return stripped.split(":", 1)[1].strip() + reason = CompanyWorkItemExecutor._genuine_no_delegation_justification( + match.group("reason") + ) + if reason: + return reason return "" @staticmethod @@ -8344,6 +8656,7 @@ class CompanyWorkItemExecutor: task.metadata = dict(task.metadata or {}) task.metadata["manager_board_mutation_performed"] = True task.metadata.pop("manager_no_delegation_justification", None) + task.metadata.pop("manager_dispatch_guard_unresolved", None) return [] justification = self._extract_no_delegation_justification(task, result) if justification: @@ -8355,6 +8668,7 @@ class CompanyWorkItemExecutor: ] task.metadata = dict(task.metadata or {}) task.metadata["manager_no_delegation_justification"] = justification + task.metadata.pop("manager_dispatch_guard_unresolved", None) return [] direct_reports = [ str(item).strip() @@ -8994,15 +9308,84 @@ class CompanyWorkItemExecutor: ) await self.save_task(task) return False # intake does not park; it closes out + # A dependency only counts as pending while it can still move on its + # own: terminal deps (APPROVED/FAILED/CANCELLED) and doomed deps + # (transitively blocked by a terminal failure) are settled. Without + # this, a triage turn that accepts partial results re-parks forever + # on the already-FAILED child it just triaged — the settlement + # release and this gate must agree on what "settled" means. + park_doomed_ids: set[str] = set() + park_items_by_id: dict[str, Any] = {} + if parent_work_item is not None and hasattr(self.store, "list_delegation_work_items"): + try: + park_run_items = await self.store.list_delegation_work_items(parent_work_item.run_id) + except Exception: + park_run_items = [] + park_items_by_id = { + str(getattr(item, "work_item_id", "") or "").strip(): item + for item in park_run_items + if str(getattr(item, "work_item_id", "") or "").strip() + } + park_doomed_ids = compute_doomed_work_item_ids(park_items_by_id) pending_dependency_ids: list[str] = [] + settled_failures_present = False for dep_id in dependency_ids: - dependency = await self.store.get_delegation_work_item(dep_id) - if dependency is None or dependency.phase != Phase.APPROVED: + dependency = park_items_by_id.get(dep_id) + if dependency is None: + dependency = await self.store.get_delegation_work_item(dep_id) + if dependency is None: pending_dependency_ids.append(dep_id) + continue + if dependency.phase == Phase.APPROVED: + continue + if dependency.phase in DONE_PHASES or dep_id in park_doomed_ids: + settled_failures_present = True + continue + pending_dependency_ids.append(dep_id) task.metadata = dict(task.metadata) task.metadata["delegation_wait_for_work_item_ids"] = dependency_ids if not pending_dependency_ids: task.metadata.pop("delegation_pending_work_item_ids", None) + parent_meta_for_stamp = ( + dict(getattr(parent_work_item, "metadata", {}) or {}) + if parent_work_item is not None + else {} + ) + has_settlement_stamp = bool( + dict(parent_meta_for_stamp.get("dependency_settlement", {}) or {}) + ) + if settled_failures_present and not has_settlement_stamp: + # Race: the children settled with failures while this turn + # was still running, so the failure's transition hook fired + # before this card could park — no triage release has been + # scheduled (and the concurrent refresh may already have + # regressed our phase to WAITING_FOR_CHILDREN without a + # stamp). Park now and run the frontier immediately: the + # settlement release re-arms the triage turn. + await transition_work_item_from_task( + self.store, task, + target_status_or_phase=Phase.WAITING_FOR_CHILDREN, + reason="park_for_settled_failures", + metadata_updates={ + "dependency_work_item_ids": dependency_ids, + "waiting_on_work_item_ids": [], + "delegated_children_pending": True, + }, + ) + await self.save_task(task) + try: + await refresh_dependents_for_run( + self.store, + run_id=str(getattr(parent_work_item, "run_id", "") or "").strip(), + source_work_item_id=parent_work_item_id, + source_task_id=task.id, + ) + except Exception: + logger.opt(exception=True).debug( + "park_for_settled_failures: frontier refresh failed for " + f"{parent_work_item_id}" + ) + return True return False task.metadata["delegation_pending_work_item_ids"] = pending_dependency_ids await self._append_progress( @@ -9011,6 +9394,33 @@ class CompanyWorkItemExecutor: + ", ".join(pending_dependency_ids[:8]) + (" ..." if len(pending_dependency_ids) > 8 else ""), ) + park_metadata_updates: dict[str, Any] = { + "dependency_work_item_ids": dependency_ids, + "waiting_on_work_item_ids": pending_dependency_ids, + "delegated_children_pending": True, + } + parent_meta_now = ( + dict(getattr(parent_work_item, "metadata", {}) or {}) + if parent_work_item is not None + else {} + ) + if bool(parent_meta_now.get("synthesis_turn_started")) or parent_meta_now.get( + "dependency_settlement" + ): + # Re-parking after a synthesis / failure-triage turn rebuilt the + # board: reset the one-shot synthesis marker and the stale + # settlement stamp so the next wake runs a clean synthesis pass + # over the new children (and the old failed-dep release cannot + # leak into the rebuilt card's runnability check). + park_metadata_updates["synthesis_turn_started"] = False + park_metadata_updates["dependency_settlement"] = {} + pre_kind = str(parent_meta_now.get("pre_synthesis_work_kind", "") or "").strip() + if pre_kind and str(parent_meta_now.get("work_kind", "") or "").strip().lower() in { + "synthesis", + "synthesize", + }: + park_metadata_updates["work_kind"] = pre_kind + park_metadata_updates["delegation_turn_kind"] = pre_kind # Phase A: single phase write, hook projects task.status=BLOCKED and # syncs local. Replaces the old "write task.status BLOCKED, save, # then separately write work_item.phase=WAITING_FOR_CHILDREN" double-pass. @@ -9018,11 +9428,7 @@ class CompanyWorkItemExecutor: self.store, task, target_status_or_phase=Phase.WAITING_FOR_CHILDREN, reason="park_for_delegated_children", - metadata_updates={ - "dependency_work_item_ids": dependency_ids, - "waiting_on_work_item_ids": pending_dependency_ids, - "delegated_children_pending": True, - }, + metadata_updates=park_metadata_updates, ) await self.save_task(task) return True @@ -13321,6 +13727,10 @@ class CompanyWorkItemExecutor: @staticmethod def _task_has_delegated_downstream_work(task: Task) -> bool: metadata = dict(getattr(task, "metadata", {}) or {}) + # Persisted completion classification (survives the per-turn reset + # of manager_board_mutation_performed below). + if str(metadata.get("turn_output_kind", "") or "").strip().lower() == "delegated": + return True if bool(metadata.get("manager_board_mutation_performed", False)): return True if bool(metadata.get("delegated_children_pending", False)): diff --git a/opc/layer2_organization/work_item_identity.py b/opc/layer2_organization/work_item_identity.py index b597cab..33355d2 100644 --- a/opc/layer2_organization/work_item_identity.py +++ b/opc/layer2_organization/work_item_identity.py @@ -155,7 +155,25 @@ def is_delivery_turn(value_or_metadata: Any) -> bool: def is_manager_reviewable_turn(value_or_metadata: Any) -> bool: - """Return True when a finished WorkItem should enter manager review flow.""" + """Return True when a finished WorkItem should enter manager review flow. + + An explicit persisted ``turn_output_kind`` wins over the turn-type + default: delegation-kind turns (dispatch/intake/plan) are review-exempt + only because their deliverable is normally a child card set that gets + reviewed per child. When such a turn completed with the manager's own + work product instead, the done-transition stamps + ``turn_output_kind=self_produced`` on the WorkItem and that output is + reviewable like any execute turn — every consumer of this predicate + (DONE routing, report spawn, report completion, recovery scans) follows + automatically. + """ + metadata: Mapping[str, Any] | None = None + if isinstance(value_or_metadata, Mapping): + metadata = value_or_metadata + elif hasattr(value_or_metadata, "metadata"): + metadata = getattr(value_or_metadata, "metadata", None) or {} + if metadata and str(metadata.get("turn_output_kind", "") or "").strip().lower() == "self_produced": + return True turn_type = _turn_type_for_value(value_or_metadata, fallback="") if not turn_type: return False diff --git a/opc/layer2_organization/work_item_transition.py b/opc/layer2_organization/work_item_transition.py index 3e9e5d5..7689fc0 100644 --- a/opc/layer2_organization/work_item_transition.py +++ b/opc/layer2_organization/work_item_transition.py @@ -401,6 +401,23 @@ _SYNTHESIS_SKIP_KINDS: frozenset[str] = frozenset({ "synthesize", }) +# Default dependency class when a dependency id has no entry in +# metadata.dependency_classes. Shared with the claim-side runnability check +# in company_mode so the frontier pass and the dispatcher never disagree on +# whether an unlabelled dependency is hard. +DEPENDENCY_CLASS_DEFAULT = "hard" + +# Roll-up kinds that may be released from WAITING_DEPENDENCIES by the +# failure-triage settlement path: their whole job is to integrate child +# results (including partial/failed ones) and carry them upward. +_SETTLEMENT_ROLLUP_KINDS: frozenset[str] = frozenset({ + "aggregate", + "deliver", + "delivery", + "synthesis", + "synthesize", +}) + def _work_item_id(item: DelegationWorkItem | Any | None) -> str: return str(getattr(item, "work_item_id", "") or "").strip() @@ -491,6 +508,94 @@ def normalize_dependency_work_item_ids( ) +def compute_doomed_work_item_ids( + work_item_by_id: dict[str, DelegationWorkItem | Any], +) -> set[str]: + """Ids of work items that can never reach APPROVED on their own. + + Seeds are FAILED/CANCELLED items. Propagation: a QUEUED / + WAITING_DEPENDENCIES / READY item whose hard dependency is doomed (and + not APPROVED) can never become runnable, so it is doomed too. Pure + fixpoint over the run snapshot — no IO. Dependencies are normalized + first, so a doomed card that was rewired via ``delete_work_item`` + + replacement ids drops back out of the set. + """ + doomed: set[str] = { + item_id + for item_id, item in work_item_by_id.items() + if getattr(item, "phase", None) in (Phase.FAILED, Phase.CANCELLED) + } + if not doomed: + return doomed + changed = True + while changed: + changed = False + for item_id, item in work_item_by_id.items(): + if item_id in doomed: + continue + if getattr(item, "phase", None) not in ( + Phase.QUEUED, + Phase.WAITING_DEPENDENCIES, + Phase.READY, + ): + continue + metadata = dict(getattr(item, "metadata", {}) or {}) + # A card carrying a settlement stamp was RELEASED by the + # frontier pass over its failures — it is alive (a runnable + # triage card), not doomed. Marking it doomed would let an + # upper parent settle early and, worse, let that parent's + # cascade cancel a triage card that is about to run. + if dict(metadata.get("dependency_settlement", {}) or {}): + continue + raw_ids = [ + str(dep).strip() + for dep in list(metadata.get("dependency_work_item_ids", []) or []) + if str(dep).strip() + ] + if not raw_ids: + continue + dep_ids, _pruned = normalize_dependency_work_item_ids( + raw_ids, work_item_by_id, owner_work_item_id=item_id + ) + dependency_classes = dict(metadata.get("dependency_classes", {}) or {}) + for dep_id in dep_ids: + dep = work_item_by_id.get(dep_id) + if dep is None or dep_id not in doomed: + continue + dep_class = str( + dependency_classes.get(dep_id, DEPENDENCY_CLASS_DEFAULT) + or DEPENDENCY_CLASS_DEFAULT + ).strip().lower() + if dep_class in ("soft", "info"): + continue + if getattr(dep, "phase", None) == Phase.APPROVED: + continue + doomed.add(item_id) + changed = True + break + return doomed + + +def settled_failure_dependency_ids(metadata: dict[str, Any] | None) -> set[str]: + """Dependency ids this card was explicitly released over despite failure. + + Includes the ``stuck`` ids (transitively-blocked, still non-terminal): + a released triage card must be claimable even while its stuck + dependencies linger — they are settled context awaiting rebuild or the + settlement cascade, not blockers. Reads the ``dependency_settlement`` + stamp that ``refresh_dependents_for_run`` writes in the same update + that wakes the card, so only the frontier pass — never ad-hoc metadata + edits — can authorize running over an unsatisfied hard dependency. + """ + settlement = dict((metadata or {}).get("dependency_settlement", {}) or {}) + return { + str(item).strip() + for key in ("failed", "cancelled", "stuck") + for item in list(settlement.get(key, []) or []) + if str(item).strip() + } + + def _work_item_kind(item: DelegationWorkItem, metadata: dict[str, Any]) -> str: return str( metadata.get("work_kind") @@ -500,6 +605,120 @@ def _work_item_kind(item: DelegationWorkItem, metadata: dict[str, Any]) -> str: ).strip().lower() +def _dependency_settlement_snapshot( + metadata: dict[str, Any], + dependency_phases: dict[str, Any], + doomed_ids: set[str], +) -> dict[str, Any]: + """Single truth for the failure-triage gate over one card's deps. + + "Settled with failures": every dependency is terminal or doomed + (transitively blocked by a terminal failure), so waiting longer cannot + change the outcome. The failure may be purely transitive (a direct dep + is stuck behind a FAILED card elsewhere), so the trigger counts stuck + deps too. Info-class deps never gate claiming, so they do not gate + settlement either. Missing deps (phase None) stay unsettled. + """ + dependency_class_map = dict(metadata.get("dependency_classes", {}) or {}) + + def _dep_is_info(dep_id: str) -> bool: + return str( + dependency_class_map.get(dep_id, DEPENDENCY_CLASS_DEFAULT) + or DEPENDENCY_CLASS_DEFAULT + ).strip().lower() == "info" + + all_approved = all(p == Phase.APPROVED for p in dependency_phases.values()) + failed = [d for d, p in dependency_phases.items() if p == Phase.FAILED] + cancelled = [d for d, p in dependency_phases.items() if p == Phase.CANCELLED] + stuck = [ + d + for d, p in dependency_phases.items() + if d in doomed_ids and p not in (Phase.FAILED, Phase.CANCELLED) + ] + settled_with_failures = ( + not all_approved + and bool(failed or cancelled or stuck) + and all( + (p is not None and p in DONE_PHASES) + or dep_id in doomed_ids + or _dep_is_info(dep_id) + for dep_id, p in dependency_phases.items() + ) + ) + return { + "all_approved": all_approved, + "failed": failed, + "cancelled": cancelled, + "stuck": stuck, + "settled_with_failures": settled_with_failures, + } + + +def _is_settlement_release_candidate( + item: DelegationWorkItem | Any, + metadata: dict[str, Any], +) -> bool: + """Cards a failure-triage release may wake: the delegating parent, a + roll-up card, or an already-released (stamped) in-flight triage card.""" + phase = getattr(item, "phase", None) + if phase == Phase.WAITING_FOR_CHILDREN: + return True + if phase == Phase.WAITING_DEPENDENCIES and _work_item_kind(item, metadata) in _SETTLEMENT_ROLLUP_KINDS: + return True + return phase in (Phase.READY, Phase.READY_FOR_REWORK, Phase.RUNNING) and bool( + dict(metadata.get("dependency_settlement", {}) or {}) + ) + + +def has_pending_settlement_release( + work_item_by_id: dict[str, DelegationWorkItem | Any], +) -> bool: + """True when some card is due a failure-triage release. + + Used by the dispatcher tick for cards created AFTER their dependency + already failed: the failure's transition hook predates the card, so no + future event would ever run the frontier for it. Cards already + released (stamp present, phase moved) return False here, keeping the + tick idempotent. + """ + doomed_ids = compute_doomed_work_item_ids(work_item_by_id) + if not doomed_ids: + return False + for item_id, item in work_item_by_id.items(): + metadata = dict(getattr(item, "metadata", {}) or {}) + phase = getattr(item, "phase", None) + if not ( + phase == Phase.WAITING_FOR_CHILDREN + or ( + phase == Phase.WAITING_DEPENDENCIES + and _work_item_kind(item, metadata) in _SETTLEMENT_ROLLUP_KINDS + ) + ): + continue + raw_ids = [ + str(dep).strip() + for dep in list(metadata.get("dependency_work_item_ids", []) or []) + if str(dep).strip() + ] + if not raw_ids: + continue + dependency_ids, _pruned = normalize_dependency_work_item_ids( + raw_ids, work_item_by_id, owner_work_item_id=item_id + ) + dependency_phases = { + dep_id: ( + getattr(work_item_by_id[dep_id], "phase", None) + if dep_id in work_item_by_id + else None + ) + for dep_id in dependency_ids + } + snapshot = _dependency_settlement_snapshot(metadata, dependency_phases, doomed_ids) + if snapshot["settled_with_failures"]: + return True + return False + + def _should_enter_synthesis_turn( item: DelegationWorkItem, metadata: dict[str, Any], @@ -534,6 +753,30 @@ def _synthesis_turn_summary(item: DelegationWorkItem, dependency_ids: list[str]) ) +def _failure_triage_turn_summary( + item: DelegationWorkItem, + dependency_phases: dict[str, Any], + failed_ids: list[str], + cancelled_ids: list[str], + stuck_ids: list[str], +) -> str: + title = str(item.title or "delegated work").strip() + manager_label = str(item.manager_role_id or "the upstream owner").strip() + approved_count = sum(1 for p in dependency_phases.values() if p == Phase.APPROVED) + parts = [ + f"{approved_count} approved", + f"{len(failed_ids) + len(cancelled_ids)} failed/cancelled", + ] + if stuck_ids: + parts.append(f"{len(stuck_ids)} blocked downstream") + return ( + f"Triage the delegated results for `{title}` ({', '.join(parts)}). " + "Decide how to handle the failures — rebuild the failed work, accept " + "partial results, or escalate the gap — then prepare the handoff for " + f"{manager_label}." + ) + + async def refresh_dependents_for_run( store: Any, *, @@ -554,6 +797,13 @@ async def refresh_dependents_for_run( children are all approved; otherwise ``WAITING_FOR_CHILDREN → RUNNING``. Both paths release the parent's stale claim so the dispatcher can re-pick it cleanly. + - Failure-triage release: when every dep is settled (terminal) or + doomed (transitively blocked by a FAILED/CANCELLED dep) and at least + one failed, the delegating parent / roll-up card is released anyway + with a ``dependency_settlement`` stamp, so a single failure can + never wedge the whole tree in WAITING_FOR_CHILDREN forever. Once + that card reaches APPROVED, leftover doomed descendants it did not + rebuild are cancelled (settlement cascade) so the run can finalize. - Reverse direction: a RUNNING item whose deps regress (new dep appeared) goes to ``WAITING_FOR_CHILDREN``; a READY item to ``WAITING_DEPENDENCIES``. @@ -590,6 +840,7 @@ async def refresh_dependents_for_run( ) return False work_item_by_id = {item.work_item_id: item for item in work_items} + doomed_ids = compute_doomed_work_item_ids(work_item_by_id) changed = False for work_item in work_items: metadata = dict(work_item.metadata or {}) @@ -609,7 +860,15 @@ async def refresh_dependents_for_run( dep_id: (work_item_by_id[dep_id].phase if dep_id in work_item_by_id else None) for dep_id in dependency_ids } - all_approved = all(p == Phase.APPROVED for p in dependency_phases.values()) + settlement_snapshot = _dependency_settlement_snapshot( + metadata, dependency_phases, doomed_ids + ) + all_approved = settlement_snapshot["all_approved"] + failed_dep_ids = settlement_snapshot["failed"] + cancelled_dep_ids = settlement_snapshot["cancelled"] + stuck_dep_ids = settlement_snapshot["stuck"] + all_settled_with_failures = settlement_snapshot["settled_with_failures"] + settlement_release = False target_phase = work_item.phase metadata_updates: dict[str, Any] = {} summary_update: str | None = None @@ -677,6 +936,86 @@ async def refresh_dependents_for_run( metadata_updates["delegated_children_pending"] = False if str(metadata.get("frontier", "") or "") == "waiting_for_children" and not entered_synthesis_turn: metadata_updates["frontier"] = "resumed" + elif all_settled_with_failures and _is_settlement_release_candidate( + work_item, metadata + ): + # Failure-triage release: a FAILED/CANCELLED child must not + # wedge the whole tree forever. Wake the card that owns the + # decision (the delegating parent, or a roll-up card) and put + # the failure context in the SAME write, so the released turn + # cannot silently succeed without seeing it. Ordinary sibling + # cards blocked on the failure are NOT released — they are in + # ``stuck`` and get rebuilt, rewired, or cancelled by the + # settlement cascade once the triage card completes. + settlement_release = True + if work_item.phase in (Phase.READY, Phase.READY_FOR_REWORK, Phase.RUNNING): + # Already released over this settlement: leave it + # untouched. Regressing it back to a waiting phase (the + # generic not-all-approved branch below) would oscillate + # a released triage card straight back into the deadlock. + pass + elif work_item.phase == Phase.WAITING_FOR_CHILDREN: + if _should_enter_synthesis_turn(work_item, metadata, dependency_ids): + entered_synthesis_turn = True + target_phase = Phase.READY + previous_kind = _work_item_kind(work_item, metadata) + metadata_updates.update( + { + "pre_synthesis_work_kind": previous_kind, + "work_kind": "synthesize", + "delegation_turn_kind": "synthesize", + **work_item_identity_payload( + projection_id=str(work_item.projection_id or work_item.work_item_id or ""), + turn_type="aggregate", + ), + "current_turn_mode": "synthesize_required", + "synthesis_turn_started": True, + "synthesis_ready_at": datetime.now().isoformat(), + "synthesis_source_work_item_ids": list(dependency_ids), + "synthesis_reports_to_role_id": str(work_item.manager_role_id or "").strip(), + "synthesis_reports_to_seat_id": str(work_item.manager_seat_id or "").strip(), + "needs_manager_attention": False, + } + ) + else: + target_phase = Phase.RUNNING + if _work_item_kind(work_item, metadata) in {"deliver", "delivery"}: + metadata_updates.update( + { + "work_kind": "delivery", + "delegation_turn_kind": "delivery", + **work_item_identity_payload( + projection_id=str(work_item.projection_id or work_item.work_item_id or ""), + turn_type="deliver", + ), + "current_turn_mode": "deliver_required", + "delivery_turn_ready_at": datetime.now().isoformat(), + } + ) + else: + target_phase = ( + Phase.READY_FOR_REWORK + if str(metadata.get("rework_feedback", "") or "").strip() + else Phase.READY + ) + if target_phase != work_item.phase: + summary_update = _failure_triage_turn_summary( + work_item, + dependency_phases, + failed_dep_ids, + cancelled_dep_ids, + stuck_dep_ids, + ) + metadata_updates["dependency_settlement"] = { + "failed": list(failed_dep_ids), + "cancelled": list(cancelled_dep_ids), + "stuck": list(stuck_dep_ids), + "settled_at": datetime.now().isoformat(), + } + metadata_updates["frontier"] = "settlement_ready" + metadata_updates["waiting_on_work_item_ids"] = [] + if metadata.get("delegated_children_pending"): + metadata_updates["delegated_children_pending"] = False else: if work_item.phase == Phase.READY: target_phase = Phase.WAITING_DEPENDENCIES @@ -703,7 +1042,7 @@ async def refresh_dependents_for_run( await store.update_delegation_work_item( work_item.work_item_id, phase=target_phase if target_phase != work_item.phase else None, - blocked_reason="" if all_approved else None, + blocked_reason="" if (all_approved or settlement_release) else None, metadata_updates=metadata_updates or None, summary=summary_update, claimed_by_role_runtime_session_id="" if clear_claim_on_wake else None, @@ -715,6 +1054,108 @@ async def refresh_dependents_for_run( "refresh_dependents_for_run: update_delegation_work_item failed " f"wid={work_item.work_item_id}" ) + # Settlement cascade: once a failure-triage card reaches APPROVED + # (auto-approved synthesis or human-approved delivery), any stuck + # descendants it chose not to rebuild are dead branches — cancel + # them so the run reaches a fully-terminal state and can finalize. + # Children the manager rebuilt or rewired (delete_work_item + + # replacement ids) have dropped out of the doomed set and survive. + for work_item in work_items: + metadata = dict(work_item.metadata or {}) + settlement = dict(metadata.get("dependency_settlement", {}) or {}) + if not settlement or settlement.get("cascaded_at"): + continue + if work_item.phase != Phase.APPROVED: + continue + # Transitive closure through the doomed set: cancelling a stuck + # child kills anything hard-chained onto it, and those deeper + # nodes appear in no released card's direct dependency list — + # without the closure they would linger non-terminal forever. + # `covered` seeds from ALL stamped stuck ids regardless of + # phase: a stuck child already cancelled by a previous + # (partially failed) cascade attempt must still conduct the + # traversal, or the deeper nodes behind it become unreachable + # on retry. Growth only follows edges INTO the covered set, so + # doomed subtrees owned by a different (not-yet-approved) + # triage card are left for that card's own decision. + covered_ids: set[str] = { + str(item).strip() + for item in list(settlement.get("stuck", []) or []) + if str(item).strip() and str(item).strip() in work_item_by_id + } + grew = bool(covered_ids) + while grew: + grew = False + for item_id, item in work_item_by_id.items(): + if item_id in covered_ids or item_id not in doomed_ids: + continue + item_metadata = dict(getattr(item, "metadata", {}) or {}) + raw_ids = [ + str(dep).strip() + for dep in list(item_metadata.get("dependency_work_item_ids", []) or []) + if str(dep).strip() + ] + if not raw_ids: + continue + dep_ids, _pruned = normalize_dependency_work_item_ids( + raw_ids, work_item_by_id, owner_work_item_id=item_id + ) + item_classes = dict(item_metadata.get("dependency_classes", {}) or {}) + for dep_id in dep_ids: + if dep_id not in covered_ids: + continue + dep_class = str( + item_classes.get(dep_id, DEPENDENCY_CLASS_DEFAULT) + or DEPENDENCY_CLASS_DEFAULT + ).strip().lower() + if dep_class in ("soft", "info"): + continue + covered_ids.add(item_id) + grew = True + break + cancel_ids = { + covered_id + for covered_id in covered_ids + if covered_id in doomed_ids + and getattr(work_item_by_id.get(covered_id), "phase", None) + not in DONE_PHASES + } + cascade_complete = True + for cancel_id in sorted(cancel_ids): + try: + await transition_work_item( + store, + cancel_id, + target_phase=Phase.CANCELLED, + reason="upstream_dependency_failed_parent_settled", + release_claim=True, + ) + changed = True + except Exception: + cascade_complete = False + logger.opt(exception=True).debug( + "refresh_dependents_for_run: settlement cascade cancel failed " + f"wid={cancel_id}" + ) + if not cascade_complete: + # Leave cascaded_at unset so the next refresh retries the + # leftover cancels instead of permanently orphaning them. + continue + try: + await store.update_delegation_work_item( + work_item.work_item_id, + metadata_updates={ + "dependency_settlement": { + **settlement, + "cascaded_at": datetime.now().isoformat(), + } + }, + ) + except Exception: + logger.opt(exception=True).debug( + "refresh_dependents_for_run: settlement cascade stamp failed " + f"wid={work_item.work_item_id}" + ) if changed and hasattr(store, "save_delegation_event"): try: await store.save_delegation_event( diff --git a/tests/test_actor_runtime_company_mode.py b/tests/test_actor_runtime_company_mode.py index 5cb2cae..316a0f0 100644 --- a/tests/test_actor_runtime_company_mode.py +++ b/tests/test_actor_runtime_company_mode.py @@ -11,6 +11,7 @@ from opc.core.events import EventBus from opc.core.models import CompanyMemberSession, DelegationWorkItem, Phase, SeatState, Task, TaskResult, TaskStatus from opc.database.store import OPCStore from opc.layer2_organization.communication import CommunicationManager +from opc.layer2_organization.phase import DONE_PHASES from opc.layer2_organization.company_mode import CompanyWorkItemExecutor from opc.layer2_organization.company_runtime import CompanyRuntime from opc.layer2_organization.org_engine import OrgEngine @@ -837,6 +838,77 @@ class ActorRuntimeManagerDispatchGuardTests(unittest.IsolatedAsyncioTestCase): "This turn is a root-only scoping decision before any downstream split.", ) + async def test_manager_dispatch_guard_accepts_markdown_decorated_justification(self) -> None: + """Project-4444 regression: the CTO wrote the escape line as + `**NO_DELEGATION_JUSTIFICATION**:` and the bare startswith parser + missed it, driving the work item to FAILED. Decorated variants must + all parse.""" + variants = [ + ("**NO_DELEGATION_JUSTIFICATION**: The task needs my own research.", + "The task needs my own research."), + ("**NO_DELEGATION_JUSTIFICATION: Whole line is bold.**", + "Whole line is bold."), + ("- NO_DELEGATION_JUSTIFICATION: Listed as a bullet.", + "Listed as a bullet."), + ("### NO_DELEGATION_JUSTIFICATION: Written as a heading.", + "Written as a heading."), + ("NO_DELEGATION_JUSTIFICATION： Fullwidth colon variant.", + "Fullwidth colon variant."), + ] + for content, expected in variants: + with self.subTest(content=content): + self.task.metadata.pop("manager_no_delegation_justification", None) + before = await self.executor._snapshot_manager_dispatch_state(self.task) + issues = await self.executor._enforce_manager_dispatch_guard( + self.task, + TaskResult(status=TaskStatus.DONE, content=f"Some preamble.\n{content}"), + before_state=before, + ) + self.assertEqual(issues, []) + self.assertEqual( + self.task.metadata["manager_no_delegation_justification"], + expected, + ) + + async def test_manager_dispatch_guard_ignores_echoed_instruction_template(self) -> None: + """Echoing the guard's own `NO_DELEGATION_JUSTIFICATION: ` template is not a justification — including with trailing + punctuation or quoting around the placeholder.""" + echoes = [ + "I should finish with `NO_DELEGATION_JUSTIFICATION: ` next time.", + "NO_DELEGATION_JUSTIFICATION: .", + 'NO_DELEGATION_JUSTIFICATION: ""', + "NO_DELEGATION_JUSTIFICATION: ****", + "NO_DELEGATION_JUSTIFICATION: __", + ] + for content in echoes: + with self.subTest(content=content): + self.task.metadata.pop("manager_no_delegation_justification", None) + before = await self.executor._snapshot_manager_dispatch_state(self.task) + issues = await self.executor._enforce_manager_dispatch_guard( + self.task, + TaskResult(status=TaskStatus.DONE, content=content), + before_state=before, + ) + self.assertEqual(len(issues), 1) + self.assertNotIn("manager_no_delegation_justification", self.task.metadata) + + async def test_manager_dispatch_guard_ignores_placeholder_in_artifacts(self) -> None: + """The artifact/metadata escape hatch gets the same placeholder + filter as the free-text line.""" + before = await self.executor._snapshot_manager_dispatch_state(self.task) + issues = await self.executor._enforce_manager_dispatch_guard( + self.task, + TaskResult( + status=TaskStatus.DONE, + content="Done.", + artifacts={"no_delegation_justification": "."}, + ), + before_state=before, + ) + self.assertEqual(len(issues), 1) + self.assertNotIn("manager_no_delegation_justification", self.task.metadata) + async def test_manager_dispatch_guard_rejects_no_delegation_for_collab_infra_failure(self) -> None: before = await self.executor._snapshot_manager_dispatch_state(self.task) @@ -941,6 +1013,256 @@ class ActorRuntimeManagerDispatchGuardTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(issues, []) + async def test_manager_dispatch_guard_exhaustion_accepts_turn_instead_of_failing(self) -> None: + # Dispatch is a soft constraint: when the guard reminders run out, + # the turn output is accepted as normal completion (annotated via + # `manager_dispatch_guard_unresolved`) instead of flipping the work + # item to FAILED — not every task needs every seat to delegate. + # No lifecycle helpers are mocked: the whole turn loop runs against + # the real store, so the review chain must actually materialize. + task = await self._make_cto_dispatch_task( + metadata_extra={ + "manager_dispatch_guard_max_retries": 1, + "direct_report_role_ids": ["dev"], + "direct_report_seat_ids": ["seat::team::cto::dev"], + # Stale flag from an imaginary earlier delegating turn: the + # per-turn reset must clear it, otherwise the guard would + # silently pass without reminders. + "manager_board_mutation_performed": True, + }, + ) + task.status = TaskStatus.PENDING + self.executor.execute_task = AsyncMock( + side_effect=[ + TaskResult(status=TaskStatus.DONE, content="Scoped and handled the work directly."), + TaskResult(status=TaskStatus.DONE, content="Still handled directly; no delegation needed."), + ] + ) + + result = await self.executor._run_work_item(task, {}) + + self.assertIsNotNone(result) + self.assertNotEqual(result.status, TaskStatus.FAILED) + # The reminder loop still ran once before acceptance (proves the + # stale mutation flag was reset at turn start). + self.assertEqual(self.executor.execute_task.await_count, 2) + self.assertIn( + "delegate_work", + str(task.metadata.get("manager_dispatch_guard_unresolved", "")), + ) + # Self-produced manager output goes through manager review: the + # card waits for CEO review and a live report card drives it. + work_item = await self.store.get_delegation_work_item("cto-dispatch-item") + self.assertEqual(work_item.phase, Phase.AWAITING_MANAGER_REVIEW) + self.assertEqual( + str((work_item.metadata or {}).get("turn_output_kind", "")), "self_produced" + ) + self.assertEqual( + str((work_item.metadata or {}).get("turn_output_source", "")), + "dispatch_guard_exhausted", + ) + report_cards = await self._aux_cards_targeting("cto-dispatch-item", "report_target_work_item_id") + self.assertEqual(len(report_cards), 1) + self.assertNotIn(report_cards[0].phase, DONE_PHASES) + + async def _aux_cards_targeting(self, work_item_id: str, key: str) -> list[DelegationWorkItem]: + run_items = await self.store.list_delegation_work_items("run-1") + return [ + item for item in run_items + if str((item.metadata or {}).get(key, "") or "").strip() == work_item_id + ] + + async def _make_cto_dispatch_task(self, *, metadata_extra: dict | None = None) -> Task: + task = Task( + id="cto-dispatch-task", + title="CTO Dispatch", + project_id="proj1", + assigned_to="cto", + status=TaskStatus.DONE, + metadata={ + "execution_mode": "company_mode", + "runtime_model": "multi_team_org", + "delegation_run_id": "run-1", + "delegation_seat_id": "seat::team::ceo::cto", + "current_turn_mode": "dispatch_required", + "work_kind": "dispatch", + "manager_role_id": "ceo", + "manager_seat_id": "seat::team::ceo::ceo", + **(metadata_extra or {}), + }, + ) + set_linked_work_item_id(task, "cto-dispatch-item") + await self.store.save_delegation_work_item( + DelegationWorkItem( + work_item_id="cto-dispatch-item", + run_id="run-1", + cell_id="team::ceo", + team_instance_id="team-instance::run-1::team::ceo", + team_id="team::ceo", + role_id="cto", + seat_id="seat::team::ceo::cto", + seat_state_id="seat-state::run-1::seat::team::ceo::cto", + role_runtime_session_id="role-runtime::run-1::seat::team::ceo::cto", + title="CTO Dispatch", + summary="Route the engineering work.", + kind="dispatch", + projection_id="cto-dispatch-item", + phase=Phase.RUNNING, + manager_role_id="ceo", + manager_seat_id="seat::team::ceo::ceo", + metadata={"work_item_runtime": True, "runtime_model": "multi_team_org"}, + ) + ) + return task + + async def test_self_produced_dispatch_runs_full_report_review_chain(self) -> None: + # Real store, no lifecycle mocks: justified self-produced dispatch + # output must route to manager review, actually spawn the report + # card, and — once the report turn finishes — actually spawn the + # review card in the manager seat. + task = await self._make_cto_dispatch_task( + metadata_extra={"manager_no_delegation_justification": "single-seat scoping decision"}, + ) + + phase = await self.executor._apply_done_transition( + task, result=TaskResult(status=TaskStatus.DONE, content="Scoped the work; no delegation needed."), + ) + + self.assertEqual(phase, Phase.AWAITING_MANAGER_REVIEW) + work_item = await self.store.get_delegation_work_item("cto-dispatch-item") + self.assertEqual(work_item.phase, Phase.AWAITING_MANAGER_REVIEW) + self.assertEqual( + str((work_item.metadata or {}).get("turn_output_source", "")), "justified" + ) + report_cards = await self._aux_cards_targeting("cto-dispatch-item", "report_target_work_item_id") + self.assertEqual(len(report_cards), 1) + report_card = report_cards[0] + self.assertNotIn(report_card.phase, DONE_PHASES) + + # Drive the report turn to completion — the review card must appear. + report_task = Task( + id="cto-report-task", + title=report_card.title, + project_id="proj1", + assigned_to="cto", + status=TaskStatus.DONE, + metadata={ + **dict(report_card.metadata or {}), + "execution_mode": "company_mode", + "delegation_run_id": "run-1", + "delegation_seat_id": "seat::team::ceo::cto", + "manager_role_id": "ceo", + "manager_seat_id": "seat::team::ceo::ceo", + }, + ) + set_linked_work_item_id(report_task, report_card.work_item_id) + await self.executor._apply_done_transition( + report_task, + result=TaskResult(status=TaskStatus.DONE, content="Structured handoff report."), + ) + + review_cards = await self._aux_cards_targeting("cto-dispatch-item", "review_target_work_item_id") + self.assertEqual(len(review_cards), 1) + self.assertNotIn(review_cards[0].phase, DONE_PHASES) + self.assertEqual(str(review_cards[0].role_id or ""), "ceo") + + async def test_dispatch_turn_that_delegated_keeps_auto_approve(self) -> None: + # Normal delegation flow: a live child card exists in the store, so + # the dispatch exemption applies (children carry the reviewable + # output) — classification comes from store ground truth, not from + # transient task markers. + task = await self._make_cto_dispatch_task() + await self.store.save_delegation_work_item( + DelegationWorkItem( + work_item_id="cto-delegated-child", + run_id="run-1", + cell_id="team::ceo", + team_instance_id="team-instance::run-1::team::ceo", + team_id="team::ceo", + role_id="cto", + seat_id="seat::team::ceo::cto", + seat_state_id="seat-state::run-1::seat::team::ceo::cto", + role_runtime_session_id="role-runtime::run-1::seat::team::ceo::cto", + parent_work_item_id="cto-dispatch-item", + title="Delegated child", + summary="Build the feature.", + kind="execute", + projection_id="cto-delegated-child", + phase=Phase.READY, + manager_role_id="cto", + manager_seat_id="seat::team::ceo::cto", + metadata={"work_item_runtime": True, "runtime_model": "multi_team_org"}, + ) + ) + + phase = await self.executor._apply_done_transition( + task, result=TaskResult(status=TaskStatus.DONE, content="Delegated to the team."), + ) + + self.assertEqual(phase, Phase.APPROVED) + work_item = await self.store.get_delegation_work_item("cto-dispatch-item") + self.assertEqual( + str((work_item.metadata or {}).get("turn_output_kind", "")), "delegated" + ) + report_cards = await self._aux_cards_targeting("cto-dispatch-item", "report_target_work_item_id") + self.assertEqual(report_cards, []) + + async def test_self_produced_intake_routes_to_manager_review(self) -> None: + # The delegation-output rule covers every review-exempt delegation + # kind, not just dispatch — an intake turn that answered the work + # itself needs review too. + task = await self._make_cto_dispatch_task(metadata_extra={"work_kind": "intake"}) + + phase = await self.executor._apply_done_transition( + task, result=TaskResult(status=TaskStatus.DONE, content="Handled the intake question directly."), + ) + + self.assertEqual(phase, Phase.AWAITING_MANAGER_REVIEW) + work_item = await self.store.get_delegation_work_item("cto-dispatch-item") + self.assertEqual( + str((work_item.metadata or {}).get("turn_output_kind", "")), "self_produced" + ) + + async def test_top_seat_self_produced_falls_back_to_auto_approve(self) -> None: + # The CEO has no manager to review: the existing no-reviewer + # fallback auto-approves instead of stranding the card. + self.task.status = TaskStatus.DONE + self.task.metadata["work_kind"] = "dispatch" + self.task.metadata["manager_dispatch_guard_unresolved"] = "no children" + await self.store.update_delegation_work_item("ceo-work-item", phase=Phase.RUNNING) + + phase = await self.executor._apply_done_transition( + self.task, result=TaskResult(status=TaskStatus.DONE, content="Handled at the top."), + ) + + self.assertEqual(phase, Phase.APPROVED) + + async def test_reconcile_rebuilds_missing_report_card(self) -> None: + # Legacy shape: card already parked in AWAITING_MANAGER_REVIEW with + # the self_produced marker but no live report/review card (the + # historical spawn refused dispatch parents). The dispatcher-tick + # reconcile must rebuild the report card idempotently. + await self._make_cto_dispatch_task() + await self.store.update_delegation_work_item( + "cto-dispatch-item", + phase=Phase.AWAITING_MANAGER_REVIEW, + metadata_updates={ + "turn_output_kind": "self_produced", + "turn_output_source": "dispatch_guard_exhausted", + }, + ) + run_items = await self.store.list_delegation_work_items("run-1") + + run_items = await self.executor._reconcile_missing_review_chain(run_items) + + report_cards = await self._aux_cards_targeting("cto-dispatch-item", "report_target_work_item_id") + self.assertEqual(len(report_cards), 1) + # Idempotent: a second pass with the live report card present must + # not spawn another attempt. + run_items = await self.executor._reconcile_missing_review_chain(run_items) + report_cards = await self._aux_cards_targeting("cto-dispatch-item", "report_target_work_item_id") + self.assertEqual(len(report_cards), 1) + class CompanyModeParallelIsolationTests(unittest.IsolatedAsyncioTestCase): async def test_execute_multi_team_org_isolates_claimed_work_item_exception(self) -> None: diff --git a/tests/test_company_runtime_suspend_resume.py b/tests/test_company_runtime_suspend_resume.py index 3b95a97..f308fd9 100644 --- a/tests/test_company_runtime_suspend_resume.py +++ b/tests/test_company_runtime_suspend_resume.py @@ -1301,6 +1301,108 @@ class CompanyRuntimeSuspendResumeTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(refreshed_item.metadata.get("dispatch_hold"), "") self.assertEqual(resumed_task.status, TaskStatus.BLOCKED) + async def test_continue_keeps_settlement_released_triage_card_running(self) -> None: + """Stop/Resume must not re-lock a failure-triage card the frontier + pass already released over a FAILED dependency: the failed dep is + terminal, so after a regression no event would ever wake it again.""" + store = await self._store() + _, task = await self._seed_runtime(store) + dependency = DelegationWorkItem( + work_item_id="dep-item", + run_id="run-1", + role_id="designer", + seat_id="seat-2", + title="Dependency", + projection_id="dependency", + phase=Phase.FAILED, + metadata={"runtime_model": "multi_team_org"}, + ) + await store.save_delegation_work_item(dependency) + item = await store.get_delegation_work_item("work-item-1") + assert item is not None + item.metadata = { + **dict(item.metadata or {}), + "runtime_model": "multi_team_org", + "dependency_work_item_ids": ["dep-item"], + "dependency_classes": {"dep-item": "hard"}, + "dependency_settlement": { + "failed": ["dep-item"], + "cancelled": [], + "stuck": [], + "settled_at": "2026-07-13T00:00:00", + }, + } + await store.save_delegation_work_item(item) + engine = self._engine(store) + await engine.suspend_company_runtime( + origin_task_id=task.id, + session_id="sess-parent", + reason="user_stop", + ) + captured: dict[str, Any] = {} + + class DummyCompanyExecutor: + async def execute(self, plan: CompanyWorkItemRuntimePlan, tasks: list[Task]) -> str: + captured["tasks"] = tasks + 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) + + def test_resume_dependency_check_honors_settlement_for_soft_stuck_deps(self) -> None: + """The soft-class branch must honor the settlement stamp too: a + soft dep parked in WAITING_DEPENDENCIES (not in-progress, not + terminal) that the frontier stamped as stuck must not re-lock the + released card on resume.""" + stuck_dep = DelegationWorkItem( + work_item_id="soft-dep", + run_id="run-soft", + role_id="w", + seat_id="seat-w", + title="Stuck soft dep", + projection_id="soft-dep", + phase=Phase.WAITING_DEPENDENCIES, + ) + card = DelegationWorkItem( + work_item_id="soft-card", + run_id="run-soft", + role_id="m", + seat_id="seat-m", + title="Released triage card", + projection_id="soft-card", + phase=Phase.RUNNING, + metadata={ + "dependency_work_item_ids": ["soft-dep"], + "dependency_classes": {"soft-dep": "soft"}, + "dependency_settlement": { + "failed": [], + "cancelled": [], + "stuck": ["soft-dep"], + "settled_at": "2026-07-13T00:00:00", + }, + }, + ) + by_id = {"soft-dep": stuck_dep, "soft-card": card} + self.assertTrue( + OPCEngine._company_runtime_dependencies_satisfied(card, by_id) + ) + card.metadata = { + **dict(card.metadata or {}), + "dependency_settlement": {}, + } + self.assertFalse( + OPCEngine._company_runtime_dependencies_satisfied(card, by_id) + ) + async def test_continue_does_not_use_synthetic_external_session_id_as_provider_token(self) -> None: store = await self._store() _, task = await self._seed_runtime( diff --git a/tests/test_review_verdict_and_dep_refresh.py b/tests/test_review_verdict_and_dep_refresh.py index 08fe982..1ed9cab 100644 --- a/tests/test_review_verdict_and_dep_refresh.py +++ b/tests/test_review_verdict_and_dep_refresh.py @@ -333,8 +333,10 @@ class RefreshDependentsForRunTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(after.metadata.get("waiting_on_work_item_ids"), []) async def test_child_cancelled_triggers_parent_refresh(self) -> None: - """Fix 3 core: non-APPROVED terminal (CANCELLED) must still fire - the refresh hook. Before Fix 3, only APPROVED did.""" + """Fix 3 core + failure-triage release: a non-APPROVED terminal + (CANCELLED) fires the refresh hook, and because every dependency is + now settled the parent is RELEASED for a triage turn instead of + waiting on the dead child forever (the project-4444 deadlock).""" parent = _make_work_item( work_item_id="parent-c", run_id="run-b", @@ -357,14 +359,534 @@ class RefreshDependentsForRunTests(unittest.IsolatedAsyncioTestCase): await self.store.update_delegation_work_item( "child-b", phase=Phase.CANCELLED ) - # Parent still WAITING_FOR_CHILDREN (not all approved), but the - # hook ran — verify by checking waiting_on_work_item_ids. after = await self.store.get_delegation_work_item("parent-c") + self.assertEqual(after.phase, Phase.RUNNING) + self.assertEqual(after.claimed_by_role_runtime_session_id, "") + settlement = dict(after.metadata.get("dependency_settlement", {}) or {}) + self.assertEqual(list(settlement.get("cancelled", [])), ["child-b"]) + self.assertEqual(list(settlement.get("failed", [])), []) + self.assertEqual(list(after.metadata.get("waiting_on_work_item_ids", [])), []) + + async def test_failed_child_releases_parent_for_triage_synthesis(self) -> None: + """Project-4444 regression: a FAILED child must release the + delegating parent into a synthesis/triage turn with the failure + stamped, not pin it in WAITING_FOR_CHILDREN forever.""" + parent = _make_work_item( + work_item_id="parent-f", + run_id="run-f", + phase=Phase.WAITING_FOR_CHILDREN, + dependency_ids=["child-ok", "child-bad"], + claimed_by="claim-y", + metadata={"delegated_children_pending": True}, + ) + child_ok = _make_work_item( + work_item_id="child-ok", run_id="run-f", phase=Phase.APPROVED + ) + child_bad = _make_work_item( + work_item_id="child-bad", run_id="run-f", phase=Phase.RUNNING + ) + await self._save(parent, child_ok, child_bad) + + await self.store.update_delegation_work_item("child-bad", phase=Phase.FAILED) + + after = await self.store.get_delegation_work_item("parent-f") + self.assertEqual(after.phase, Phase.READY) + self.assertEqual(after.claimed_by_role_runtime_session_id, "") + self.assertEqual(after.metadata.get("work_kind"), "synthesize") + self.assertEqual(after.metadata.get("current_turn_mode"), "synthesize_required") + settlement = dict(after.metadata.get("dependency_settlement", {}) or {}) + self.assertEqual(list(settlement.get("failed", [])), ["child-bad"]) + self.assertIn("failed/cancelled", str(after.summary or "")) + + async def test_doomed_chain_marks_stuck_and_releases_parent(self) -> None: + """Transitive settlement: sibling B hard-depends on FAILED A, so B + can never run. The parent (deps [A, B]) must still be released, + with B recorded as stuck rather than waited on forever.""" + parent = _make_work_item( + work_item_id="parent-chain", + run_id="run-chain", + phase=Phase.WAITING_FOR_CHILDREN, + dependency_ids=["chain-a", "chain-b"], + metadata={"runtime_model": "multi_team_org"}, + ) + chain_a = _make_work_item( + work_item_id="chain-a", run_id="run-chain", phase=Phase.RUNNING + ) + chain_b = _make_work_item( + work_item_id="chain-b", + run_id="run-chain", + phase=Phase.WAITING_DEPENDENCIES, + dependency_ids=["chain-a"], + ) + await self._save(parent, chain_a, chain_b) + + await self.store.update_delegation_work_item("chain-a", phase=Phase.FAILED) + + after_parent = await self.store.get_delegation_work_item("parent-chain") + after_b = await self.store.get_delegation_work_item("chain-b") + self.assertEqual(after_parent.phase, Phase.RUNNING) + settlement = dict(after_parent.metadata.get("dependency_settlement", {}) or {}) + self.assertEqual(list(settlement.get("failed", [])), ["chain-a"]) + self.assertEqual(list(settlement.get("stuck", [])), ["chain-b"]) + # The stuck sibling itself is NOT released — the triage turn decides. + self.assertEqual(after_b.phase, Phase.WAITING_DEPENDENCIES) + # The released parent must actually be claimable despite the + # non-terminal stuck dep, or the release is cosmetic. + run_items = await self.store.list_delegation_work_items("run-chain") + by_id = {item.work_item_id: item for item in run_items} + self.assertTrue( + CompanyWorkItemExecutor._work_item_is_runnable(by_id["parent-chain"], by_id) + ) + + async def test_transitive_only_failure_still_releases_parent(self) -> None: + """`parent → B → FAILED A` where A is NOT a direct dep of the + parent: no direct dep ever turns FAILED, but B is doomed, so the + parent must still be released instead of waiting forever.""" + parent = _make_work_item( + work_item_id="parent-t", + run_id="run-t", + phase=Phase.WAITING_FOR_CHILDREN, + dependency_ids=["t-b"], + metadata={"runtime_model": "multi_team_org"}, + ) + t_b = _make_work_item( + work_item_id="t-b", + run_id="run-t", + phase=Phase.WAITING_DEPENDENCIES, + dependency_ids=["t-a"], + ) + t_a = _make_work_item(work_item_id="t-a", run_id="run-t", phase=Phase.RUNNING) + await self._save(t_b, parent, t_a) + + await self.store.update_delegation_work_item("t-a", phase=Phase.FAILED) + + after_parent = await self.store.get_delegation_work_item("parent-t") + self.assertEqual(after_parent.phase, Phase.RUNNING) + settlement = dict(after_parent.metadata.get("dependency_settlement", {}) or {}) + self.assertEqual(list(settlement.get("failed", [])), []) + self.assertEqual(list(settlement.get("stuck", [])), ["t-b"]) + run_items = await self.store.list_delegation_work_items("run-t") + by_id = {item.work_item_id: item for item in run_items} + self.assertTrue( + CompanyWorkItemExecutor._work_item_is_runnable(by_id["parent-t"], by_id) + ) + + async def test_delivery_rollup_released_over_failed_dependency(self) -> None: + """A delivery card WAITING_DEPENDENCIES on a FAILED child is released + READY so the failure report reaches the user instead of wedging.""" + delivery = _make_work_item( + work_item_id="delivery-f", + run_id="run-dlv", + phase=Phase.WAITING_DEPENDENCIES, + dependency_ids=["dlv-child"], + metadata={"work_kind": "delivery"}, + ) + child = _make_work_item( + work_item_id="dlv-child", run_id="run-dlv", phase=Phase.RUNNING + ) + await self._save(delivery, child) + + await self.store.update_delegation_work_item("dlv-child", phase=Phase.FAILED) + + after = await self.store.get_delegation_work_item("delivery-f") + self.assertEqual(after.phase, Phase.READY) + settlement = dict(after.metadata.get("dependency_settlement", {}) or {}) + self.assertEqual(list(settlement.get("failed", [])), ["dlv-child"]) + + async def test_settlement_cascade_cancels_unrebuilt_stuck_children(self) -> None: + """Once the settled triage card is APPROVED, leftover doomed stuck + children are cancelled so the run can reach a fully-terminal state.""" + parent = _make_work_item( + work_item_id="parent-casc", + run_id="run-casc", + phase=Phase.RUNNING, + dependency_ids=["casc-a", "casc-b"], + metadata={ + "dependency_settlement": { + "failed": ["casc-a"], + "cancelled": [], + "stuck": ["casc-b"], + "settled_at": "2026-07-13T00:00:00", + }, + }, + ) + casc_a = _make_work_item( + work_item_id="casc-a", run_id="run-casc", phase=Phase.FAILED + ) + casc_b = _make_work_item( + work_item_id="casc-b", + run_id="run-casc", + phase=Phase.WAITING_DEPENDENCIES, + dependency_ids=["casc-a"], + ) + await self._save(casc_b, casc_a, parent) + + await self.store.update_delegation_work_item("parent-casc", phase=Phase.APPROVED) + + after_b = await self.store.get_delegation_work_item("casc-b") + after_parent = await self.store.get_delegation_work_item("parent-casc") + self.assertEqual(after_b.phase, Phase.CANCELLED) + self.assertEqual( + after_b.metadata.get("last_transition_reason"), + "upstream_dependency_failed_parent_settled", + ) + settlement = dict(after_parent.metadata.get("dependency_settlement", {}) or {}) + self.assertTrue(settlement.get("cascaded_at")) + + async def test_settlement_cascade_cancels_deep_doomed_chain(self) -> None: + """Closure: C hard-depends on stuck B; C is in no released card's + direct dependency list, but cancelling B dooms it — the cascade + must cancel the whole chain or the run never terminalizes.""" + parent = _make_work_item( + work_item_id="parent-deep", + run_id="run-deep", + phase=Phase.RUNNING, + dependency_ids=["deep-a", "deep-b"], + metadata={ + "dependency_settlement": { + "failed": ["deep-a"], + "cancelled": [], + "stuck": ["deep-b"], + "settled_at": "2026-07-13T00:00:00", + }, + }, + ) + deep_a = _make_work_item( + work_item_id="deep-a", run_id="run-deep", phase=Phase.FAILED + ) + deep_b = _make_work_item( + work_item_id="deep-b", + run_id="run-deep", + phase=Phase.WAITING_DEPENDENCIES, + dependency_ids=["deep-a"], + ) + deep_c = _make_work_item( + work_item_id="deep-c", + run_id="run-deep", + phase=Phase.WAITING_DEPENDENCIES, + dependency_ids=["deep-b"], + ) + await self._save(deep_c, deep_b, deep_a, parent) + + await self.store.update_delegation_work_item("parent-deep", phase=Phase.APPROVED) + + after_b = await self.store.get_delegation_work_item("deep-b") + after_c = await self.store.get_delegation_work_item("deep-c") + self.assertEqual(after_b.phase, Phase.CANCELLED) + self.assertEqual(after_c.phase, Phase.CANCELLED) + + async def test_settlement_cascade_spares_rewired_children(self) -> None: + """Stuck children the manager rewired onto a live replacement stay + alive: rewiring drops them out of the doomed set.""" + parent = _make_work_item( + work_item_id="parent-rw", + run_id="run-rw", + phase=Phase.RUNNING, + dependency_ids=["rw-a", "rw-b"], + metadata={ + "dependency_settlement": { + "failed": ["rw-a"], + "cancelled": [], + "stuck": ["rw-b"], + "settled_at": "2026-07-13T00:00:00", + }, + }, + ) + # Failed child was manager-deleted with a replacement, so rw-b's + # dependency normalizes onto the live replacement card. + rw_a = _make_work_item( + work_item_id="rw-a", + run_id="run-rw", + phase=Phase.CANCELLED, + metadata={ + "deleted_by_manager_tool": True, + "replacement_dependency_work_item_ids": ["rw-a2"], + }, + ) + rw_a2 = _make_work_item( + work_item_id="rw-a2", run_id="run-rw", phase=Phase.RUNNING + ) + rw_b = _make_work_item( + work_item_id="rw-b", + run_id="run-rw", + phase=Phase.WAITING_DEPENDENCIES, + dependency_ids=["rw-a"], + ) + await self._save(rw_a2, rw_b, rw_a, parent) + + await self.store.update_delegation_work_item("parent-rw", phase=Phase.APPROVED) + + after_b = await self.store.get_delegation_work_item("rw-b") + self.assertEqual(after_b.phase, Phase.WAITING_DEPENDENCIES) + + async def test_park_does_not_wait_on_settled_failed_or_doomed_deps(self) -> None: + """A triage turn that accepts partial results must be able to + finish: settled deps (terminal or doomed) are not pending, so the + card completes instead of re-parking forever on the FAILED child + it just triaged.""" + parent = _make_work_item( + work_item_id="park-parent", + run_id="run-park", + phase=Phase.RUNNING, + dependency_ids=["park-ok", "park-bad", "park-stuck"], + metadata={ + "dependency_settlement": { + "failed": ["park-bad"], + "cancelled": [], + "stuck": ["park-stuck"], + "settled_at": "2026-07-13T00:00:00", + }, + }, + ) + park_ok = _make_work_item( + work_item_id="park-ok", run_id="run-park", phase=Phase.APPROVED + ) + park_bad = _make_work_item( + work_item_id="park-bad", run_id="run-park", phase=Phase.FAILED + ) + park_stuck = _make_work_item( + work_item_id="park-stuck", + run_id="run-park", + phase=Phase.WAITING_DEPENDENCIES, + dependency_ids=["park-bad"], + ) + await self._save(park_ok, park_bad, park_stuck, parent) + + task = Task( + id="park-task", + title="Triage turn", + project_id="proj1", + assigned_to="m", + status=TaskStatus.RUNNING, + metadata={"work_item_projection_id": "park-parent"}, + ) + set_linked_work_item_id(task, "park-parent") + await self.store.save_task(task) + + executor = self._executor() + parked = await executor._park_for_delegated_children(task) + + self.assertFalse(parked) + after = await self.store.get_delegation_work_item("park-parent") + self.assertEqual(after.phase, Phase.RUNNING) + # Settlement stamp survives (needed by the cascade at APPROVED). + self.assertTrue(dict(after.metadata.get("dependency_settlement", {}) or {})) + + async def test_park_race_releases_triage_when_failure_beat_the_park(self) -> None: + """Race: children failed while the manager turn was still running, + so the failure hook fired before the card parked (and may have + regressed it to WAITING_FOR_CHILDREN without a stamp). Park must + re-arm the triage release instead of silently returning False.""" + parent = _make_work_item( + work_item_id="race-parent", + run_id="run-race", + phase=Phase.WAITING_FOR_CHILDREN, + dependency_ids=["race-bad"], + ) + race_bad = _make_work_item( + work_item_id="race-bad", run_id="run-race", phase=Phase.FAILED + ) + await self._save(race_bad, parent) + + task = Task( + id="race-task", + title="Dispatch turn", + project_id="proj1", + assigned_to="m", + status=TaskStatus.RUNNING, + metadata={"work_item_projection_id": "race-parent"}, + ) + set_linked_work_item_id(task, "race-parent") + await self.store.save_task(task) + + executor = self._executor() + parked = await executor._park_for_delegated_children(task) + + self.assertTrue(parked) + after = await self.store.get_delegation_work_item("race-parent") + self.assertNotEqual(after.phase, Phase.WAITING_FOR_CHILDREN) + settlement = dict(after.metadata.get("dependency_settlement", {}) or {}) + self.assertEqual(list(settlement.get("failed", [])), ["race-bad"]) + + async def test_released_triage_card_is_not_doomed_for_upper_parents(self) -> None: + """A released (stamped) triage card is alive: it must not count as + doomed, or the upper parent settles early and its cascade could + cancel a triage card that is about to run.""" + triage = _make_work_item( + work_item_id="alive-triage", + run_id="run-alive", + phase=Phase.READY, + dependency_ids=["alive-bad"], + metadata={ + "runtime_model": "multi_team_org", + "work_kind": "synthesize", + "dependency_settlement": { + "failed": ["alive-bad"], + "cancelled": [], + "stuck": [], + "settled_at": "2026-07-13T00:00:00", + }, + }, + ) + alive_bad = _make_work_item( + work_item_id="alive-bad", run_id="run-alive", phase=Phase.FAILED + ) + upper = _make_work_item( + work_item_id="alive-upper", + run_id="run-alive", + phase=Phase.WAITING_FOR_CHILDREN, + dependency_ids=["alive-triage"], + ) + await self._save(alive_bad, triage, upper) + + from opc.layer2_organization.work_item_transition import ( + compute_doomed_work_item_ids, + ) + + run_items = await self.store.list_delegation_work_items("run-alive") + by_id = {item.work_item_id: item for item in run_items} + doomed = compute_doomed_work_item_ids(by_id) + self.assertNotIn("alive-triage", doomed) + + changed = await refresh_dependents_for_run(self.store, run_id="run-alive") + after_upper = await self.store.get_delegation_work_item("alive-upper") + # The upper parent keeps waiting for the live triage card — no + # premature settlement over it. + self.assertEqual(after_upper.phase, Phase.WAITING_FOR_CHILDREN) + self.assertFalse( + dict(after_upper.metadata.get("dependency_settlement", {}) or {}) + ) + + async def test_settlement_cascade_retries_after_partial_failure(self) -> None: + """Retry: B (stamped stuck) was cancelled by a previous cascade + attempt but deeper C failed to cancel. The next pass must still + traverse through terminal B and cancel C before stamping.""" + parent = _make_work_item( + work_item_id="retry-parent", + run_id="run-retry", + phase=Phase.APPROVED, + dependency_ids=["retry-a", "retry-b"], + metadata={ + "dependency_settlement": { + "failed": ["retry-a"], + "cancelled": [], + "stuck": ["retry-b"], + "settled_at": "2026-07-13T00:00:00", + }, + }, + ) + retry_a = _make_work_item( + work_item_id="retry-a", run_id="run-retry", phase=Phase.FAILED + ) + retry_b = _make_work_item( + work_item_id="retry-b", + run_id="run-retry", + phase=Phase.CANCELLED, + dependency_ids=["retry-a"], + ) + retry_c = _make_work_item( + work_item_id="retry-c", + run_id="run-retry", + phase=Phase.WAITING_DEPENDENCIES, + dependency_ids=["retry-b"], + ) + await self._save(retry_c, retry_b, retry_a, parent) + + await refresh_dependents_for_run(self.store, run_id="run-retry") + + after_c = await self.store.get_delegation_work_item("retry-c") + after_parent = await self.store.get_delegation_work_item("retry-parent") + self.assertEqual(after_c.phase, Phase.CANCELLED) + settlement = dict(after_parent.metadata.get("dependency_settlement", {}) or {}) + self.assertTrue(settlement.get("cascaded_at")) + + async def test_info_dependency_does_not_block_settlement(self) -> None: + """Info-class deps never gate claiming, so an in-flight info dep + must not keep a failed board from settling either.""" + parent = _make_work_item( + work_item_id="info-parent", + run_id="run-info", + phase=Phase.WAITING_FOR_CHILDREN, + dependency_ids=["info-bad", "info-fyi"], + metadata={"dependency_classes": {"info-fyi": "info"}}, + ) + info_bad = _make_work_item( + work_item_id="info-bad", run_id="run-info", phase=Phase.RUNNING + ) + info_fyi = _make_work_item( + work_item_id="info-fyi", run_id="run-info", phase=Phase.RUNNING + ) + await self._save(info_fyi, info_bad, parent) + + await self.store.update_delegation_work_item("info-bad", phase=Phase.FAILED) + + after = await self.store.get_delegation_work_item("info-parent") + self.assertNotEqual(after.phase, Phase.WAITING_FOR_CHILDREN) + settlement = dict(after.metadata.get("dependency_settlement", {}) or {}) + self.assertEqual(list(settlement.get("failed", [])), ["info-bad"]) + self.assertNotIn("info-fyi", list(settlement.get("stuck", []))) + + async def test_late_created_rollup_card_gets_settlement_on_dispatcher_tick(self) -> None: + """A delivery card created AFTER its dependency failed sees no + failure hook; the dispatcher tick must run the frontier for it.""" + late_bad = _make_work_item( + work_item_id="late-bad", run_id="run-late", phase=Phase.FAILED + ) + await self._save(late_bad) + delivery = _make_work_item( + work_item_id="late-delivery", + run_id="run-late", + phase=Phase.WAITING_DEPENDENCIES, + dependency_ids=["late-bad"], + metadata={"work_kind": "delivery"}, + ) + await self._save(delivery) + + executor = self._executor() + run_items = await self.store.list_delegation_work_items("run-late") + await executor._refresh_ready_work_items(run_items, tasks=[]) + + after = await self.store.get_delegation_work_item("late-delivery") + self.assertEqual(after.phase, Phase.READY) + settlement = dict(after.metadata.get("dependency_settlement", {}) or {}) + self.assertEqual(list(settlement.get("failed", [])), ["late-bad"]) + + async def test_park_still_waits_on_live_children(self) -> None: + """Control: genuinely in-flight children still park the manager.""" + parent = _make_work_item( + work_item_id="live-parent", + run_id="run-live", + phase=Phase.RUNNING, + dependency_ids=["live-ok", "live-running"], + ) + live_ok = _make_work_item( + work_item_id="live-ok", run_id="run-live", phase=Phase.APPROVED + ) + live_running = _make_work_item( + work_item_id="live-running", run_id="run-live", phase=Phase.RUNNING + ) + await self._save(live_ok, live_running, parent) + + task = Task( + id="live-task", + title="Dispatch turn", + project_id="proj1", + assigned_to="m", + status=TaskStatus.RUNNING, + metadata={"work_item_projection_id": "live-parent"}, + ) + set_linked_work_item_id(task, "live-parent") + await self.store.save_task(task) + + executor = self._executor() + parked = await executor._park_for_delegated_children(task) + + self.assertTrue(parked) + after = await self.store.get_delegation_work_item("live-parent") self.assertEqual(after.phase, Phase.WAITING_FOR_CHILDREN) - # Waiting list should reflect the deps (hook wrote it). self.assertEqual( list(after.metadata.get("waiting_on_work_item_ids", [])), - ["child-a", "child-b"], + ["live-running"], ) async def test_manager_deleted_child_is_pruned_from_parent_dependencies(self) -> None: diff --git a/tests/test_worker_report_handoff.py b/tests/test_worker_report_handoff.py index 7f36fe7..730faa1 100644 --- a/tests/test_worker_report_handoff.py +++ b/tests/test_worker_report_handoff.py @@ -183,12 +183,21 @@ class WorkerExecuteDoneSpawnsReportTests(unittest.IsolatedAsyncioTestCase): org_engine = _make_org_engine(root) executor = _build_executor(store, org_engine) - # CEO dispatch card. work_kind=dispatch routes directly - # to APPROVED — no review, and no report turn. + # Dispatch card that actually delegated (a live child card + # exists in the store). Delegated output routes directly to + # APPROVED — the children carry the reviewable output, so no + # review and no report turn for the dispatch card itself. + # (A dispatch card WITHOUT children is the self-produced + # case and does get the report/review chain.) child = _build_child_work_item() child.metadata = dict(child.metadata or {}) child.metadata["work_kind"] = "dispatch" await store.save_delegation_work_item(child) + delegated = _build_child_work_item() + delegated.work_item_id = "wi-grandchild" + delegated.projection_id = "wi-grandchild" + delegated.parent_work_item_id = "wi-child" + await store.save_delegation_work_item(delegated) worker_task = _build_worker_task() worker_task.metadata = dict(worker_task.metadata or {}) worker_task.metadata["work_kind"] = "dispatch"