diff --git a/opc/engine.py b/opc/engine.py index 6e35396..1e5a17e 100644 --- a/opc/engine.py +++ b/opc/engine.py @@ -103,6 +103,7 @@ from opc.layer2_organization.company_runtime_identity import ( load_company_runtime_identity_index, ) from opc.layer2_organization.metadata_ownership import ( + build_company_resume_identity_restore, build_work_item_owner_execution_copy, ) from opc.layer2_organization.phase import ( @@ -6253,13 +6254,13 @@ class OPCEngine: identity.get("selected_execution_agent_source", "") or "" ).strip() - metadata["work_item_role_id"] = identity["role_id"] or task_role_id - if identity["seat_id"]: - metadata["delegation_seat_id"] = identity["seat_id"] - if identity["role_runtime_session_id"]: - metadata["delegation_role_session_id"] = identity[ - "role_runtime_session_id" - ] + metadata.update( + build_company_resume_identity_restore( + role_id=identity["role_id"] or task_role_id, + seat_id=identity["seat_id"], + role_runtime_session_id=identity["role_runtime_session_id"], + ) + ) if identity.get("explicit") or identity["employee_assignment"]: metadata["employee_assignment"] = copy.deepcopy( identity["employee_assignment"] diff --git a/opc/layer2_organization/metadata_ownership.py b/opc/layer2_organization/metadata_ownership.py index b5ef7b6..d80759f 100644 --- a/opc/layer2_organization/metadata_ownership.py +++ b/opc/layer2_organization/metadata_ownership.py @@ -387,6 +387,33 @@ def build_work_item_owner_execution_copy(work_item: DelegationWorkItem | None) - } +def build_company_resume_identity_restore( + *, + role_id: str, + seat_id: str, + role_runtime_session_id: str, +) -> dict[str, Any]: + """Map checkpoint-validated resume identity onto Task execution-copy keys. + + Company resume restores missing Task projection fields from the durable + checkpoint after the authoritative WorkItem has been cross-checked (the + checkpoint may hold values the WorkItem row lacks, so this is not always + derivable via ``build_work_item_owner_execution_copy``). The key spelling + lives here, next to the owner spec, so runtime code never hand-writes + WorkItem-owned execution-copy keys. + """ + payload: dict[str, Any] = { + "work_item_role_id": str(role_id or "").strip(), + } + seat = str(seat_id or "").strip() + if seat: + payload["delegation_seat_id"] = seat + session = str(role_runtime_session_id or "").strip() + if session: + payload["delegation_role_session_id"] = session + return payload + + def strip_disallowed_work_item_metadata_from_runtime_task(task: Task) -> list[str]: """Remove WorkItem-owned fields that are not valid Task execution copies.""" metadata = dict(getattr(task, "metadata", {}) or {}) diff --git a/opc/layer2_organization/work_item_transition.py b/opc/layer2_organization/work_item_transition.py index 1a5ee89..d1c433d 100644 --- a/opc/layer2_organization/work_item_transition.py +++ b/opc/layer2_organization/work_item_transition.py @@ -102,6 +102,8 @@ async def transition_work_item( metadata_updates: dict[str, Any] | None = None, release_claim: bool = False, attempt_outcome: str | None = None, + blocked_reason: str | None = None, + handoff_status: str | None = None, ) -> DelegationWorkItem | None: """Transition a work item to ``target_phase``. @@ -124,6 +126,10 @@ async def transition_work_item( metadata_updates: Extra metadata keys to merge onto the work item. release_claim: When True, clears the current claim so the dispatcher can re-acquire. Useful for cancel / timeout / forced-release paths. + blocked_reason: Optional ``blocked_reason`` column value, folded into + the same write as the phase change (pass ``""`` to clear). + handoff_status: Optional ``handoff_status`` column value, folded into + the same write as the phase change. Returns: The updated ``DelegationWorkItem``, or ``None`` when the store lacks @@ -166,6 +172,10 @@ async def transition_work_item( } if summary is not None: kwargs["summary"] = summary + if blocked_reason is not None: + kwargs["blocked_reason"] = blocked_reason + if handoff_status is not None: + kwargs["handoff_status"] = handoff_status if release_claim: # Fold claim release into the same write as the phase change: the # legacy two-call sequence could commit the phase and then fail the diff --git a/opc/layer4_tools/collaboration.py b/opc/layer4_tools/collaboration.py index 657297b..c6be95a 100644 --- a/opc/layer4_tools/collaboration.py +++ b/opc/layer4_tools/collaboration.py @@ -45,6 +45,7 @@ from opc.layer2_organization.work_item_transition import ( is_prunable_dependency_work_item, normalize_dependency_work_item_ids, refresh_dependents_for_run, + transition_work_item, ) from opc.layer4_tools.output_budget import clip_text from opc.layer4_tools.registry import ToolDefinition @@ -2689,33 +2690,32 @@ def create_collaboration_tools( claimed_by_seat_id="", ) if descendant.phase not in DONE_PHASES: - await store.update_delegation_work_item( + await transition_work_item( + store, descendant.work_item_id, - phase=Phase.CANCELLED, + target_phase=Phase.CANCELLED, + reason=descendant_reason, + release_claim=True, blocked_reason=descendant_reason[:500], handoff_status="cancelled", - claimed_by_role_runtime_session_id="", - claimed_by_seat_id="", ) + # Task.status is projected from the phase by the + # transition hooks; here we only stamp the task-owned + # audit trail and release the execution lock. + if callable(get_runtime_task): + runtime_task = await get_runtime_task(descendant.work_item_id) + if runtime_task is not None: + runtime_task.execution_lock = False + runtime_task.execution_locked_at = None + runtime_task.metadata = { + **dict(runtime_task.metadata or {}), + "last_stop_reason": "manager_deleted_ancestor_work_item", + "deleted_by_manager_tool": True, + "cascade_deleted_by_work_item_id": item.work_item_id, + } + if hasattr(store, "save_task"): + await store.save_task(runtime_task) cascade_deleted_ids.append(descendant.work_item_id) - if callable(get_runtime_task): - runtime_task = await get_runtime_task(descendant.work_item_id) - if runtime_task is not None and runtime_task.status not in { - TaskStatus.DONE, - TaskStatus.FAILED, - TaskStatus.CANCELLED, - }: - runtime_task.status = TaskStatus.CANCELLED - runtime_task.execution_lock = False - runtime_task.execution_locked_at = None - runtime_task.metadata = { - **dict(runtime_task.metadata or {}), - "last_stop_reason": "manager_deleted_ancestor_work_item", - "deleted_by_manager_tool": True, - "cascade_deleted_by_work_item_id": item.work_item_id, - } - if hasattr(store, "save_task"): - await store.save_task(runtime_task) except Exception: logger.opt(exception=True).warning( "delete_work_item: failed to cascade delete descendant {}",