From 1268b672230c423e78145fbf7a82bc92b2f2cee8 Mon Sep 17 00:00:00 2001 From: Mao Weiming Date: Wed, 29 Jul 2026 15:42:07 +0800 Subject: [PATCH] fix(company): stop Report #N storm and spill large Cursor prompts to files Cursor-agent puts prompts on argv, so oversized company/report handoffs hit OS ARG_MAX, crash the card as FAILED, and reconcile kept minting new Report attempts forever. Spill large Cursor prompts to a workspace file and hold the report chain after consecutive failures. --- opc/layer2_organization/company_mode.py | 156 ++++++++++++++++++++ opc/layer3_agent/adapters/cursor_adapter.py | 100 ++++++++++++- tests/test_external_agent_monitoring.py | 32 ++++ tests/test_worker_report_handoff.py | 102 +++++++++++++ 4 files changed, 382 insertions(+), 8 deletions(-) diff --git a/opc/layer2_organization/company_mode.py b/opc/layer2_organization/company_mode.py index f07c390..5b723b2 100644 --- a/opc/layer2_organization/company_mode.py +++ b/opc/layer2_organization/company_mode.py @@ -235,6 +235,12 @@ DEFAULT_MAX_PRE_DELIVERY_REWORKS = 3 # honest-but-rejected work. MAX_VERDICT_PARSE_RETRIES = 2 +# Cap consecutive FAILED report cards for one parent while it stays in +# AWAITING_MANAGER_REVIEW. Reconcile treats FAILED as "no active report" and +# would otherwise mint Report #N forever (seen when cursor-agent spawn dies +# with ARG_MAX before the process starts). +DEFAULT_MAX_CONSECUTIVE_REPORT_FAILURES = 3 + _REVIEW_VERDICT_PARSE_RETRY_HINT = ( "\n\n[REVIEW RETRY — Your previous verdict could not be parsed. The " "runtime needs an explicit approve/reject decision to drive the next " @@ -3881,6 +3887,13 @@ class CompanyWorkItemExecutor: else: # No unconsumed durable report exists. This is either the # first handoff for the phase or a later rework cycle. + # Brake: consecutive FAILED report cards must not mint a + # new Report #N on every reconcile tick. + if await self._should_hold_report_chain( + parent, + work_items=work_items, + ): + continue spawned = await self._ensure_report_work_item_for_work_item( target_id, run_items=work_items, @@ -6278,6 +6291,37 @@ class CompanyWorkItemExecutor: review_evidence["manager_dispatch"] = dict(manager_turn_context) if review_evidence: metadata_updates["review_evidence"] = review_evidence + if target_phase == Phase.AWAITING_MANAGER_REVIEW: + # A fresh execute→review handoff must clear any prior report + # storm hold so rework cycles can spawn Report #1 again. + # Baseline the failure streak on already-minted report attempts + # so historical FAILED cards from a previous cycle do not + # immediately re-trigger the hold. + prior_attempts = 0 + if linked_work_item is not None: + try: + prior_items = await self._run_items_for_parent(linked_work_item) + prior_attempts = max( + ( + self._auxiliary_attempt_number(item, kind="report") + for item in self._targeting_auxiliary_items( + prior_items, + work_item_id, + kind="report", + ) + ), + default=0, + ) + except Exception: + prior_attempts = int( + (linked_work_item_metadata or {}).get("report_attempt_count", 0) + or 0 + ) + metadata_updates["report_chain_hold"] = "" + metadata_updates["report_chain_hold_reason"] = "" + metadata_updates["report_chain_hold_at"] = "" + metadata_updates["report_chain_failed_attempts"] = 0 + metadata_updates["report_failure_baseline_attempt"] = prior_attempts # Phase write + local status sync via the canonical helper. Returns # False only if wid disappeared between our lookup and the call — @@ -6792,6 +6836,111 @@ class CompanyWorkItemExecutor: ] return active[-1] if active else None + @classmethod + def _consecutive_failed_auxiliary_attempts( + cls, + run_items: list[DelegationWorkItem], + target_work_item_id: str, + *, + kind: str, + baseline_attempt: int = 0, + ) -> int: + """Count trailing FAILED auxiliary cards since the last non-failed one. + + Why this exists: each crashed report card settles as Phase.FAILED + (a DONE_PHASE), so ``_active_auxiliary_item`` returns None and + reconcile mints a fresh attempt. Counting the trailing failure + streak lets the runtime stop before Report #295. + + ``baseline_attempt`` ignores older cards from a previous execute→ + review cycle so a rework handoff can spawn again after the hold + was cleared. + """ + floor = max(0, int(baseline_attempt or 0)) + streak = 0 + for item in reversed( + cls._targeting_auxiliary_items( + run_items, + target_work_item_id, + kind=kind, + ) + ): + attempt_no = cls._auxiliary_attempt_number(item, kind=kind) + if attempt_no <= floor: + break + if getattr(item, "phase", None) == Phase.FAILED: + streak += 1 + continue + break + return streak + + @classmethod + def _report_failure_limit(cls, parent_item: DelegationWorkItem) -> int: + metadata = dict(getattr(parent_item, "metadata", {}) or {}) + raw = metadata.get("max_consecutive_report_failures") + try: + value = int(raw) if raw is not None else DEFAULT_MAX_CONSECUTIVE_REPORT_FAILURES + except (TypeError, ValueError): + value = DEFAULT_MAX_CONSECUTIVE_REPORT_FAILURES + return max(1, value) + + async def _should_hold_report_chain( + self, + parent_item: DelegationWorkItem, + *, + work_items: list[DelegationWorkItem] | None = None, + ) -> bool: + """True when reconcile must stop minting new report cards for parent.""" + parent_metadata = dict(getattr(parent_item, "metadata", {}) or {}) + if str(parent_metadata.get("report_chain_hold", "") or "").strip(): + return True + all_run_items = await self._run_items_for_parent(parent_item, work_items) + try: + baseline = int(parent_metadata.get("report_failure_baseline_attempt", 0) or 0) + except (TypeError, ValueError): + baseline = 0 + consecutive_failures = self._consecutive_failed_auxiliary_attempts( + all_run_items, + parent_item.work_item_id, + kind="report", + baseline_attempt=baseline, + ) + limit = self._report_failure_limit(parent_item) + if consecutive_failures < limit: + return False + hold_reason = ( + f"{consecutive_failures} consecutive report cards failed " + f"(limit {limit}); refusing to spawn another Report #N until " + "the parent leaves AWAITING_MANAGER_REVIEW or the hold is cleared" + ) + try: + await self.store.update_delegation_work_item( + parent_item.work_item_id, + metadata_updates={ + "report_chain_hold": "consecutive_report_failures", + "report_chain_hold_reason": hold_reason, + "report_chain_hold_at": datetime.now().isoformat(), + "report_chain_failed_attempts": consecutive_failures, + }, + ) + except Exception: + logger.opt(exception=True).warning( + "Failed to stamp report_chain_hold on " + f"work_item_id={parent_item.work_item_id}" + ) + await self._record_work_item_runtime_diagnostic( + code="report_chain_held_after_failures", + severity="error", + work_item=parent_item, + message=hold_reason, + details={ + "consecutive_failures": consecutive_failures, + "limit": limit, + "baseline_attempt": baseline, + }, + ) + return True + @classmethod def _next_auxiliary_attempt( cls, @@ -7511,6 +7660,13 @@ class CompanyWorkItemExecutor: kind="review", ) is not None: return None + # Defense in depth: even callers outside reconcile (e.g. DONE + # transition) must not mint Report #N after a failure storm. + if await self._should_hold_report_chain( + worker_item, + work_items=all_run_items, + ): + return None existing_card = self._active_auxiliary_item( all_run_items, target_work_item_id, diff --git a/opc/layer3_agent/adapters/cursor_adapter.py b/opc/layer3_agent/adapters/cursor_adapter.py index 01fa56a..1c133a2 100644 --- a/opc/layer3_agent/adapters/cursor_adapter.py +++ b/opc/layer3_agent/adapters/cursor_adapter.py @@ -3,8 +3,10 @@ from __future__ import annotations import asyncio +import hashlib import re import shutil +from pathlib import Path from typing import Any from loguru import logger @@ -18,6 +20,12 @@ class CursorAdapter(ExternalAgentAdapter): agent_type = "cursor" default_command = "cursor-agent" + # cursor-agent takes the prompt as a positional argv token (not stdin in + # --print mode). Large company/report prompts routinely exceed OS ARG_MAX + # and crash spawn with ``OSError: [Errno 7] Argument list too long``. Keep + # small prompts on argv; spill larger ones to a workspace file and pass a + # short pointer prompt instead. + _ARGV_PROMPT_MAX_BYTES = 16 * 1024 def __init__(self, config=None) -> None: super().__init__(config=config) @@ -90,13 +98,71 @@ class CursorAdapter(ExternalAgentAdapter): def agent_isolation_home_slug(self) -> str: return "cursor" + def _prompt_arg_for_invocation( + self, + prompt: str, + *, + workspace_path: str | None = None, + task: Task | None = None, + ) -> tuple[str, dict[str, object]]: + """Return argv-safe prompt text plus transport metadata. + + Why file spill exists: ``cursor-agent -p`` consumes the prompt as a + positional CLI argument. Putting a 100k+ company handoff prompt on + argv trips Linux/macOS ``Argument list too long`` before the process + starts, which previously crashed report cards and triggered an + infinite Report #N reconcile storm. + """ + prompt_text = str(prompt or "") + prompt_bytes = len(prompt_text.encode("utf-8")) + if prompt_bytes <= self._ARGV_PROMPT_MAX_BYTES: + return prompt_text, { + "prompt_transport": "argv", + "prompt_bytes": prompt_bytes, + } + + root = Path(str(workspace_path or "").strip() or ".").expanduser() + try: + root = root.resolve() + except OSError: + root = Path(".").resolve() + prompt_dir = root / ".opc" / "external_prompts" + prompt_dir.mkdir(parents=True, exist_ok=True) + task_id = str(getattr(task, "id", "") or "").strip() or "task" + digest = hashlib.sha256(prompt_text.encode("utf-8")).hexdigest()[:16] + prompt_path = prompt_dir / f"cursor_{task_id}_{digest}.md" + prompt_path.write_text(prompt_text, encoding="utf-8") + pointer = ( + "Open and follow the complete task instructions in this file exactly:\n" + f"{prompt_path}\n\n" + "Treat the file contents as your full prompt. Do not ask for confirmation " + "before starting; do not recreate the file." + ) + return pointer, { + "prompt_transport": "file", + "prompt_bytes": prompt_bytes, + "prompt_file": str(prompt_path), + "prompt_transport_reason": "prompt_too_large_for_argv", + } + + @staticmethod + def _redact_prompt_arg(cmd: list[str], prompt: str) -> list[str]: + redacted = list(cmd) + if redacted: + redacted[-1] = f"" + return redacted + def build_invocation( self, task: Task, workspace_path: str | None = None, ) -> tuple[list[str], dict[str, object]]: - _ = workspace_path - prompt = self.build_task_prompt(task) + full_prompt = self.build_task_prompt(task) + prompt_arg, transport_meta = self._prompt_arg_for_invocation( + full_prompt, + workspace_path=workspace_path, + task=task, + ) command = self._runtime_command() or self.configured_command() cmd = [ command, @@ -108,10 +174,18 @@ class CursorAdapter(ExternalAgentAdapter): *self._build_model_args(), *self._build_session_args(), *list(self.config.extra_args), - prompt, + prompt_arg, ] - metadata = self.build_invocation_metadata(cmd) + # Redact large/file-backed prompts from audit command strings so logs + # stay small and never re-inflate ARG_MAX-sized text into metadata. + display_cmd = ( + self._redact_prompt_arg(cmd, full_prompt) + if transport_meta.get("prompt_transport") == "file" or len(full_prompt) > 160 + else cmd + ) + metadata = self.build_invocation_metadata(display_cmd) metadata["binary"] = command + metadata.update(transport_meta) return cmd, metadata def build_interactive_invocation( @@ -119,8 +193,12 @@ class CursorAdapter(ExternalAgentAdapter): task: Task, workspace_path: str | None = None, ) -> tuple[list[str], dict[str, object]]: - _ = workspace_path - prompt = self.build_task_prompt(task) + full_prompt = self.build_task_prompt(task) + prompt_arg, transport_meta = self._prompt_arg_for_invocation( + full_prompt, + workspace_path=workspace_path, + task=task, + ) command = self._runtime_command() or self.configured_command() cmd = [ command, @@ -132,10 +210,16 @@ class CursorAdapter(ExternalAgentAdapter): *self._build_model_args(), *self._build_session_args(), *list(self.config.extra_args), - prompt, + prompt_arg, ] - metadata = self.build_invocation_metadata(cmd) + display_cmd = ( + self._redact_prompt_arg(cmd, full_prompt) + if transport_meta.get("prompt_transport") == "file" or len(full_prompt) > 160 + else cmd + ) + metadata = self.build_invocation_metadata(display_cmd) metadata["binary"] = command + metadata.update(transport_meta) return cmd, metadata def extract_resume_session_id(self, output: str) -> str: diff --git a/tests/test_external_agent_monitoring.py b/tests/test_external_agent_monitoring.py index fcfadea..89dcb78 100644 --- a/tests/test_external_agent_monitoring.py +++ b/tests/test_external_agent_monitoring.py @@ -3503,6 +3503,38 @@ class ExternalAgentMonitoringTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(result.status, TaskStatus.DONE) self.assertEqual(spawn_mock.await_args.kwargs["stdin"], asyncio.subprocess.DEVNULL) + def test_cursor_adapter_large_prompt_spills_to_workspace_file(self) -> None: + adapter = CursorAdapter(config=ExternalAgentConfig(command="cursor-agent")) + prompt = "x" * (CursorAdapter._ARGV_PROMPT_MAX_BYTES + 1) + task = Task(id="task-large", title="large", description=prompt) + full_prompt = adapter.build_task_prompt(task) + tmpdir = _make_test_dir("cursor-large-prompt-file") + try: + cmd, metadata = adapter.build_interactive_invocation( + task, workspace_path=tmpdir + ) + self.assertEqual(metadata["prompt_transport"], "file") + self.assertEqual( + metadata["prompt_transport_reason"], + "prompt_too_large_for_argv", + ) + prompt_file = Path(str(metadata["prompt_file"])) + self.assertTrue(prompt_file.is_file()) + self.assertEqual(prompt_file.read_text(encoding="utf-8"), full_prompt) + self.assertNotIn(full_prompt, cmd) + self.assertIn(str(prompt_file), str(cmd[-1])) + self.assertNotIn(prompt[:64], metadata["command"]) + self.assertLess(len(cmd[-1].encode("utf-8")), CursorAdapter._ARGV_PROMPT_MAX_BYTES) + finally: + _cleanup_test_dir(tmpdir) + + def test_cursor_adapter_small_prompt_stays_on_argv(self) -> None: + adapter = CursorAdapter(config=ExternalAgentConfig(command="cursor-agent")) + task = Task(title="demo", description="short body") + cmd, metadata = adapter.build_invocation(task, workspace_path="/tmp/opc-ws") + self.assertEqual(metadata["prompt_transport"], "argv") + self.assertEqual(cmd[-1], adapter.build_task_prompt(task)) + def test_codex_adapter_mirrors_user_auth_and_config_with_copy_fallback(self) -> None: adapter = CodexAdapter() tmpdir = _make_test_dir("codex-mirror-user-config") diff --git a/tests/test_worker_report_handoff.py b/tests/test_worker_report_handoff.py index 73725e2..dc0456e 100644 --- a/tests/test_worker_report_handoff.py +++ b/tests/test_worker_report_handoff.py @@ -1433,5 +1433,107 @@ class ReportCardRunnableFilterTests(unittest.TestCase): ) +class ReportFailureStormBrakeTests(unittest.IsolatedAsyncioTestCase): + """FAILED report cards must not mint Report #N forever on reconcile.""" + + async def asyncSetUp(self) -> None: + self._tmpdir = tempfile.TemporaryDirectory() + self.root = Path(self._tmpdir.name) + self.store = OPCStore(self.root / "tasks.db") + await self.store.initialize() + self.executor = _build_executor(self.store, _make_org_engine(self.root)) + + async def asyncTearDown(self) -> None: + await self.store.close() + self._tmpdir.cleanup() + + def _failed_report(self, attempt: int) -> DelegationWorkItem: + report_id = report_work_item_id_for_attempt("wi-child", attempt) + return DelegationWorkItem( + work_item_id=report_id, + run_id="run-1", + cell_id="team::cto", + team_id="team::cto", + role_id="engineer", + seat_id="seat::team::cto::engineer", + parent_work_item_id="wi-child", + kind="report", + projection_id=report_id, + phase=Phase.FAILED, + batch_index=attempt, + metadata={ + "runtime_model": "multi_team_org", + "report_execution_work_item": True, + "report_attempt": attempt, + "report_target_work_item_id": "wi-child", + "hidden_from_company_kanban": True, + }, + ) + + async def test_reconcile_stops_after_consecutive_report_failures(self) -> None: + parent = _build_child_work_item() + parent.phase = Phase.AWAITING_MANAGER_REVIEW + parent.metadata = { + **dict(parent.metadata or {}), + "review_owner_role_id": "cto", + "review_owner_seat_id": "seat::team::cto::cto", + "max_consecutive_report_failures": 3, + } + await self.store.save_delegation_work_item(parent) + for attempt in (1, 2, 3): + await self.store.save_delegation_work_item(self._failed_report(attempt)) + + before = [ + item.work_item_id + for item in await self.store.list_delegation_work_items("run-1") + if item.kind == "report" + ] + await self.executor._reconcile_missing_review_chain( + await self.store.list_delegation_work_items("run-1") + ) + after = [ + item.work_item_id + for item in await self.store.list_delegation_work_items("run-1") + if item.kind == "report" + ] + self.assertEqual(before, after) + self.assertIsNone( + await self.store.get_delegation_work_item( + report_work_item_id_for_attempt("wi-child", 4) + ) + ) + refreshed = await self.store.get_delegation_work_item("wi-child") + self.assertEqual( + refreshed.metadata.get("report_chain_hold"), + "consecutive_report_failures", + ) + + async def test_reconcile_allows_retry_below_failure_limit(self) -> None: + parent = _build_child_work_item() + parent.phase = Phase.AWAITING_MANAGER_REVIEW + parent.metadata = { + **dict(parent.metadata or {}), + "review_owner_role_id": "cto", + "review_owner_seat_id": "seat::team::cto::cto", + "max_consecutive_report_failures": 3, + } + await self.store.save_delegation_work_item(parent) + await self.store.save_delegation_work_item(self._failed_report(1)) + await self.store.save_delegation_work_item(self._failed_report(2)) + + await self.executor._reconcile_missing_review_chain( + await self.store.list_delegation_work_items("run-1") + ) + next_report = await self.store.get_delegation_work_item( + report_work_item_id_for_attempt("wi-child", 3) + ) + self.assertIsNotNone(next_report) + self.assertEqual(next_report.phase, Phase.READY) + refreshed = await self.store.get_delegation_work_item("wi-child") + self.assertFalse( + str((refreshed.metadata or {}).get("report_chain_hold", "") or "").strip() + ) + + if __name__ == "__main__": unittest.main()