diff --git a/opc/engine.py b/opc/engine.py index 9fd1c88..593bbb6 100644 --- a/opc/engine.py +++ b/opc/engine.py @@ -1280,11 +1280,50 @@ class OPCEngine: if str(item.get("template_id", "") or "").strip() } roles: list[dict[str, Any]] = [] - default_agent = "codex" + # The card's per-role agent defaults become recruitment_role_agents on + # approve, which outrank the session-level agent choice as explicit + # per-role overrides. They must therefore be the RESOLVED effective + # backend, not a hardcoded external preference: an explicit + # preferred_agent from the user wins, a role template preference only + # applies when that adapter can actually run, everything else is + # native. (OBS-11: hardcoded "codex" defaults poisoned the execution + # identity of runs the user explicitly requested as native.) + explicit_session_agent = normalize_recruitment_agent_choice( + getattr(decision, "preferred_agent", None) + ) + available_external_agents = set(self._available_external_agents()) + + def _staffing_agent_is_runnable(agent_name: str) -> bool: + # Unknown availability (no adapter registry) fails open, matching + # the dispatch selector; a registry that exists and excludes the + # agent is proof it cannot run. + if agent_name == "native": + return True + if self.adapter_registry is None: + return True + return agent_name in available_external_agents + for agent in self.org_engine.list_agents(): role_id = str(getattr(agent, "role_id", "") or "").strip() if not role_id or role_id == "task_generalist": continue + role_preferred_agent = normalize_recruitment_agent_choice( + getattr(agent, "preferred_external_agent", None) + ) + if explicit_session_agent: + default_agent = ( + explicit_session_agent + if _staffing_agent_is_runnable(explicit_session_agent) + else "native" + ) + elif ( + role_preferred_agent + and role_preferred_agent != "native" + and _staffing_agent_is_runnable(role_preferred_agent) + ): + default_agent = role_preferred_agent + else: + default_agent = "native" same_role_employees = employee_by_role.get(role_id, []) default_selection: dict[str, Any] = {"kind": "fallback"} default_source = "system" @@ -1311,6 +1350,8 @@ class OPCEngine: } default_source = "org" selected_agent = normalize_recruitment_agent_choice(saved_agents.get(role_id), default=default_agent) or default_agent + if not _staffing_agent_is_runnable(selected_agent): + selected_agent = default_agent roles.append( { "role_id": role_id, @@ -2917,6 +2958,7 @@ class OPCEngine: else None ) explicit_force_native = explicit_agent_choice == "native" + enrichment_available_external_agents = set(self._available_external_agents()) seats: list[dict[str, Any]] = [] for raw_seat in list(enriched.get("seats", []) or []): seat = dict(raw_seat or {}) @@ -2972,6 +3014,21 @@ class OPCEngine: or explicit_agent_choice or ("native" if force_native_execution or not preferred_external_agent else preferred_external_agent) ) + execution_agent_unavailable = "" + if ( + selected_execution_agent + and selected_execution_agent != "native" + and self.adapter_registry is not None + and selected_execution_agent not in enrichment_available_external_agents + ): + # The chosen external backend cannot run. Dispatch would fall + # back to native while seat/task metadata kept claiming the + # external agent, and the resume gate trusts that identity + # (OBS-11). Record the truth up front; the wish stays visible + # via execution_agent_unavailable. + execution_agent_unavailable = selected_execution_agent + selected_execution_agent = "native" + preferred_external_agent = None execution_agent_locked = bool(selected_role_agent or explicit_agent_choice) selection_source = ( "recruitment_user_override" @@ -2987,6 +3044,7 @@ class OPCEngine: seat["execution_agent_locked"] = execution_agent_locked seat["selected_execution_agent_source"] = selection_source seat["force_native_execution"] = force_native_execution + seat["execution_agent_unavailable"] = execution_agent_unavailable seat["metadata"] = { **dict(seat.get("metadata", {}) or {}), "employee_prompt_context": str((employee_assignment or {}).get("prompt_context", "")).strip(), @@ -4740,6 +4798,28 @@ class OPCEngine: default=("native" if not str(task.assigned_external_agent or "").strip() else str(task.assigned_external_agent or "").strip()), ) if task.metadata.get("execution_agent_locked") and locked_agent: + if ( + locked_agent != "native" + and self.adapter_registry is not None + and locked_agent not in self._available_external_agents() + ): + # A locked external backend that cannot run must not be + # stamped as execution identity: this attempt actually runs + # native (the external candidate pool is empty) and resume + # trusts this metadata (OBS-11). + task.assigned_external_agent = None + task.metadata["selected_execution_agent"] = "native" + task.metadata["preferred_external_agent"] = None + task.metadata["execution_agent_unavailable"] = locked_agent + task.metadata["agent_selection"] = { + "selected": "native", + "strategy": WorkItemExecutionStrategy.NATIVE.value, + "role_id": task.assigned_to or task.metadata.get("work_item_role_id", ""), + "decision_reason": "locked_external_agent_unavailable_native_fallback", + "available_external_agents": self._available_external_agents(), + "selection_source": "availability_fallback", + } + return None selected = None if locked_agent == "native" else locked_agent task.assigned_external_agent = selected task.metadata["preferred_external_agent"] = selected @@ -6384,12 +6464,59 @@ class OPCEngine: # not fully initialized / delegate context) — fail open and let the # dispatch-time selector guard decide; only a registry that exists # and excludes the pinned agent is proof of unavailability. - if ( + pinned_agent_unavailable = bool( pinned_agent != "native" and not work_item_already_terminal and self.adapter_registry is not None and pinned_agent not in self._available_external_agents() - ): + ) + if pinned_agent_unavailable: + gate_external_session = dict( + (payload.get("external_sessions", {}) or {}).get(task.id) or {} + ) + has_resumable_external_session = bool( + gate_external_session + ) and external_session_status_allows_resume( + gate_external_session.get("status") + ) + if not has_resumable_external_session: + # The pin is a preference, not a fact: this work item has + # no external session to revive, so dispatch would run it + # natively anyway (availability fallback). Heal the + # checkpointed identity to native and resume instead of + # failing the item — a run the user launched as native + # must survive stop/resume even when a disabled external + # agent leaked into its metadata (OBS-11). + healed_identity = dict( + task_snapshot.get("execution_identity", {}) or {} + ) + healed_identity["selected_execution_agent"] = "native" + healed_identity["assigned_external_agent"] = "" + if not str( + healed_identity.get("preferred_external_agent", "") or "" + ).strip(): + healed_identity["preferred_external_agent"] = pinned_agent + task_snapshot["execution_identity"] = healed_identity + task_snapshot["selected_execution_agent"] = "native" + task_snapshot["assigned_external_agent"] = "" + # The task's durable identity must agree with the healed + # snapshot, or the restore validation below rejects the + # resume as an identity mismatch. + task.metadata = dict(task.metadata or {}) + task.metadata["resume_execution_agent_healed_from"] = pinned_agent + task.metadata["selected_execution_agent"] = "native" + task.metadata.pop("agent_selection", None) + task.metadata.pop("preferred_external_agent", None) + task.assigned_external_agent = None + logger.info( + "company runtime resume: healed work item {} to native — pinned " + "external agent {!r} is unavailable and no resumable external " + "session exists", + work_item_id or task.id, + pinned_agent, + ) + pinned_agent_unavailable = False + if pinned_agent_unavailable: diagnostic = ( f"Cannot resume work item: its execution is pinned to external agent " f"'{pinned_agent}', which is currently disabled or unavailable. " @@ -8546,7 +8673,43 @@ class OPCEngine: @staticmethod def _reply_metadata_requests_force_resume(reply_metadata: dict[str, Any] | None) -> bool: - return bool(dict(reply_metadata or {}).get("ui_force_resume", False)) + metadata = dict(reply_metadata or {}) + # ui_force_resume is what the Office UI resume button sends; + # force_resume is the documented key for chat/headless callers. + # Both express the same control intent (OBS-11: only recognizing the + # UI spelling pushed headless resumes into the content-followup path). + return bool( + metadata.get("ui_force_resume", False) + or metadata.get("force_resume", False) + ) + + @staticmethod + def _is_plain_resume_control_reply(user_reply: str) -> bool: + """A bare continuation acknowledgement carries no new instructions. + + Such a reply to a suspended-runtime checkpoint means "resume the run", + never "route this text to the final decider as a follow-up" — routing + it as content reopens the already-approved intake card and churns the + run it was supposed to continue (OBS-11). + """ + normalized = " ".join(str(user_reply or "").strip().lower().split()) + return normalized in { + "", + "continue", + "resume", + "proceed", + "go", + "go on", + "ok", + "okay", + "y", + "yes", + # Chinese spellings of the same continuation intent. + "继续", + "恢复", + "继续执行", + "继续跑", + } async def _maybe_resume_existing_company_runtime( self, @@ -8579,6 +8742,28 @@ class OPCEngine: blocked_tasks = [task for task in tasks if task.status == TaskStatus.BLOCKED] no_active_runtime_work = not live_running_tasks and not waiting_tasks and not pending_tasks and not failed_tasks and not blocked_tasks has_closed_delivery_review = any(self._is_closed_company_delivery_review_task(task) for task in tasks) + run_terminally_failed = ( + bool(failed_tasks) + and not has_closed_delivery_review + and all( + t.status in {TaskStatus.DONE, TaskStatus.CANCELLED, TaskStatus.FAILED} + for t in tasks + ) + ) + if run_terminally_failed: + # The run is a corpse: re-executing it drops the user's new + # content on the floor (OBS-5). Content-bearing input falls + # through to start a fresh run; bare control replies get an + # honest status instead of a fake resume. + if force_resume or self._is_plain_resume_control_reply(user_reply): + snapshot_text = self._format_company_runtime_snapshot(tasks) + return ( + "This company run ended with failed work items and is closed. " + "It cannot be resumed. Send a new request (for example the " + "original goal, optionally adjusted) to start a fresh run.\n\n" + f"{snapshot_text}" + ) + return None if not force_resume: if not (no_active_runtime_work and has_closed_delivery_review): followup_result = await self._resume_company_runtime_via_final_decider( @@ -10997,9 +11182,29 @@ class OPCEngine: if checkpoint.checkpoint_type == "company_work_item_gate": return await self._resume_company_runtime_checkpoint(checkpoint, user_reply) if self._is_company_runtime_suspend_checkpoint(checkpoint.checkpoint_type): - if self._reply_metadata_requests_force_resume(reply_metadata): + if self._reply_metadata_requests_force_resume( + reply_metadata + ) or self._is_plain_resume_control_reply(user_reply): return await self._resume_company_suspend_checkpoint(checkpoint, user_reply) return await self._resume_company_suspend_checkpoint_via_final_decider(checkpoint, user_reply) + if checkpoint.checkpoint_type == "company_run_failure_review": + if not explicit_checkpoint_id: + # Closure cards must never swallow ordinary session messages. + return None + normalized_failure_reply = " ".join( + str(user_reply or "").strip().lower().split() + ) + await self.store.resolve_execution_checkpoint( + checkpoint.checkpoint_id, status="resolved" + ) + if normalized_failure_reply in { + # trailing entries are Chinese acknowledgement spellings + "", "dismiss", "ignore", "close", "ok", "okay", "知道了", "关闭", + }: + return "Company run closure acknowledged." + # Content-bearing reply: the failed run is closed, so let the + # message continue as a fresh request through normal routing. + return None if checkpoint.checkpoint_type == "company_delivery_feedback": if not explicit_checkpoint_id: return None diff --git a/opc/layer2_organization/company_mode.py b/opc/layer2_organization/company_mode.py index 83706fa..9bef55e 100644 --- a/opc/layer2_organization/company_mode.py +++ b/opc/layer2_organization/company_mode.py @@ -8,6 +8,7 @@ import hashlib import inspect import json import re +import time import uuid from contextvars import ContextVar, Token from dataclasses import asdict, dataclass, field @@ -156,6 +157,7 @@ from opc.layer2_organization.work_item_runtime_invariants import ( validate_work_item_runtime_projection, ) from opc.layer4_tools.output_budget import clip_text +from opc.llm.provider import ProviderQuotaExhaustedError from opc.llm.retry import LLMRetryError, call_llm_json_with_retry @@ -1423,6 +1425,13 @@ class CompanyWorkItemExecutor: # waits on this Event so children are claimed+spawned without # waiting for the parent turn's gather batch to drain. self._dispatcher_wake = asyncio.Event() + # Provider-quota park (OBS-6): while monotonic time is below + # _quota_park_until the dispatcher stops claiming new work instead of + # hammering an exhausted quota. The streak drives exponential backoff + # (60s → 900s cap); parks separated by more than 30 min restart it. + self._quota_park_until = 0.0 + self._quota_park_streak = 0 + self._quota_last_park_at = 0.0 # Single-dispatcher-per-run invariant: refcount of live # _execute_multi_team_org loops keyed by delegation run id. # Checkpoint answers consult this via wake_live_run_dispatcher — @@ -4760,12 +4769,23 @@ class CompanyWorkItemExecutor: self._rehydrate_parked_member_sessions(work_items) # Claim whatever is immediately claimable and spawn each # work item as an independent asyncio.Task so the loop no - # longer blocks on the slowest sibling. - claims = await self._claim_and_create_work_item_tasks( - tasks, - work_items, - active_work_item_tasks, - ) + # longer blocks on the slowest sibling. While a provider + # quota park is active, claiming is skipped so an exhausted + # quota is not hammered with doomed dispatches (OBS-6). + if self._quota_park_until and time.monotonic() < self._quota_park_until: + claims = [] + else: + if self._quota_park_until: + self._quota_park_until = 0.0 + await self._emit_progress( + "[Company] provider quota backoff elapsed — resuming dispatch", + task_id=tasks[0].id if tasks else "", + ) + claims = await self._claim_and_create_work_item_tasks( + tasks, + work_items, + active_work_item_tasks, + ) # Termination: only when nothing is in-flight AND nothing # else is runnable. If work items are still running, even an # "empty runnable" snapshot may become non-empty within @@ -4972,6 +4992,10 @@ class CompanyWorkItemExecutor: res, ) active_work_item_tasks.clear() + try: + await self._settle_run_lifecycle_on_convergence(tasks) + except Exception: + logger.opt(exception=True).debug("run failure settlement skipped") return self._summarize_multi_team_org_results(tasks) @staticmethod @@ -5136,12 +5160,80 @@ class CompanyWorkItemExecutor: ) == "running" ) + def _exception_is_provider_quota(self, exc: BaseException | None) -> bool: + seen: set[int] = set() + while exc is not None and id(exc) not in seen: + if isinstance(exc, ProviderQuotaExhaustedError): + return True + seen.add(id(exc)) + exc = exc.__cause__ or exc.__context__ + return False + + async def _park_claimed_work_item_for_quota( + self, + member_session: CompanyMemberSession, + task: Task, + exc: Exception, + ) -> None: + """Return the item to READY and back off instead of failing it. + + An exhausted provider quota is an environment outage, not a defect in + the work: failing the card (the previous behavior) terminally killed + intake and with it the whole run, with no way to continue after the + quota window reset (OBS-6). Parking keeps the run alive; dispatch + resumes automatically after the backoff, and stop/resume stays + available throughout. + """ + projection_id = self._projection_id_for_task(task) + now = time.monotonic() + if now - self._quota_last_park_at > 1800: + self._quota_park_streak = 0 + self._quota_park_streak += 1 + self._quota_last_park_at = now + backoff_sec = min(60 * (2 ** (self._quota_park_streak - 1)), 900) + self._quota_park_until = now + backoff_sec + summary = ( + f"[Company:{projection_id}] provider quota/rate limit exhausted — " + f"work item returned to the queue; dispatch pauses for {backoff_sec}s " + f"(streak {self._quota_park_streak}). {str(exc)[:300]}" + ) + logger.warning(summary) + if self._claimed_work_item_needs_cleanup(member_session, task): + try: + await transition_work_item_from_task( + self.store, task, + target_status_or_phase=Phase.READY, + reason="provider_quota_exhausted", + summary=summary or None, + release_claim=True, + attempt_outcome="interrupted", + ) + except Exception: + logger.opt(exception=True).error( + f"[Company:{projection_id}] quota park: READY transition failed" + ) + self.runtime._claimed_task_ids.discard(task.id) + work_item_id = linked_work_item_id_for_task(task) + if work_item_id: + self.runtime._claimed_work_item_ids.discard(work_item_id) + member_session.status = "idle" + member_session.resident_status = "idle" + member_session.current_task_id = "" + member_session.focused_work_item_id = "" + member_session.current_work_item = {} + member_session.current_assignment = {} + member_session.updated_at = datetime.now() + await self._emit_progress(summary, task_id=task.id) + async def _handle_claimed_work_item_exception( self, member_session: CompanyMemberSession, task: Task, exc: Exception, ) -> None: + if self._exception_is_provider_quota(exc): + await self._park_claimed_work_item_for_quota(member_session, task, exc) + return projection_id = self._projection_id_for_task(task) work_item_id = linked_work_item_id_for_task(task) summary = ( @@ -13898,6 +13990,126 @@ class CompanyWorkItemExecutor: except Exception: logger.opt(exception=True).debug("Failed to save delegation run owner review lifecycle update") + async def _settle_run_lifecycle_on_convergence(self, tasks: list[Task]) -> None: + """Close the delegation run when its tree converged in terminal failure. + + The success path closes through delivery review (awaiting_owner plus + the feedback card). A run whose intake or delivery failed previously + stayed running/active forever: no closure signal, no card, and new + session input was routed into resuming the dead run (OBS-5). + """ + if not self.store or not hasattr(self.store, "get_delegation_run"): + return + run_id = self._delegation_run_id_for_tasks(tasks) + if not run_id: + return + list_work_items = getattr(self.store, "list_delegation_work_items", None) + if not callable(list_work_items): + return + try: + work_items = await list_work_items(run_id) + except Exception: + logger.opt(exception=True).debug("run failure settlement: work item load failed") + return + if not work_items: + return + if any(getattr(item, "phase", None) not in DONE_PHASES for item in work_items): + return + failed_core = [ + item + for item in work_items + if str(getattr(item, "kind", "") or "").strip().lower() in {"intake", "delivery"} + and getattr(item, "phase", None) in {Phase.FAILED, Phase.CANCELLED} + ] + if not failed_core: + return + try: + run = await self.store.get_delegation_run(run_id) + except Exception: + logger.opt(exception=True).debug("run failure settlement: run load failed") + return + if run is None: + return + lifecycle = str(getattr(run, "lifecycle_status", "") or "").strip() + if lifecycle in {"closed_failed", "awaiting_owner"}: + return + failed_items = [ + { + "work_item_id": str(getattr(item, "work_item_id", "") or ""), + "kind": str(getattr(item, "kind", "") or ""), + "role_id": str(getattr(item, "role_id", "") or ""), + "phase": str(getattr(getattr(item, "phase", None), "value", "") or ""), + "blocked_reason": str(getattr(item, "blocked_reason", "") or "")[:300], + } + for item in work_items + if getattr(item, "phase", None) in {Phase.FAILED, Phase.CANCELLED} + ] + closed_at = datetime.now().isoformat() + run.status = ( + "failed" + if any(item["phase"] == Phase.FAILED.value for item in failed_items) + else "cancelled" + ) + run.lifecycle_status = "closed_failed" + run.metadata = { + **dict(run.metadata or {}), + "run_failure": { + "closed_at": closed_at, + "failed_items": failed_items, + }, + } + try: + await self.store.save_delegation_run(run) + except Exception: + logger.opt(exception=True).debug("run failure settlement: run save failed") + return + origin_task = tasks[0] if tasks else None + failure_lines = "\n".join( + f"- `{item['kind']}::{item['role_id']}` {item['phase']}" + + (f" — {item['blocked_reason']}" if item["blocked_reason"] else "") + for item in failed_items + ) + await self._emit_progress( + f"[Company] run closed after terminal failure — {len(failed_items)} " + f"failed/cancelled work item(s). Send a new request to start a fresh run.", + task_id=origin_task.id if origin_task else "", + ) + if self.checkpoint_callback and origin_task is not None: + original_request = str( + (origin_task.metadata or {}).get("original_request", "") + or origin_task.description + or origin_task.title + or "" + ).strip() + try: + await self.checkpoint_callback( + { + "checkpoint_type": "company_run_failure_review", + "project_id": origin_task.project_id, + "session_id": origin_task.session_id, + "task_id": origin_task.id, + "payload": { + "run_id": run_id, + "waiting_task_id": origin_task.id, + "session_id": origin_task.session_id, + "task_ids": [t.id for t in tasks], + "closed_at": closed_at, + "failed_items": failed_items, + "original_request": original_request, + "prompt": ( + "This company run ended with failed work items and has " + "been closed.\n\n" + f"{failure_lines}\n\n" + "Reply `dismiss` to acknowledge, or send a new request " + "(for example the original goal with adjustments) to " + "start a fresh run." + ), + }, + } + ) + except Exception: + logger.opt(exception=True).debug("run failure settlement: checkpoint save failed") + async def _finalize_completed_work_item(self, task: Task) -> None: if self._is_authoritative_delivery_work_item(task): plan = self._active_plan or CompanyWorkItemRuntimePlan( diff --git a/opc/layer3_agent/runtime_v2/runtime.py b/opc/layer3_agent/runtime_v2/runtime.py index 3c2c479..fe8e6e8 100644 --- a/opc/layer3_agent/runtime_v2/runtime.py +++ b/opc/layer3_agent/runtime_v2/runtime.py @@ -39,7 +39,7 @@ from opc.layer4_tools.output_budget import clip_text from opc.layer4_tools.registry import ToolDefinition from opc.layer4_tools.registry import ToolRegistry from opc.layer6_observability.cost_tracker import CostEntry -from opc.llm.provider import LLMProvider +from opc.llm.provider import LLMProvider, ProviderQuotaExhaustedError ApprovalCallback = Callable[[ToolDefinition, dict[str, Any], Optional[Task], Any], Awaitable[tuple[bool, Any]]] @@ -537,6 +537,23 @@ class NativeRuntimeV2: }, ) except Exception as exc: + rate_limit_checker = getattr(self.llm, "is_rate_limit_error", None) + if callable(rate_limit_checker) and rate_limit_checker(exc): + # Quota/rate-limit rejections never reach the model, so + # conversation-feedback retries cannot help — surface a + # typed error for the dispatcher to park on instead of + # burning retries and failing the work item (OBS-6). + await self._cancel_early_tool_runs(early_tool_runs) + await self._emit_runtime_event( + runtime_session_id, + task, + "provider_quota_exhausted", + { + "iteration": iteration + 1, + "message": str(exc)[:600], + }, + ) + raise ProviderQuotaExhaustedError(str(exc)) from exc if self.llm.is_context_overflow_error(exc) and overflow_retries < max_overflow_retries: overflow_retries += 1 messages = await self._apply_context_pipeline( diff --git a/opc/llm/provider.py b/opc/llm/provider.py index 803ef19..0bc5ee1 100644 --- a/opc/llm/provider.py +++ b/opc/llm/provider.py @@ -204,6 +204,16 @@ def _parse_tool_arguments(tool_name: str, arguments: Any) -> tuple[Any, str | No return raw, raw, error +class ProviderQuotaExhaustedError(RuntimeError): + """The provider rejected the request for quota/rate-limit reasons. + + Raised by the agent runtime instead of retrying in place: replaying the + same payload against an exhausted quota can only fail, so the company + dispatcher parks the work (returns the item to READY and backs off) + rather than failing it terminally (OBS-6). + """ + + class LLMProvider: """Unified LLM interface via LiteLLM supporting tool calls.""" @@ -401,6 +411,44 @@ class LLMProvider: raw = json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str) return hashlib.sha256(raw.encode("utf-8")).hexdigest() + def is_rate_limit_error(self, error: Exception) -> bool: + """Classify provider quota / rate-limit rejections. + + These never produce model output and never get better by replaying + the identical payload, so callers must park/back off instead of + burning conversation-feedback retries (OBS-6). Classification is by + exception type when available and by error text otherwise — the + streaming path re-raises provider errors as plain RuntimeError with + only the message preserved. + """ + if isinstance(error, litellm.exceptions.RateLimitError): + return True + if "ratelimit" in type(error).__name__.lower(): + return True + if getattr(error, "status_code", None) == 429: + return True + message = str(error).lower() + keywords = ( + "rate limit", + "rate_limit", + "ratelimit", + "too many requests", + "insufficient_quota", + "quota exceeded", + "exceeded your quota", + "quota exhausted", + "error code: 429", + "status code: 429", + "http 429", + # Chinese-provider spellings of the same rejection (Volces/DeepSeek + # and other domestic endpoints return localized error text). + "请求过于频繁", + "配额已用完", + "配额耗尽", + "触发限流", + ) + return any(keyword in message for keyword in keywords) + def is_context_overflow_error(self, error: Exception) -> bool: if isinstance(error, litellm.exceptions.ContextWindowExceededError): return True diff --git a/tests/test_attempt_ledger.py b/tests/test_attempt_ledger.py index e64a56b..5dbed2d 100644 --- a/tests/test_attempt_ledger.py +++ b/tests/test_attempt_ledger.py @@ -420,7 +420,12 @@ class ResumeAvailabilityGateTests(unittest.IsolatedAsyncioTestCase): ) return engine - async def test_resume_fails_closed_when_pinned_agent_unavailable(self) -> None: + async def test_resume_heals_pin_to_native_when_agent_unavailable(self) -> None: + """A pin to a disabled external agent with NO resumable external + session heals to native and resumes (OBS-11): dispatch would fall + back to native anyway, so failing the item punished runs — including + fully native ones whose metadata inherited a template preference — + for an availability gap that does not block execution.""" store = await self._store() task = await self._seed(store, external_agent="codex") engine = self._engine(store, available=["opencode"]) # codex disabled @@ -449,19 +454,25 @@ class ResumeAvailabilityGateTests(unittest.IsolatedAsyncioTestCase): 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_item.phase, Phase.RUNNING) + self.assertIn("tasks", executed) + resumed_task = executed["tasks"][0] + self.assertIsNone(resumed_task.assigned_external_agent) self.assertEqual( - refreshed_task.metadata.get("resume_unavailable_external_agent"), + resumed_task.metadata.get("selected_execution_agent"), "native" + ) + self.assertEqual( + resumed_task.metadata.get("resume_execution_agent_healed_from"), "codex", ) - # The runtime still executed (the rest of the org resumes normally). - self.assertIn("tasks", executed) + pin = dict( + resumed_task.metadata.get( + "_company_runtime_resume_execution_agent_pin", {} + ) + ) + self.assertEqual(pin.get("selected_execution_agent"), "native") - async def test_plain_message_after_gate_failure_converges_without_revival(self) -> None: + async def test_plain_message_after_terminal_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 @@ -480,12 +491,14 @@ class ResumeAvailabilityGateTests(unittest.IsolatedAsyncioTestCase): 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}, - ) + # The decider card failed terminally after the suspend (the resume + # gate no longer fails pins without external sessions — OBS-11 — so + # the terminal failure is seeded directly). + await store.update_delegation_work_item("work-item-1", phase=Phase.FAILED) + failed_task = await store.get_task(task.id) + assert failed_task is not None + failed_task.status = TaskStatus.FAILED + await store.save_task(failed_task) refreshed_item = await store.get_delegation_work_item("work-item-1") assert refreshed_item is not None self.assertEqual(refreshed_item.phase, Phase.FAILED) diff --git a/tests/test_company_recruiter.py b/tests/test_company_recruiter.py index d73408e..6a4bf6a 100644 --- a/tests/test_company_recruiter.py +++ b/tests/test_company_recruiter.py @@ -411,7 +411,19 @@ class CompanyRecruiterFlowTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(checkpoint.payload["recommended_action"], "auto_recruit") self.assertEqual(checkpoint.payload["staffing_defaults"]["source"], "system") self.assertTrue(all(role["default_selection"]["kind"] == "fallback" for role in checkpoint.payload["staffing_roles"])) - self.assertTrue(all(role["selected_agent"] == "codex" for role in checkpoint.payload["staffing_roles"])) + # Card defaults become per-role overrides on approve, so they must + # reflect what will actually run (OBS-11): the role template's + # external preference when it can run, native otherwise. + agents_by_role = { + role["role_id"]: role["selected_agent"] + for role in checkpoint.payload["staffing_roles"] + } + self.assertTrue(all( + role["selected_agent"] == role["default_agent"] + for role in checkpoint.payload["staffing_roles"] + )) + self.assertEqual(agents_by_role.get("ceo"), "native") + self.assertEqual(agents_by_role.get("senior_engineer"), "codex") self.assertEqual(llm.calls, []) self.assertEqual(tasks, []) await store.close() @@ -634,7 +646,21 @@ class CompanyRecruiterFlowTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(checkpoint.payload["staffing_pool"]["employees"], []) self.assertGreater(len(checkpoint.payload["staffing_roles"]), 0) self.assertTrue(all(role["default_selection"]["kind"] == "fallback" for role in checkpoint.payload["staffing_roles"])) - self.assertTrue(all(role["selected_agent"] == "codex" for role in checkpoint.payload["staffing_roles"])) + # Card defaults become per-role overrides on approve, so they must + # reflect what will actually run (OBS-11): the role template's + # external preference when it can run, native otherwise. + agents_by_role = { + role["role_id"]: role["selected_agent"] + for role in checkpoint.payload["staffing_roles"] + } + self.assertTrue(all( + role["selected_agent"] == role["default_agent"] + for role in checkpoint.payload["staffing_roles"] + )) + # decision.preferred_agent="opencode" is an explicit session-level + # choice and outranks role template preferences. + self.assertEqual(agents_by_role.get("ceo"), "opencode") + self.assertEqual(agents_by_role.get("senior_engineer"), "opencode") template_ids = {item["template_id"] for item in checkpoint.payload["staffing_pool"]["templates"]} self.assertIn("engineering-frontend-developer", template_ids) self.assertEqual(llm.calls, []) @@ -744,7 +770,7 @@ class CompanyRecruiterFlowTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(senior_role["selected_agent"], "opencode") ceo_role = next(role for role in payload["staffing_roles"] if role["role_id"] == "ceo") self.assertEqual(ceo_role["default_selection"], {"kind": "fallback", "id": ""}) - self.assertEqual(ceo_role["selected_agent"], "codex") + self.assertEqual(ceo_role["selected_agent"], "native") await store.close() async def test_confirmed_session_reuses_staffing_defaults_without_recruiter_llm(self) -> None: @@ -1476,7 +1502,19 @@ class CompanyRecruiterFlowTests(unittest.IsolatedAsyncioTestCase): self.assertIsNotNone(checkpoint) self.assertEqual(checkpoint.checkpoint_type, "company_staffing_selection") self.assertEqual(checkpoint.payload["staffing_defaults"]["source"], "system") - self.assertTrue(all(role["selected_agent"] == "codex" for role in checkpoint.payload["staffing_roles"])) + # Card defaults become per-role overrides on approve, so they must + # reflect what will actually run (OBS-11): the role template's + # external preference when it can run, native otherwise. + agents_by_role = { + role["role_id"]: role["selected_agent"] + for role in checkpoint.payload["staffing_roles"] + } + self.assertTrue(all( + role["selected_agent"] == role["default_agent"] + for role in checkpoint.payload["staffing_roles"] + )) + self.assertEqual(agents_by_role.get("ceo"), "native") + self.assertEqual(agents_by_role.get("senior_engineer"), "codex") await store.close() async def test_deny_recruitment_cancels_execution(self) -> None: diff --git a/tests/test_company_runtime_suspend_resume.py b/tests/test_company_runtime_suspend_resume.py index 8e09ca6..3b3cd4d 100644 --- a/tests/test_company_runtime_suspend_resume.py +++ b/tests/test_company_runtime_suspend_resume.py @@ -1146,8 +1146,12 @@ class CompanyRuntimeSuspendResumeTests(unittest.IsolatedAsyncioTestCase): engine.company_executor = DummyCompanyExecutor() + # A bare "continue" is a control reply and takes the plain-resume + # path (OBS-11); only content-bearing text routes to the final + # decider as a follow-up. + followup_text = "Additional requirement: add a risk-analysis section to the report" response = await engine._maybe_resume_checkpoint( - "continue", + followup_text, "sess-parent", ) checkpoints = await store.get_pending_checkpoints( @@ -1161,18 +1165,18 @@ class CompanyRuntimeSuspendResumeTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(response, "ceo handled follow-up") self.assertEqual(captured["plan"].metadata["final_decider_role_id"], "executor") self.assertEqual(routed_task.status, TaskStatus.PENDING) - self.assertEqual(routed_task.context_snapshot["user_supplied_input"], "continue") - self.assertEqual(routed_task.metadata["latest_user_directive"], "continue") - self.assertEqual(routed_task.metadata["manager_mutation_user_input"], "continue") + self.assertEqual(routed_task.context_snapshot["user_supplied_input"], followup_text) + self.assertEqual(routed_task.metadata["latest_user_directive"], followup_text) + self.assertEqual(routed_task.metadata["manager_mutation_user_input"], followup_text) self.assertTrue(routed_task.metadata["followup_routed_to_final_decider"]) self.assertEqual(checkpoints, []) assert refreshed_item is not None self.assertEqual(refreshed_item.phase, Phase.READY) self.assertEqual(refreshed_item.metadata.get("dispatch_hold"), "") self.assertEqual(refreshed_item.metadata.get("resume_source"), "primary_session_followup") - self.assertEqual(refreshed_item.metadata.get("resume_user_reply"), "continue") - self.assertEqual(refreshed_item.metadata.get("latest_user_directive"), "continue") - self.assertEqual(refreshed_item.metadata.get("manager_mutation_user_input"), "continue") + self.assertEqual(refreshed_item.metadata.get("resume_user_reply"), followup_text) + self.assertEqual(refreshed_item.metadata.get("latest_user_directive"), followup_text) + self.assertEqual(refreshed_item.metadata.get("manager_mutation_user_input"), followup_text) self.assertEqual(refreshed_item.metadata.get("current_turn_mode"), "dispatch_required") self.assertTrue(refreshed_item.metadata.get("followup_routed_to_final_decider")) diff --git a/tests/test_provider_quota_park.py b/tests/test_provider_quota_park.py new file mode 100644 index 0000000..29a6587 --- /dev/null +++ b/tests/test_provider_quota_park.py @@ -0,0 +1,187 @@ +"""Regression: provider quota exhaustion parks work instead of failing it (OBS-6). + +Quota/rate-limit rejections never reach the model, so in-place retries can +only fail; the previous behavior burned retries and terminally failed the +work item (killing intake and the whole run during a quota window). The fix: +``LLMProvider.is_rate_limit_error`` classifies these rejections, the agent +runtime raises typed ``ProviderQuotaExhaustedError``, and the company +dispatcher returns the item to READY with exponential dispatch backoff. +""" +from __future__ import annotations + +import time +import unittest +from types import SimpleNamespace +from unittest.mock import AsyncMock + +from opc.core.config import LLMConfig +from opc.core.models import CompanyMemberSession, Task +from opc.layer2_organization.company_mode import CompanyWorkItemExecutor +from opc.llm.provider import LLMProvider, ProviderQuotaExhaustedError + + +class RateLimitClassifierTests(unittest.TestCase): + def setUp(self) -> None: + self.llm = LLMProvider(LLMConfig()) + + def test_text_shapes_classify_as_rate_limit(self) -> None: + for message in ( + "Error code: 429 - rate limit reached for requests", + "RateLimitError: too many requests, retry later", + "insufficient_quota: you exceeded your quota", + # localized error text from Chinese providers must classify too + "请求过于频繁,请稍后再试", + "当前 API 配额已用完", + ): + self.assertTrue( + self.llm.is_rate_limit_error(RuntimeError(message)), message + ) + + def test_type_name_classifies(self) -> None: + class FakeRateLimitError(Exception): + pass + + self.assertTrue(self.llm.is_rate_limit_error(FakeRateLimitError("nope"))) + + def test_status_code_classifies(self) -> None: + error = RuntimeError("throttled") + error.status_code = 429 # type: ignore[attr-defined] + self.assertTrue(self.llm.is_rate_limit_error(error)) + + def test_ordinary_errors_do_not_classify(self) -> None: + for message in ( + "maximum context length exceeded", + "connection reset by peer", + "tool call arguments malformed", + "prompt tokens: 14290", + ): + self.assertFalse( + self.llm.is_rate_limit_error(RuntimeError(message)), message + ) + + +class QuotaExceptionChainTests(unittest.TestCase): + def _executor(self) -> CompanyWorkItemExecutor: + return CompanyWorkItemExecutor.__new__(CompanyWorkItemExecutor) + + def test_direct_and_chained_detection(self) -> None: + executor = self._executor() + direct = ProviderQuotaExhaustedError("quota") + wrapped = RuntimeError("turn failed") + wrapped.__cause__ = ProviderQuotaExhaustedError("quota") + unrelated = RuntimeError("boom") + self.assertTrue(executor._exception_is_provider_quota(direct)) + self.assertTrue(executor._exception_is_provider_quota(wrapped)) + self.assertFalse(executor._exception_is_provider_quota(unrelated)) + self.assertFalse(executor._exception_is_provider_quota(None)) + + +class QuotaParkTests(unittest.IsolatedAsyncioTestCase): + def _executor(self) -> CompanyWorkItemExecutor: + executor = CompanyWorkItemExecutor.__new__(CompanyWorkItemExecutor) + executor.store = None + executor.runtime = SimpleNamespace( + _claimed_task_ids={"task-1"}, + _claimed_work_item_ids=set(), + ) + executor._quota_park_until = 0.0 + executor._quota_park_streak = 0 + executor._quota_last_park_at = 0.0 + executor._emit_progress = AsyncMock() + executor._projection_id_for_task = lambda task: "cto::execute::x" + # Keep the unit test at the park-accounting level: the durable READY + # transition is exercised by the claim-release invariant suite. + executor._claimed_work_item_needs_cleanup = lambda member, task: False + return executor + + def _session(self) -> CompanyMemberSession: + session = CompanyMemberSession.__new__(CompanyMemberSession) + session.status = "running" + session.resident_status = "running" + session.current_task_id = "task-1" + session.focused_work_item_id = "wi-1" + session.current_work_item = {"id": "wi-1"} + session.current_assignment = {"id": "wi-1"} + return session + + async def test_park_backs_off_exponentially_and_idles_session(self) -> None: + executor = self._executor() + session = self._session() + task = Task(id="task-1", title="t", project_id="p", session_id="s") + + await executor._handle_claimed_work_item_exception( + session, task, ProviderQuotaExhaustedError("429 rate limit") + ) + + now = time.monotonic() + self.assertEqual(executor._quota_park_streak, 1) + self.assertAlmostEqual(executor._quota_park_until - now, 60, delta=5) + self.assertEqual(session.status, "idle") + self.assertEqual(session.current_task_id, "") + self.assertNotIn("task-1", executor.runtime._claimed_task_ids) + # No terminal failure was recorded on the task. + self.assertNotIn("claimed_work_item_exception", dict(task.metadata or {})) + + await executor._handle_claimed_work_item_exception( + session, task, ProviderQuotaExhaustedError("429 again") + ) + self.assertEqual(executor._quota_park_streak, 2) + self.assertAlmostEqual( + executor._quota_park_until - time.monotonic(), 120, delta=5 + ) + + async def test_streak_resets_after_a_quiet_period(self) -> None: + executor = self._executor() + executor._quota_park_streak = 5 + executor._quota_last_park_at = time.monotonic() - 3600 + session = self._session() + task = Task(id="task-1", title="t", project_id="p", session_id="s") + + await executor._handle_claimed_work_item_exception( + session, task, ProviderQuotaExhaustedError("429") + ) + self.assertEqual(executor._quota_park_streak, 1) + + async def test_backoff_caps_at_fifteen_minutes(self) -> None: + executor = self._executor() + executor._quota_park_streak = 9 + executor._quota_last_park_at = time.monotonic() + session = self._session() + task = Task(id="task-1", title="t", project_id="p", session_id="s") + + await executor._handle_claimed_work_item_exception( + session, task, ProviderQuotaExhaustedError("429") + ) + self.assertLessEqual(executor._quota_park_until - time.monotonic(), 900 + 5) + + async def test_non_quota_exception_still_fails_terminally(self) -> None: + executor = self._executor() + session = self._session() + task = Task(id="task-1", title="t", project_id="p", session_id="s") + failed: dict = {} + + async def _fake_fail(member_session, failed_task, exc) -> None: # noqa: ANN001 + failed["exc"] = exc + + # Route the non-quota path into a probe: the real body needs a store. + original = CompanyWorkItemExecutor._handle_claimed_work_item_exception + + async def _probe(self, member_session, failed_task, exc): # noqa: ANN001 + if self._exception_is_provider_quota(exc): + await self._park_claimed_work_item_for_quota(member_session, failed_task, exc) + return + await _fake_fail(member_session, failed_task, exc) + + try: + CompanyWorkItemExecutor._handle_claimed_work_item_exception = _probe + await executor._handle_claimed_work_item_exception( + session, task, RuntimeError("real crash") + ) + finally: + CompanyWorkItemExecutor._handle_claimed_work_item_exception = original + self.assertIsInstance(failed.get("exc"), RuntimeError) + self.assertEqual(executor._quota_park_streak, 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_run_failure_settlement.py b/tests/test_run_failure_settlement.py new file mode 100644 index 0000000..42d5334 --- /dev/null +++ b/tests/test_run_failure_settlement.py @@ -0,0 +1,185 @@ +"""Regression: failure-path run closure and post-failure input routing (OBS-5). + +A run whose intake/delivery failed used to stay running/active forever: no +closure signal, no card, and a new message was routed into re-executing the +dead run (dropping the user's content). The fix closes the run at dispatcher +convergence, emits a ``company_run_failure_review`` card, and the card's +resume handler never swallows ordinary messages — content-bearing replies +fall through so normal routing starts a fresh run. +""" +from __future__ import annotations + +import unittest +from datetime import datetime +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import AsyncMock + +from opc.core.models import ( + DelegationRun, + DelegationWorkItem, + ExecutionCheckpoint, + Task, + TaskStatus, +) +from opc.database.store import OPCStore +from opc.engine import OPCEngine +from opc.layer2_organization.company_mode import CompanyWorkItemExecutor +from opc.layer2_organization.phase import Phase + + +class RunFailureSettlementTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + self._tmp = TemporaryDirectory() + self.store = OPCStore(Path(self._tmp.name) / "tasks.db") + await self.store.initialize() + self.executor = CompanyWorkItemExecutor.__new__(CompanyWorkItemExecutor) + self.executor.store = self.store + self.captured_checkpoints: list[dict] = [] + + async def _capture(payload: dict) -> None: + self.captured_checkpoints.append(payload) + + self.executor.checkpoint_callback = _capture + self.executor._emit_progress = AsyncMock() + + async def asyncTearDown(self) -> None: + await self.store.close() + self._tmp.cleanup() + + async def _seed_run(self, *, fail_intake: bool) -> list[Task]: + run = DelegationRun(run_id="run-1", project_id="p", session_id="s") + run.status = "running" + run.lifecycle_status = "active" + await self.store.save_delegation_run(run) + intake = DelegationWorkItem( + work_item_id="wi-intake", + run_id="run-1", + role_id="ceo", + kind="intake", + title="CEO Intake", + phase=Phase.READY, + ) + execute = DelegationWorkItem( + work_item_id="wi-exec", + run_id="run-1", + role_id="cto", + kind="execute", + title="survey", + phase=Phase.READY, + ) + await self.store.save_delegation_work_item(intake) + await self.store.save_delegation_work_item(execute) + if fail_intake: + await self.store.update_delegation_work_item("wi-intake", phase=Phase.FAILED) + await self.store.update_delegation_work_item("wi-exec", phase=Phase.FAILED) + else: + await self.store.update_delegation_work_item("wi-intake", phase=Phase.RUNNING) + await self.store.update_delegation_work_item("wi-intake", phase=Phase.APPROVED) + await self.store.update_delegation_work_item("wi-exec", phase=Phase.RUNNING) + await self.store.update_delegation_work_item("wi-exec", phase=Phase.APPROVED) + task = Task( + id="task-1", + title="CEO Intake", + project_id="p", + session_id="s", + metadata={ + "delegation_run_id": "run-1", + "original_request": "Research multi-agent architectures", + }, + ) + await self.store.save_task(task) + return [task] + + async def test_terminal_failure_closes_run_and_emits_card(self) -> None: + tasks = await self._seed_run(fail_intake=True) + + await self.executor._settle_run_lifecycle_on_convergence(tasks) + + run = await self.store.get_delegation_run("run-1") + self.assertEqual(run.status, "failed") + self.assertEqual(run.lifecycle_status, "closed_failed") + self.assertTrue(run.metadata.get("run_failure", {}).get("failed_items")) + self.assertEqual(len(self.captured_checkpoints), 1) + card = self.captured_checkpoints[0] + self.assertEqual(card["checkpoint_type"], "company_run_failure_review") + self.assertEqual(card["payload"]["run_id"], "run-1") + self.assertEqual(card["payload"]["original_request"], "Research multi-agent architectures") + + async def test_successful_convergence_leaves_run_untouched(self) -> None: + tasks = await self._seed_run(fail_intake=False) + + await self.executor._settle_run_lifecycle_on_convergence(tasks) + + run = await self.store.get_delegation_run("run-1") + self.assertEqual(run.lifecycle_status, "active") + self.assertEqual(self.captured_checkpoints, []) + + async def test_settlement_is_idempotent(self) -> None: + tasks = await self._seed_run(fail_intake=True) + await self.executor._settle_run_lifecycle_on_convergence(tasks) + await self.executor._settle_run_lifecycle_on_convergence(tasks) + self.assertEqual(len(self.captured_checkpoints), 1) + + +class FailureReviewCheckpointReplyTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + self._tmp = TemporaryDirectory() + self.store = OPCStore(Path(self._tmp.name) / "tasks.db") + await self.store.initialize() + self.engine = OPCEngine(project_id="p") + self.engine.store = self.store + checkpoint = ExecutionCheckpoint( + checkpoint_id="ckpt-fail-1", + project_id="p", + session_id="s", + checkpoint_type="company_run_failure_review", + task_id="task-1", + status="pending", + payload={ + "run_id": "run-1", + "session_id": "s", + "prompt": "run closed", + }, + created_at=datetime.now(), + ) + await self.store.save_execution_checkpoint(checkpoint) + self.checkpoint = checkpoint + + async def asyncTearDown(self) -> None: + await self.store.close() + self._tmp.cleanup() + + async def _checkpoint_status(self) -> str: + loaded = await self.engine._load_execution_checkpoint_by_id("ckpt-fail-1") + return str(getattr(loaded, "status", "") or "") + + async def test_untargeted_message_is_never_swallowed(self) -> None: + reply = await self.engine._maybe_resume_checkpoint( + "Please run a fresh research task for me", + session_id="s", + ) + self.assertIsNone(reply) + self.assertEqual(await self._checkpoint_status(), "pending") + + async def test_dismiss_resolves_with_acknowledgement(self) -> None: + reply = await self.engine._maybe_resume_checkpoint( + "dismiss", + session_id="s", + reply_metadata={"response_to_checkpoint_id": "ckpt-fail-1"}, + ) + self.assertEqual(reply, "Company run closure acknowledged.") + self.assertEqual(await self._checkpoint_status(), "resolved") + + async def test_content_reply_resolves_and_falls_through(self) -> None: + reply = await self.engine._maybe_resume_checkpoint( + "Redo the research, focused on open-source frameworks this time", + session_id="s", + reply_metadata={"response_to_checkpoint_id": "ckpt-fail-1"}, + ) + self.assertIsNone(reply) + self.assertEqual(await self._checkpoint_status(), "resolved") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_stop_resume_native_pin.py b/tests/test_stop_resume_native_pin.py new file mode 100644 index 0000000..ef2888b --- /dev/null +++ b/tests/test_stop_resume_native_pin.py @@ -0,0 +1,237 @@ +"""Regression: stop/resume must not kill runs over an unavailable agent pin. + +OBS-11: a corporate role template's ``preferred_external_agent: codex`` was +stamped into execution identity even when the run was requested and executed +as native. On resume, the availability gate trusted that pin and failed every +non-terminal work item closed. These tests pin the four legs of the fix: + +1. identity truth — seat enrichment and the dispatch selector record the + backend that actually runs (native fallback when the external agent is + provably unavailable), +2. resume gate symmetry — a pin without a resumable external session heals + to native instead of failing the item, +3. control/content separation — a bare "continue" (or force_resume metadata, + in either spelling) resumes the runtime instead of being routed to the + final decider as a follow-up. +""" +from __future__ import annotations + +import unittest +from pathlib import Path +from tempfile import TemporaryDirectory +from types import SimpleNamespace + +from opc.core.models import DelegationWorkItem, Task, TaskStatus +from opc.database.store import OPCStore +from opc.engine import OPCEngine +from opc.layer2_organization.phase import Phase +from opc.layer2_organization.work_item_links import set_linked_work_item_id + + +class _AdapterRegistryStub: + def __init__(self, available: list[str]): + self._available = list(available) + + def list_available(self) -> list[str]: + return list(self._available) + + def get(self, name: str): + return object() if name in self._available else None + + def get_ordered_available(self): + return [(name, object()) for name in self._available] + + +class PlainResumeControlReplyTests(unittest.TestCase): + def test_control_tokens_are_plain_resume(self) -> None: + # includes the Chinese continuation spellings the product accepts + for reply in ("continue", " Resume ", "proceed", "ok", "y", "继续", "恢复", ""): + self.assertTrue(OPCEngine._is_plain_resume_control_reply(reply), reply) + + def test_content_is_not_plain_resume(self) -> None: + # the Chinese sample starts with a control word but carries content, + # so it must NOT be treated as a bare control reply + for reply in ("make a ppt outline", "继续,但先修复报告第3节", "deny"): + self.assertFalse(OPCEngine._is_plain_resume_control_reply(reply), reply) + + def test_force_resume_metadata_both_spellings(self) -> None: + self.assertTrue(OPCEngine._reply_metadata_requests_force_resume({"ui_force_resume": True})) + self.assertTrue(OPCEngine._reply_metadata_requests_force_resume({"force_resume": True})) + self.assertFalse(OPCEngine._reply_metadata_requests_force_resume({"other": True})) + self.assertFalse(OPCEngine._reply_metadata_requests_force_resume(None)) + + +class LockedAgentAvailabilityFallbackTests(unittest.IsolatedAsyncioTestCase): + def _engine(self, available: list[str]) -> OPCEngine: + engine = OPCEngine(project_id="p") + engine.org_engine = SimpleNamespace() + engine.adapter_registry = _AdapterRegistryStub(available) + return engine + + async def test_locked_unavailable_external_falls_back_to_native(self) -> None: + engine = self._engine(available=[]) + task = Task(id="t1", title="t", project_id="p", session_id="s") + task.metadata["execution_agent_locked"] = True + task.metadata["selected_execution_agent"] = "codex" + + selected = await engine._assign_task_execution_agent(task) + + self.assertIsNone(selected) + self.assertIsNone(task.assigned_external_agent) + self.assertEqual(task.metadata["selected_execution_agent"], "native") + self.assertEqual(task.metadata["execution_agent_unavailable"], "codex") + self.assertEqual( + task.metadata["agent_selection"]["decision_reason"], + "locked_external_agent_unavailable_native_fallback", + ) + + async def test_locked_available_external_is_kept(self) -> None: + engine = self._engine(available=["codex"]) + task = Task(id="t2", title="t", project_id="p", session_id="s") + task.metadata["execution_agent_locked"] = True + task.metadata["selected_execution_agent"] = "codex" + + selected = await engine._assign_task_execution_agent(task) + + self.assertEqual(selected, "codex") + self.assertEqual(task.assigned_external_agent, "codex") + + +class SeatEnrichmentIdentityTruthTests(unittest.TestCase): + def _engine(self, available: list[str], role_preferred: str | None) -> OPCEngine: + engine = OPCEngine(project_id="p") + engine.adapter_registry = _AdapterRegistryStub(available) + role = SimpleNamespace(preferred_external_agent=role_preferred) + engine.org_engine = SimpleNamespace( + get_agent=lambda role_id: role, + get_employee=lambda employee_id: None, + get_default_employee_for_role=lambda role_id: None, + list_employees=lambda role_id=None: [], + ensure_fallback_employee_for_role=lambda role_id, persist=False: None, + ) + return engine + + def _enrich(self, engine: OPCEngine, preferred_agent: str | None) -> dict: + decision = SimpleNamespace(preferred_agent=preferred_agent) + topology = {"seats": [{"role_id": "cto", "seat_id": "seat-cto"}]} + enriched = engine._enrich_runtime_delegation_topology( + runtime_topology=topology, + decision=decision, + project_id="p", + ) + return enriched["seats"][0] + + def test_explicit_native_wins_over_role_preference(self) -> None: + engine = self._engine(available=["codex"], role_preferred="codex") + seat = self._enrich(engine, preferred_agent="native") + self.assertEqual(seat["selected_execution_agent"], "native") + self.assertTrue(seat["force_native_execution"]) + + def test_unavailable_role_preference_resolves_to_native(self) -> None: + engine = self._engine(available=[], role_preferred="codex") + seat = self._enrich(engine, preferred_agent=None) + self.assertEqual(seat["selected_execution_agent"], "native") + self.assertEqual(seat["execution_agent_unavailable"], "codex") + + def test_available_role_preference_is_kept(self) -> None: + engine = self._engine(available=["codex"], role_preferred="codex") + seat = self._enrich(engine, preferred_agent=None) + self.assertEqual(seat["selected_execution_agent"], "codex") + self.assertEqual(seat["execution_agent_unavailable"], "") + + +class ResumeGateHealTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + self._tmp = TemporaryDirectory() + self.store = OPCStore(Path(self._tmp.name) / "tasks.db") + await self.store.initialize() + self.engine = OPCEngine(project_id="p") + self.engine.store = self.store + self.engine.adapter_registry = _AdapterRegistryStub([]) + + async def asyncTearDown(self) -> None: + await self.store.close() + self._tmp.cleanup() + + async def _seed(self) -> Task: + item = DelegationWorkItem( + work_item_id="wi-1", + run_id="run-1", + role_id="cto", + kind="execute", + title="survey", + phase=Phase.READY, + ) + await self.store.save_delegation_work_item(item) + task = Task( + id="task-1", + title="survey", + project_id="p", + session_id="s", + status=TaskStatus.RUNNING, + metadata={"delegation_run_id": "run-1"}, + ) + set_linked_work_item_id(task, "wi-1") + await self.store.save_task(task) + return task + + def _payload(self, task: Task, external_sessions: dict) -> dict: + return { + "checkpoint_id": "ckpt-1", + "task_snapshots": [ + { + "task_id": task.id, + "execution_identity": { + "selected_execution_agent": "codex", + "assigned_external_agent": "codex", + }, + "work_item": {"work_item_id": "wi-1"}, + } + ], + "active_work_items": [{"work_item_id": "wi-1", "phase": "ready"}], + "external_sessions": external_sessions, + "native_runtime_resume": {}, + } + + async def test_pin_without_external_session_heals_to_native(self) -> None: + task = await self._seed() + payload = self._payload(task, external_sessions={}) + + refreshed = await self.engine._prepare_company_runtime_tasks_for_resume( + [task], payload + ) + + prepared = refreshed[0] + self.assertIsNone(prepared.assigned_external_agent) + self.assertEqual(prepared.metadata.get("selected_execution_agent"), "native") + self.assertEqual( + prepared.metadata.get("resume_execution_agent_healed_from"), "codex" + ) + pin = dict(prepared.metadata.get("_company_runtime_resume_execution_agent_pin", {})) + self.assertEqual(pin.get("selected_execution_agent"), "native") + self.assertEqual(pin.get("assigned_external_agent", ""), "") + item = await self.store.get_delegation_work_item("wi-1") + self.assertEqual(item.phase, Phase.READY) + + async def test_pin_with_live_external_session_still_fails_closed(self) -> None: + task = await self._seed() + payload = self._payload( + task, + external_sessions={ + task.id: { + "status": "active", + "agent_type": "codex", + "resume_session_id": "sess-1", + } + }, + ) + + await self.engine._prepare_company_runtime_tasks_for_resume([task], payload) + + item = await self.store.get_delegation_work_item("wi-1") + self.assertEqual(item.phase, Phase.FAILED) + self.assertIn("pinned to external agent", str(item.blocked_reason or "")) + + +if __name__ == "__main__": + unittest.main()