diff --git a/opc/layer2_organization/company_mode.py b/opc/layer2_organization/company_mode.py index 5b723b2..eafea48 100644 --- a/opc/layer2_organization/company_mode.py +++ b/opc/layer2_organization/company_mode.py @@ -6890,7 +6890,18 @@ class CompanyWorkItemExecutor: *, work_items: list[DelegationWorkItem] | None = None, ) -> bool: - """True when reconcile must stop minting new report cards for parent.""" + """True when reconcile must stop minting new report cards for parent. + + Over the failure limit the parent is terminalized to FAILED so the + normal settlement machinery (dependents refresh, manager visibility, + rework/resume) takes over — a parent parked in + AWAITING_MANAGER_REVIEW with no spawnable report card can never + advance on its own. The ``report_chain_hold`` stamp survives only as + the quarantine fallback when even the FAILED write does not land, + mirroring the attempt ledger's terminalize pattern. + """ + if getattr(parent_item, "phase", None) in DONE_PHASES: + return True parent_metadata = dict(getattr(parent_item, "metadata", {}) or {}) if str(parent_metadata.get("report_chain_hold", "") or "").strip(): return True @@ -6908,31 +6919,75 @@ class CompanyWorkItemExecutor: 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" + block_reason = ( + f"report_chain_failure: {consecutive_failures} consecutive report " + f"cards failed (limit {limit})" + ) + summary_text = ( + "Work item failed because its report handoff pipeline is broken: " + f"{consecutive_failures} consecutive report cards died before a " + f"durable report landed (limit {limit}). The execution result is " + "preserved on the card; rework retries the handoff." ) try: - await self.store.update_delegation_work_item( + await transition_work_item( + self.store, parent_item.work_item_id, + target_phase=Phase.FAILED, + reason="report_chain_failure", + summary=summary_text, 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, }, + release_claim=True, + ) + try: + await self.store.update_delegation_work_item( + parent_item.work_item_id, + blocked_reason=block_reason, + ) + except Exception: + logger.opt(exception=True).debug( + "report_chain_failure blocked_reason write failed for {}", + parent_item.work_item_id, + ) + await self._emit_progress( + f"[Company:{projection_id_for_work_item(parent_item)}] {summary_text}" + ) + except InvalidPhaseTransition: + # A concurrent writer moved the parent (e.g. a racing manager + # approval). Skip minting this tick; the next reconcile pass + # re-evaluates from the fresh phase. + logger.opt(exception=True).debug( + "report_chain_failure terminalize lost a phase race for {}", + parent_item.work_item_id, ) except Exception: logger.opt(exception=True).warning( - "Failed to stamp report_chain_hold on " - f"work_item_id={parent_item.work_item_id}" + "report_chain_failure FAILED terminalize did not land for {}; " + "quarantining via report_chain_hold", + parent_item.work_item_id, ) + 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": block_reason, + "report_chain_hold_at": datetime.now().isoformat(), + "report_chain_failed_attempts": consecutive_failures, + }, + ) + except Exception: + logger.opt(exception=True).error( + "report_chain_hold quarantine write also failed for {}", + parent_item.work_item_id, + ) await self._record_work_item_runtime_diagnostic( - code="report_chain_held_after_failures", + code="report_chain_failure_terminalized", severity="error", work_item=parent_item, - message=hold_reason, + message=summary_text, details={ "consecutive_failures": consecutive_failures, "limit": limit, diff --git a/opc/layer3_agent/adapters/cursor_adapter.py b/opc/layer3_agent/adapters/cursor_adapter.py index 1c133a2..bf83a56 100644 --- a/opc/layer3_agent/adapters/cursor_adapter.py +++ b/opc/layer3_agent/adapters/cursor_adapter.py @@ -3,7 +3,6 @@ from __future__ import annotations import asyncio -import hashlib import re import shutil from pathlib import Path @@ -128,9 +127,15 @@ class CursorAdapter(ExternalAgentAdapter): root = Path(".").resolve() prompt_dir = root / ".opc" / "external_prompts" prompt_dir.mkdir(parents=True, exist_ok=True) + # Self-ignoring directory: agents run `git add` in the workspace and + # must never commit spilled prompts. + gitignore_path = prompt_dir / ".gitignore" + if not gitignore_path.exists(): + gitignore_path.write_text("*\n", encoding="utf-8") 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" + # One stable file per task: retries and resumes overwrite instead of + # accumulating a new file per prompt revision. + prompt_path = prompt_dir / f"cursor_{task_id}.md" prompt_path.write_text(prompt_text, encoding="utf-8") pointer = ( "Open and follow the complete task instructions in this file exactly:\n" diff --git a/tests/test_external_agent_monitoring.py b/tests/test_external_agent_monitoring.py index c20e2d1..0692433 100644 --- a/tests/test_external_agent_monitoring.py +++ b/tests/test_external_agent_monitoring.py @@ -3612,6 +3612,17 @@ class ExternalAgentMonitoringTests(unittest.IsolatedAsyncioTestCase): 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) + # Hygiene: one stable file per task (retries overwrite) and a + # self-ignoring directory so workspace git never picks it up. + self.assertEqual(prompt_file.name, "cursor_task-large.md") + _cmd2, metadata2 = adapter.build_interactive_invocation( + task, workspace_path=tmpdir + ) + self.assertEqual(metadata2["prompt_file"], str(prompt_file)) + self.assertEqual( + sorted(p.name for p in prompt_file.parent.iterdir()), + [".gitignore", "cursor_task-large.md"], + ) finally: _cleanup_test_dir(tmpdir) diff --git a/tests/test_worker_report_handoff.py b/tests/test_worker_report_handoff.py index dc0456e..288292d 100644 --- a/tests/test_worker_report_handoff.py +++ b/tests/test_worker_report_handoff.py @@ -13,7 +13,7 @@ from __future__ import annotations import tempfile import unittest from pathlib import Path -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import AsyncMock, MagicMock, patch from opc.core.config import OPCConfig, RoleConfig from opc.core.events import EventBus @@ -1503,6 +1503,39 @@ class ReportFailureStormBrakeTests(unittest.IsolatedAsyncioTestCase): ) ) refreshed = await self.store.get_delegation_work_item("wi-child") + self.assertEqual(refreshed.phase, Phase.FAILED) + self.assertIn("report_chain_failure", str(refreshed.blocked_reason or "")) + self.assertFalse( + str((refreshed.metadata or {}).get("report_chain_hold", "") or "").strip() + ) + + async def test_terminalize_write_failure_quarantines_via_hold(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)) + + with patch( + "opc.layer2_organization.company_mode.transition_work_item", + side_effect=RuntimeError("db write lost"), + ): + await self.executor._reconcile_missing_review_chain( + await self.store.list_delegation_work_items("run-1") + ) + 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.phase, Phase.AWAITING_MANAGER_REVIEW) self.assertEqual( refreshed.metadata.get("report_chain_hold"), "consecutive_report_failures",