diff --git a/opc/core/company_tools.py b/opc/core/company_tools.py index 28a8107..6602cd2 100644 --- a/opc/core/company_tools.py +++ b/opc/core/company_tools.py @@ -58,12 +58,14 @@ DEBUG_ADMIN_TOOL_NAMES: tuple[str, ...] = ( MEETING_RESPONSE_TOOL_NAMES: tuple[str, ...] = ("respond_meeting",) HUMAN_REVIEW_TOOL_NAMES: tuple[str, ...] = ("close_human_review",) +SELF_EVOLUTION_TOOL_NAMES: tuple[str, ...] = ("submit_self_evolution_patches",) COMPANY_COLLABORATION_TOOL_NAMES: tuple[str, ...] = ( *COORDINATOR_DEFAULT_TOOL_NAMES, *MEETING_RESPONSE_TOOL_NAMES, *HUMAN_REVIEW_TOOL_NAMES, + *SELF_EVOLUTION_TOOL_NAMES, ) COMPANY_DEBUG_TOOL_NAMES: tuple[str, ...] = DEBUG_ADMIN_TOOL_NAMES @@ -166,6 +168,13 @@ def _is_execute_or_review_work_item(task: object | None) -> bool: return turn_type in {"execute", "review"} +def is_self_evolution_work_item(task: object | None) -> bool: + metadata = _task_metadata(task) + if _work_item_turn_type(metadata) == "self_evolution": + return True + return bool(metadata.get("self_evolution_work_item", False)) + + def _has_active_meeting(task: object | None, runtime_state: dict[str, Any]) -> bool: metadata = _task_metadata(task) peer_wait = dict(metadata.get("peer_wait", {}) or {}) @@ -306,6 +315,8 @@ def resolve_allowed_collaboration_tools( allowed.update(MEETING_RESPONSE_TOOL_NAMES) if _human_review_close_allowed(task, state): allowed.update(HUMAN_REVIEW_TOOL_NAMES) + if is_self_evolution_work_item(task): + allowed.update(SELF_EVOLUTION_TOOL_NAMES) return allowed diff --git a/opc/engine.py b/opc/engine.py index 593bbb6..65b5f4d 100644 --- a/opc/engine.py +++ b/opc/engine.py @@ -11146,6 +11146,12 @@ class OPCEngine: return "This request is no longer active." if str(getattr(checkpoint, "status", "") or "").strip().lower() != "pending": if str(getattr(checkpoint, "checkpoint_type", "") or "").strip() == "company_delivery_feedback": + if str(getattr(checkpoint, "status", "") or "").strip().lower() == "consuming": + # Duplicate approve/feedback while the first reply is + # still driving the self-evolution run: answer + # idempotently instead of re-routing the click as a + # fresh session message. + return "Self-evolution for this delivery is already running." return None return "This request is no longer active." if not await self._checkpoint_task_still_waiting(checkpoint): @@ -12742,10 +12748,12 @@ class OPCEngine: task_brief = ( "Run employee self-evolution for this role from the completed company delivery review.\n" f"{review_text}\n\n" - "Decide whether your assigned employee should update its experience. If direct reports should also learn, " - "delegate child WorkItems with `work_kind=\"self_evolution\"`. Do not continue the original user task, " - "do not edit files, and do not produce a user-facing report. Final response must be strict JSON only: " - "`{\"patches\": [...]}`." + "Decide whether your assigned employee should update its experience, then record the result " + "by calling the `submit_self_evolution_patches` tool (an empty `patches` list means no update " + "is needed). The tool submission is the authoritative result of this turn; the final text can " + "be a brief completion note. If direct reports should also learn, delegate child WorkItems " + "with `work_kind=\"self_evolution\"`. Do not continue the original user task, do not edit " + "files, and do not produce a user-facing report." ) return make_prompt_contract( task_brief=task_brief, @@ -12757,14 +12765,15 @@ class OPCEngine: owned_outcome_kind="self_evolution", scope_key=f"self_evolution::{source.get('checkpoint_id', '')}::{role_id}", deliverables=[ - "Strict JSON only with top-level `patches` list.", + "A `submit_self_evolution_patches` tool call recording this role's patches.", "Use `patches: []` if no employee experience update is needed for this role.", "Use `delegate_work` with `work_kind=\"self_evolution\"` for direct reports that should reflect on their own work.", ], acceptance_criteria=[ - "No prose, markdown, file edits, or user-facing delivery content.", + "Patches are recorded through the `submit_self_evolution_patches` tool, not only in free text.", "Patch employee_id must be the employee assigned to this role's self-evolution work item.", "Each patch may include summary, strengths, adjustments, avoid_next_time, routing_notes, evidence_task_ids, and confidence.", + "No file edits or user-facing delivery content.", ], coordination_notes=json.dumps( { @@ -13009,6 +13018,46 @@ class OPCEngine: }) return {"recorded": recorded, "errors": errors} + # Total wall-clock budget for one delivery-review self-evolution pass + # (root reflection plus any delegated child reflections). The reply to + # the review card blocks on this pass like every other company message, + # so a stuck reflection must have a bounded lifetime. + _SELF_EVOLUTION_RUN_TIMEOUT_SEC: float = 2400.0 + + async def _settle_self_evolution_deadline(self, run_id: str) -> None: + """Cancel non-terminal self-evolution work items after the time budget.""" + if not self.store or 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("self-evolution deadline: work item load failed") + return + from opc.layer2_organization.work_item_transition import transition_work_item + + for item in list(work_items or []): + if str(getattr(item, "kind", "") or "").strip().lower() != "self_evolution": + continue + if getattr(item, "phase", None) in DONE_PHASES: + continue + try: + await transition_work_item( + self.store, + str(getattr(item, "work_item_id", "") or ""), + target_phase=Phase.CANCELLED, + reason="self_evolution_deadline", + summary="Self-evolution run exceeded its time budget and was closed.", + release_claim=True, + ) + except Exception: + logger.opt(exception=True).debug( + "self-evolution deadline: cancel failed for work item " + f"{getattr(item, 'work_item_id', '')}" + ) + async def run_company_delivery_self_evolution_checkpoint( self, checkpoint: ExecutionCheckpoint, @@ -13025,9 +13074,47 @@ class OPCEngine: ) checkpoint = await self._ensure_checkpoint_runtime_v2_payload(checkpoint) status = str(getattr(checkpoint, "status", "") or "").strip().lower() + if status == "consuming": + return "Self-evolution for this delivery is already running." if status and status != "pending": return "This self-evolution review is no longer active." + # Claim the card before spawning anything: a second approve/feedback + # while this one is processing must not re-enter and reset the live + # self-evolution work item (same defect class as duplicate resume). + claimed = await self._mark_company_runtime_checkpoint_status( + checkpoint, + status="consuming", + payload_updates={"self_evolution_claimed_at": datetime.now().isoformat()}, + expected_statuses={"pending"}, + ) + if not claimed: + return "Self-evolution for this delivery is already running." + try: + return await self._run_company_delivery_self_evolution_consumed( + checkpoint, + action=action, + feedback=feedback, + ) + except Exception as exc: + # An unexpected crash must not strand the card in "consuming" + # (that would answer every retry with "already running" forever) + # — hand the claim back so the user can retry. + await self._mark_company_runtime_checkpoint_status( + checkpoint, + status="pending", + payload_updates={"self_evolution_consume_error": str(exc)[:500]}, + expected_statuses={"consuming"}, + ) + raise + async def _run_company_delivery_self_evolution_consumed( + self, + checkpoint: ExecutionCheckpoint, + *, + action: str, + feedback: str = "", + ) -> str: + assert self.store payload = dict(checkpoint.payload or {}) waiting_task_id = str(payload.get("waiting_task_id", "") or payload.get("task_id", "") or "").strip() if not waiting_task_id: @@ -13129,11 +13216,22 @@ class OPCEngine: tasks=tasks, root_work_item=root_work_item, ) - await self.company_executor.execute(plan, tasks) + run_id = str(getattr(root_work_item, "run_id", "") or "").strip() + deadline_hit = False + try: + await asyncio.wait_for( + self.company_executor.execute(plan, tasks), + timeout=self._SELF_EVOLUTION_RUN_TIMEOUT_SEC, + ) + except asyncio.TimeoutError: + deadline_hit = True + await self._settle_self_evolution_deadline(run_id) result = await self._collect_company_self_evolution_result( checkpoint_id=checkpoint.checkpoint_id, - run_id=str(getattr(root_work_item, "run_id", "") or "").strip(), + run_id=run_id, ) + if deadline_hit: + result.setdefault("errors", []).append({"error": "self_evolution_deadline"}) waiting_task.metadata = dict(waiting_task.metadata or {}) review_record = { @@ -13165,6 +13263,12 @@ class OPCEngine: task_metadata_updates=task_metadata_updates, ) recorded_count = len(result.get("recorded", [])) + if deadline_hit: + return ( + f"Self-evolution hit its {int(self._SELF_EVOLUTION_RUN_TIMEOUT_SEC // 60)}-minute " + f"time budget and was closed with {recorded_count} recorded update(s); " + "the remaining reflection work was cancelled." + ) if recorded_count: return f"Self-evolution completed. Recorded {recorded_count} employee experience update(s)." errors = list(result.get("errors", [])) diff --git a/opc/layer2_organization/company_mode.py b/opc/layer2_organization/company_mode.py index 9bef55e..f07c390 100644 --- a/opc/layer2_organization/company_mode.py +++ b/opc/layer2_organization/company_mode.py @@ -5461,21 +5461,51 @@ class CompanyWorkItemExecutor: @staticmethod def _parse_self_evolution_patch_json(raw: str | None) -> dict[str, Any] | None: + """Extract the ``{"patches": [...]}`` object from a turn's final text. + + Native runtime turns rarely end with bare JSON: the model narrates + around it, wraps it in a markdown fence, or the runtime appends a + verification status line after it. Any of those killed the old + strict ``json.loads`` — so scan fenced blocks and every balanced + JSON object in the text, preferring the first one that carries a + ``patches`` key. + """ text = str(raw or "").strip() if not text: return None - if text.startswith("```"): - text = text.split("\n", 1)[1] if "\n" in text else text[3:] - if text.endswith("```"): - text = text[:-3] - text = text.strip() - if text.lower().startswith("json\n"): - text = text.split("\n", 1)[1].strip() - try: - parsed = json.loads(text) - except Exception: - return None - return parsed if isinstance(parsed, dict) else None + decoder = json.JSONDecoder() + + def _scan(candidate: str) -> tuple[dict[str, Any] | None, dict[str, Any] | None]: + with_patches: dict[str, Any] | None = None + bare: dict[str, Any] | None = None + for match in re.finditer(r"\{", candidate): + try: + value, _ = decoder.raw_decode(candidate, match.start()) + except Exception: + continue + if not isinstance(value, dict): + continue + if "patches" in value: + return value, bare + if bare is None: + bare = value + return with_patches, bare + + candidates = [ + fence.group(1) + for fence in re.finditer( + r"```(?:json)?\s*(.*?)```", text, flags=re.DOTALL | re.IGNORECASE + ) + ] + candidates.append(text) + first_bare: dict[str, Any] | None = None + for candidate in candidates: + found, bare = _scan(candidate) + if found is not None: + return found + if first_bare is None and bare is not None: + first_bare = bare + return first_bare @staticmethod def _self_evolution_patch_validation_error(patches: list[Any], employee_id: str) -> str: @@ -5531,15 +5561,19 @@ class CompanyWorkItemExecutor: "self_evolution_error": error_record, "self_evolution_recorded": [], }) + # Self-evolution is an opt-in reflection pass over an already + # delivered run. Settling as CANCELLED (not FAILED) keeps the + # abandoned reflection from polluting the delivered run's terminal + # verdict; the error record above preserves the diagnosis. await transition_work_item_from_task( self.store, task, - target_status_or_phase=Phase.FAILED, - reason="invalid_self_evolution_json", + target_status_or_phase=Phase.CANCELLED, + reason="self_evolution_abandoned", summary=feedback, ) await self.save_task(task) - return TaskResult(status=TaskStatus.FAILED, content=feedback, artifacts={"self_evolution_error": error_record}) + return TaskResult(status=TaskStatus.CANCELLED, content=feedback, artifacts={"self_evolution_error": error_record}) async def _finalize_self_evolution_work_item( self, @@ -5547,14 +5581,30 @@ class CompanyWorkItemExecutor: result: TaskResult, ) -> TaskResult | None: content = str(result.content or "").strip() - data = self._parse_self_evolution_patch_json(content) + # The tool submission is the authoritative channel: patches recorded + # via `submit_self_evolution_patches` survive whatever shape the + # final narration takes. Text parsing is the fallback only. + submitted = dict((task.metadata or {}).get("self_evolution_submitted_patch", {}) or {}) + if isinstance(submitted.get("patches"), list): + data: dict[str, Any] | None = {"patches": list(submitted.get("patches") or [])} + else: + data = self._parse_self_evolution_patch_json(content) patches = data.get("patches") if isinstance(data, dict) else None retry_count = int((task.metadata or {}).get("self_evolution_patch_retry_count", 0) or 0) max_retries = int((task.metadata or {}).get("self_evolution_patch_max_retries", 3) or 3) if data is None or not isinstance(patches, list): + excerpt = content[:200].replace("\n", " ") + problem = ( + "no JSON object with a top-level `patches` list was found in the final response" + if data is None + else "the JSON object found has no `patches` list" + ) feedback = ( - "Self-evolution output must be strict JSON with a top-level `patches` list. " - "Do not include prose, markdown, or delivery content." + "Self-evolution result was not machine-readable: " + f"{problem}. Received: `{excerpt}`. " + "Call the `submit_self_evolution_patches` tool with your `patches` list " + "(pass an empty list when no experience update is needed); the tool " + "records the patches regardless of your final text." ) return await self._retry_or_fail_self_evolution_output( task, @@ -5607,6 +5657,7 @@ class CompanyWorkItemExecutor: task.context_snapshot = dict(task.context_snapshot or {}) task.metadata.pop("self_evolution_patch_retry_feedback", None) task.context_snapshot.pop("self_evolution_patch_retry_feedback", None) + task.metadata.pop("self_evolution_submitted_patch", None) task.metadata["self_evolution_patch_retry_count"] = retry_count task.metadata["self_evolution_recorded"] = list(recorded) task.metadata["self_evolution_patch"] = {"patches": patches} @@ -14011,6 +14062,13 @@ class CompanyWorkItemExecutor: except Exception: logger.opt(exception=True).debug("run failure settlement: work item load failed") return + # Self-evolution items are an opt-in post-delivery reflection pass; + # their outcome must never decide the business run's terminal verdict. + work_items = [ + item + for item in work_items + if str(getattr(item, "kind", "") or "").strip().lower() != "self_evolution" + ] if not work_items: return if any(getattr(item, "phase", None) not in DONE_PHASES for item in work_items): diff --git a/opc/layer4_tools/collaboration.py b/opc/layer4_tools/collaboration.py index c6be95a..e9fd7bd 100644 --- a/opc/layer4_tools/collaboration.py +++ b/opc/layer4_tools/collaboration.py @@ -14,7 +14,10 @@ from typing import Any from loguru import logger -from opc.core.company_tools import COMPANY_COLLABORATION_TOOL_NAMES +from opc.core.company_tools import ( + COMPANY_COLLABORATION_TOOL_NAMES, + is_self_evolution_work_item, +) from opc.core.models import ( AgentMessage, CommsSemanticType, @@ -861,6 +864,16 @@ _EXTERNAL_BRIDGE_ARGUMENT_EXAMPLES: dict[str, dict[str, Any]] = { "summary": "The user accepted this delivery and no further internal work is required.", "user_message": "Acknowledged. I am closing the human review for this delivery.", }, + "submit_self_evolution_patches": { + "patches": [ + { + "summary": "One-paragraph lesson from the delivered work cycle.", + "strengths": ["Concrete behavior worth repeating"], + "adjustments": ["Concrete behavior to change next time"], + "confidence": 0.8, + } + ], + }, "send_dm": { "to_agent": "reviewer", "subject": "Need review", @@ -913,6 +926,10 @@ _EXTERNAL_BRIDGE_ARGUMENT_NOTES: dict[str, tuple[str, ...]] = { "Use only when you decide the owner-facing delivery review is complete and should be closed.", "Do not call this for requested changes; revise the board, delegate work, or respond instead.", ), + "submit_self_evolution_patches": ( + "Available only on self-evolution work items; the recorded patches are the authoritative result of the turn.", + "Pass `patches: []` when no experience update is needed; omit `employee_id` to target this work item's assigned employee.", + ), } @@ -923,6 +940,7 @@ _EXTERNAL_CLI_KEY_ARGUMENTS: dict[str, tuple[str, ...]] = { "inbox": ("action", "message_ids", "limit"), "manager_board_read": ("parent_work_item_id", "include_children"), "close_human_review": ("summary", "user_message"), + "submit_self_evolution_patches": ("patches",), "send_dm": ("to_agent", "subject", "body", "blocking", "timeout_action", "timeout_seconds"), "reply_message": ("message_id", "body", "subject"), "broadcast_issue": ("to_agents", "subject", "body", "blocking", "timeout_action", "timeout_seconds"), @@ -1602,6 +1620,51 @@ def create_collaboration_tools( "user_message": close_user_message, } + async def submit_self_evolution_patches( + patches: list[dict[str, Any]] | None = None, + task: Task | None = None, + ) -> dict[str, Any]: + role_id = _active_role(task) + if not task or not role_id: + raise ValueError("submit_self_evolution_patches requires an active assigned task") + if not is_self_evolution_work_item(task): + raise ValueError( + "submit_self_evolution_patches is only available on self-evolution work items" + ) + normalized: list[dict[str, Any]] = [] + for index, patch in enumerate(list(patches or [])): + if not isinstance(patch, dict): + raise ValueError(f"patches[{index}] must be a JSON object") + employee_id = str(patch.get("employee_id", "") or "").strip() + if not employee_id: + assignment = dict(task.metadata.get("employee_assignment", {}) or {}) + employee_id = str(assignment.get("employee_id", "") or "").strip() + if not employee_id: + raise ValueError( + f"patches[{index}] needs an employee_id and this work item has no assigned employee" + ) + normalized.append({**patch, "employee_id": employee_id}) + task.metadata = dict(task.metadata or {}) + task.context_snapshot = dict(task.context_snapshot or {}) + record = { + "patches": normalized, + "submitted_at": datetime.now().isoformat(), + "submitted_by_role": role_id, + } + task.metadata["self_evolution_submitted_patch"] = record + task.context_snapshot["self_evolution_submitted_patch"] = dict(record) + store = getattr(communication, "store", None) + if store is not None and hasattr(store, "save_task"): + await store.save_task(task) + return { + "status": "recorded", + "patch_count": len(normalized), + "note": ( + "Patches recorded for this self-evolution work item. " + "Finish the turn with a short completion note." + ), + } + async def delegate_work( items: list[dict[str, Any]], planning_context: str = "", @@ -3064,6 +3127,40 @@ def create_collaboration_tools( func=close_human_review, category="collaboration", ), + ToolDefinition( + name="submit_self_evolution_patches", + description=( + "Record employee experience patches for the current self-evolution work item. " + "This tool is the authoritative channel for self-evolution results: patches recorded " + "here are applied regardless of what the final turn text says. Pass an empty patches " + "list when no experience update is needed for this role. Each patch targets this work " + "item's assigned employee (employee_id is filled in automatically when omitted)." + ), + parameters={ + "type": "object", + "properties": { + "patches": { + "type": "array", + "items": { + "type": "object", + "properties": { + "employee_id": {"type": "string"}, + "summary": {"type": "string"}, + "strengths": {"type": "array", "items": {"type": "string"}}, + "adjustments": {"type": "array", "items": {"type": "string"}}, + "avoid_next_time": {"type": "array", "items": {"type": "string"}}, + "routing_notes": {"type": "string"}, + "evidence_task_ids": {"type": "array", "items": {"type": "string"}}, + "confidence": {"type": "number"}, + }, + }, + }, + }, + "required": ["patches"], + }, + func=submit_self_evolution_patches, + category="collaboration", + ), ToolDefinition( name="delegate_work", description=( diff --git a/tests/test_company_collaboration.py b/tests/test_company_collaboration.py index 14c05ac..1d05242 100644 --- a/tests/test_company_collaboration.py +++ b/tests/test_company_collaboration.py @@ -3369,7 +3369,11 @@ class CompanyCollaborationTests(unittest.IsolatedAsyncioTestCase): ) self.assertEqual(retry.status, TaskStatus.PENDING) self.assertEqual(task.metadata["self_evolution_patch_retry_count"], 1) - self.assertIn("strict JSON", task.context_snapshot["self_evolution_patch_retry_feedback"]) + # Retry feedback points the agent at the authoritative tool channel. + self.assertIn( + "submit_self_evolution_patches", + task.context_snapshot["self_evolution_patch_retry_feedback"], + ) save_task.assert_awaited_once() task.metadata["self_evolution_patch_retry_count"] = 2 @@ -3377,8 +3381,10 @@ class CompanyCollaborationTests(unittest.IsolatedAsyncioTestCase): task, TaskResult(status=TaskStatus.DONE, content="still not json", artifacts={}), ) - self.assertEqual(failed.status, TaskStatus.FAILED) - self.assertEqual(task.status, TaskStatus.FAILED) + # Abandoned reflection settles CANCELLED so it cannot pollute the + # delivered run's terminal verdict (the error record keeps the why). + self.assertEqual(failed.status, TaskStatus.CANCELLED) + self.assertEqual(task.status, TaskStatus.CANCELLED) self.assertEqual(task.metadata["self_evolution_error"]["attempts"], 3) async def test_self_evolution_work_item_retries_patch_for_wrong_employee(self) -> None: diff --git a/tests/test_self_evolution_pipeline.py b/tests/test_self_evolution_pipeline.py new file mode 100644 index 0000000..c486b87 --- /dev/null +++ b/tests/test_self_evolution_pipeline.py @@ -0,0 +1,413 @@ +"""Regression: delivery-review self-evolution pipeline (OBS-10). + +Approving (or sending feedback on) the delivery review card runs employee +self-evolution as company work items. Three defects made that pipeline fail +in production while plain ``ignore`` worked: + +1. Output-channel mismatch — the finalizer required the turn's FINAL chat + text to be bare JSON, but the native runtime appends a verification + status line, the manager dispatch guard displaces the final message with + a justification, and models narrate around the JSON. Fix: a dedicated + ``submit_self_evolution_patches`` tool is the authoritative channel and + the text parser scans fenced blocks / balanced objects as fallback. +2. Failure pollution — an abandoned reflection settled ``FAILED`` inside + the delivered run and dirtied its terminal verdict. Fix: settle + ``CANCELLED`` and exclude ``kind=self_evolution`` from run settlement. +3. No idempotent consumption — the card stayed ``pending`` for the whole + (potentially long) reflection run, so a duplicate approve re-entered and + reset the live work item. Fix: CAS the card to ``consuming`` before + spawning, plus a wall-clock deadline that cancels a stuck reflection. +""" +from __future__ import annotations + +import asyncio +import unittest +from pathlib import Path +from tempfile import TemporaryDirectory +from types import SimpleNamespace +from unittest.mock import AsyncMock + +from opc.core.models import ( + DelegationRun, + DelegationWorkItem, + ExecutionCheckpoint, + Task, + TaskResult, + 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 +from opc.layer2_organization.work_item_links import set_linked_work_item_id +from opc.layer4_tools.collaboration import create_collaboration_tools + + +class PatchJsonParserTests(unittest.TestCase): + def parse(self, text: str): + return CompanyWorkItemExecutor._parse_self_evolution_patch_json(text) + + def test_bare_json_still_parses(self) -> None: + parsed = self.parse('{"patches":[{"employee_id":"e1"}]}') + self.assertEqual(parsed["patches"][0]["employee_id"], "e1") + + def test_verification_status_line_suffix(self) -> None: + # runtime_v2 appends this line to every final text; it must not + # break extraction of a perfectly valid JSON payload before it. + parsed = self.parse( + '{"patches":[]}\n\nVerification: not required because no code ' + "edits or risky runtime actions were detected." + ) + self.assertEqual(parsed["patches"], []) + + def test_prose_wrapped_fenced_json(self) -> None: + text = ( + "Both children failed the strict format. I will synthesize.\n\n" + '```json\n{"patches": [{"employee_id": "ceo-1", "summary": "s"}]}\n```\n' + "Done." + ) + parsed = self.parse(text) + self.assertEqual(parsed["patches"][0]["employee_id"], "ceo-1") + + def test_prose_embedded_unfenced_json(self) -> None: + parsed = self.parse('Here is my patch: {"patches": [{"employee_id": "x"}]} recorded.') + self.assertEqual(parsed["patches"][0]["employee_id"], "x") + + def test_justification_only_text_returns_none(self) -> None: + parsed = self.parse( + "The JSON patch was delivered in the previous response.\n" + "NO_DELEGATION_JUSTIFICATION: purely local reflection." + ) + self.assertIsNone(parsed) + + def test_prefers_object_with_patches_key(self) -> None: + parsed = self.parse('{"a": 1} and later {"patches": []}') + self.assertEqual(parsed["patches"], []) + + +class SubmitPatchesToolTests(unittest.IsolatedAsyncioTestCase): + def _tool(self, store=None): + tools = create_collaboration_tools(SimpleNamespace(store=store)) + return next(t for t in tools if t.name == "submit_self_evolution_patches") + + def _task(self, *, self_evo: bool = True) -> Task: + metadata = { + "execution_mode": "company_mode", + "work_item_role_id": "cto", + "employee_assignment": {"employee_id": "emp-1", "role_id": "cto"}, + } + if self_evo: + metadata["work_item_turn_type"] = "self_evolution" + metadata["self_evolution_work_item"] = True + return Task(id="t1", title="t", project_id="p", session_id="s", metadata=metadata) + + async def test_records_patches_and_autofills_employee(self) -> None: + task = self._task() + result = await self._tool().func( + patches=[{"summary": "lesson"}], + task=task, + ) + self.assertEqual(result["status"], "recorded") + self.assertEqual(result["patch_count"], 1) + recorded = task.metadata["self_evolution_submitted_patch"] + self.assertEqual(recorded["patches"][0]["employee_id"], "emp-1") + self.assertEqual(recorded["patches"][0]["summary"], "lesson") + + async def test_empty_patch_list_is_valid(self) -> None: + task = self._task() + result = await self._tool().func(patches=[], task=task) + self.assertEqual(result["patch_count"], 0) + self.assertEqual(task.metadata["self_evolution_submitted_patch"]["patches"], []) + + async def test_rejected_outside_self_evolution_turns(self) -> None: + task = self._task(self_evo=False) + with self.assertRaises(ValueError): + await self._tool().func(patches=[], task=task) + + async def test_non_object_patch_rejected(self) -> None: + task = self._task() + with self.assertRaises(ValueError): + await self._tool().func(patches=["not-a-dict"], task=task) + + +class _EvolutionSink: + def __init__(self) -> None: + self.calls: list[dict] = [] + + def apply_employee_evolution_patch(self, **kwargs): + self.calls.append(kwargs) + return [ + {"employee_id": patch.get("employee_id", "")} + for patch in kwargs["patch"]["patches"] + ] + + +class FinalizeSelfEvolutionTests(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.executor.save_task = self.store.save_task + self.executor._emit_progress = AsyncMock() + self.executor._projection_id_for_task = lambda task: "cto::self_evolution::x" + self.sink = _EvolutionSink() + self.executor.memory = SimpleNamespace(employee_evolution=self.sink) + + async def asyncTearDown(self) -> None: + await self.store.close() + self._tmp.cleanup() + + async def _seed(self, retry_count: int = 0) -> Task: + item = DelegationWorkItem( + work_item_id="wi-se", + run_id="run-1", + role_id="cto", + kind="self_evolution", + title="Self-Evolution Review", + phase=Phase.RUNNING, + ) + await self.store.save_delegation_work_item(item) + task = Task( + id="task-se", + title="Self-Evolution Review", + project_id="p", + session_id="s", + assigned_to="cto", + metadata={ + "work_item_turn_type": "self_evolution", + "self_evolution_work_item": True, + "self_evolution_patch_retry_count": retry_count, + "self_evolution_patch_max_retries": 3, + "employee_assignment": {"employee_id": "emp-1", "role_id": "cto"}, + }, + ) + set_linked_work_item_id(task, "wi-se") + await self.store.save_task(task) + return task + + async def test_tool_submission_wins_over_prose_final_text(self) -> None: + task = await self._seed() + task.metadata["self_evolution_submitted_patch"] = { + "patches": [{"employee_id": "emp-1", "summary": "tool lesson"}], + } + result = TaskResult( + status=TaskStatus.DONE, + content="Reflection complete.\n\nVerification: not required.", + ) + outcome = await self.executor._finalize_self_evolution_work_item(task, result) + self.assertIsNone(outcome) + self.assertEqual(len(self.sink.calls), 1) + self.assertEqual( + task.metadata["self_evolution_patch"]["patches"][0]["summary"], "tool lesson" + ) + self.assertNotIn("self_evolution_submitted_patch", task.metadata) + + async def test_text_fallback_parses_fenced_json(self) -> None: + task = await self._seed() + result = TaskResult( + status=TaskStatus.DONE, + content='Summary.\n```json\n{"patches": [{"employee_id": "emp-1"}]}\n```', + ) + outcome = await self.executor._finalize_self_evolution_work_item(task, result) + self.assertIsNone(outcome) + self.assertEqual(len(self.sink.calls), 1) + + async def test_unreadable_output_retries_with_tool_instruction(self) -> None: + task = await self._seed() + result = TaskResult(status=TaskStatus.DONE, content="I finished reflecting, all good.") + outcome = await self.executor._finalize_self_evolution_work_item(task, result) + self.assertIsNotNone(outcome) + self.assertEqual(outcome.status, TaskStatus.PENDING) + feedback = task.metadata["self_evolution_patch_retry_feedback"] + self.assertIn("submit_self_evolution_patches", feedback) + self.assertIn("I finished reflecting", feedback) + + async def test_exhausted_retries_settle_cancelled_not_failed(self) -> None: + task = await self._seed(retry_count=2) + result = TaskResult(status=TaskStatus.DONE, content="still prose") + outcome = await self.executor._finalize_self_evolution_work_item(task, result) + self.assertEqual(outcome.status, TaskStatus.CANCELLED) + item = await self.store.get_delegation_work_item("wi-se") + self.assertEqual(item.phase, Phase.CANCELLED) + self.assertEqual( + dict(item.metadata or {}).get("last_transition_reason"), + "self_evolution_abandoned", + ) + self.assertEqual( + task.metadata["self_evolution_error"]["error"], "invalid_self_evolution_json" + ) + + +class RunSettlementIsolationTests(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.executor.checkpoint_callback = AsyncMock() + self.executor._emit_progress = AsyncMock() + + async def asyncTearDown(self) -> None: + await self.store.close() + self._tmp.cleanup() + + async def _seed_run(self, *, intake_phase: Phase, selfevo_phase: Phase | None) -> 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="Intake", phase=Phase.READY, + ) + await self.store.save_delegation_work_item(intake) + if intake_phase is not Phase.READY: + await self.store.update_delegation_work_item("wi-intake", phase=intake_phase) + if selfevo_phase is not None: + se = DelegationWorkItem( + work_item_id="wi-se", run_id="run-1", role_id="ceo", + kind="self_evolution", title="Self-Evolution", phase=Phase.READY, + ) + await self.store.save_delegation_work_item(se) + if selfevo_phase is not Phase.READY: + await self.store.update_delegation_work_item("wi-se", phase=selfevo_phase) + task = Task( + id="task-1", title="Intake", project_id="p", session_id="s", + metadata={"delegation_run_id": "run-1", "original_request": "goal"}, + ) + await self.store.save_task(task) + return [task] + + async def test_running_selfevo_item_does_not_block_failure_settlement(self) -> None: + tasks = await self._seed_run(intake_phase=Phase.FAILED, selfevo_phase=Phase.RUNNING) + await self.executor._settle_run_lifecycle_on_convergence(tasks) + run = await self.store.get_delegation_run("run-1") + self.assertEqual(run.lifecycle_status, "closed_failed") + + async def test_cancelled_selfevo_item_never_fails_a_delivered_run(self) -> None: + tasks = await self._seed_run(intake_phase=Phase.RUNNING, selfevo_phase=Phase.CANCELLED) + await self.store.update_delegation_work_item("wi-intake", phase=Phase.APPROVED) + 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.executor.checkpoint_callback.assert_not_called() + + +class DeliveryFeedbackConsumingTests(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 + + async def asyncTearDown(self) -> None: + await self.store.close() + self._tmp.cleanup() + + async def _seed_checkpoint(self, *, status: str = "pending") -> ExecutionCheckpoint: + checkpoint = ExecutionCheckpoint( + checkpoint_id="ckpt-fb", + project_id="p", + session_id="s", + checkpoint_type="company_delivery_feedback", + task_id="task-1", + status=status, + payload={"session_id": "s"}, + ) + await self.store.save_execution_checkpoint(checkpoint) + return checkpoint + + async def test_concurrent_approve_claims_exactly_once(self) -> None: + await self._seed_checkpoint() + # Both controllers hold a pending copy of the card before either + # acts — the DB-level CAS must let exactly one proceed. The payload + # has no waiting_task_id, so the winner exits right after claiming, + # which keeps the race observable without a full runtime. + first = await self.engine._load_execution_checkpoint_by_id("ckpt-fb") + second = await self.engine._load_execution_checkpoint_by_id("ckpt-fb") + replies = await asyncio.gather( + self.engine.run_company_delivery_self_evolution_checkpoint(first, action="approve"), + self.engine.run_company_delivery_self_evolution_checkpoint(second, action="approve"), + ) + already = [r for r in replies if r == "Self-evolution for this delivery is already running."] + proceeded = [r for r in replies if "delivery task reference is missing" in r] + self.assertEqual(len(already), 1, replies) + self.assertEqual(len(proceeded), 1, replies) + + async def test_duplicate_reply_while_consuming_is_idempotent(self) -> None: + await self._seed_checkpoint(status="consuming") + reply = await self.engine._maybe_resume_checkpoint( + "approve", + session_id="s", + reply_metadata={"response_to_checkpoint_id": "ckpt-fb"}, + ) + self.assertEqual(reply, "Self-evolution for this delivery is already running.") + + async def test_crash_hands_the_claim_back_for_retry(self) -> None: + await self._seed_checkpoint() + + async def _boom(checkpoint, *, action, feedback=""): + raise RuntimeError("mid-flight crash") + + self.engine._run_company_delivery_self_evolution_consumed = _boom + loaded = await self.engine._load_execution_checkpoint_by_id("ckpt-fb") + with self.assertRaises(RuntimeError): + await self.engine.run_company_delivery_self_evolution_checkpoint( + loaded, action="approve" + ) + refreshed = await self.engine._load_execution_checkpoint_by_id("ckpt-fb") + self.assertEqual(refreshed.status, "pending") + self.assertIn( + "mid-flight crash", + str(dict(refreshed.payload or {}).get("self_evolution_consume_error", "")), + ) + + +class SelfEvolutionDeadlineTests(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 + + async def asyncTearDown(self) -> None: + await self.store.close() + self._tmp.cleanup() + + async def test_deadline_cancels_only_selfevo_items(self) -> None: + for work_item_id, kind, phase in ( + ("wi-se-1", "self_evolution", Phase.RUNNING), + ("wi-se-2", "self_evolution", Phase.READY), + ("wi-se-3", "self_evolution", Phase.APPROVED), + ("wi-exec", "execute", Phase.RUNNING), + ): + item = DelegationWorkItem( + work_item_id=work_item_id, run_id="run-1", role_id="cto", + kind=kind, title=work_item_id, phase=Phase.READY, + ) + await self.store.save_delegation_work_item(item) + if phase is not Phase.READY: + await self.store.update_delegation_work_item(work_item_id, phase=Phase.RUNNING) + if phase not in (Phase.READY, Phase.RUNNING): + await self.store.update_delegation_work_item(work_item_id, phase=phase) + + await self.engine._settle_self_evolution_deadline("run-1") + + expectations = { + "wi-se-1": Phase.CANCELLED, + "wi-se-2": Phase.CANCELLED, + "wi-se-3": Phase.APPROVED, + "wi-exec": Phase.RUNNING, + } + for work_item_id, expected in expectations.items(): + item = await self.store.get_delegation_work_item(work_item_id) + self.assertEqual(item.phase, expected, work_item_id) + + +if __name__ == "__main__": + unittest.main()