diff --git a/opc/engine.py b/opc/engine.py index 12d0c99..1967013 100644 --- a/opc/engine.py +++ b/opc/engine.py @@ -37,6 +37,7 @@ from opc.core.config import ( company_org_path, get_opc_home, get_project_workplace, + validate_organization_id, ) from opc.core.events import EventBus from opc.core.models import ( @@ -3563,6 +3564,39 @@ class OPCEngine: return False return bool(dict(getattr(task, "metadata", {}) or {}).get("shared_role_session", False)) + @staticmethod + def _normalize_durable_org_id(value: Any) -> str: + try: + return validate_organization_id(value) + except ValueError: + return "" + + @staticmethod + def _runtime_org_id_for_identity( + decision: RouterDecision | None, + metadata: dict[str, Any] | None, + org_config: Any | None, + ) -> str | None: + """Return the durable custom-org ID for a company runtime task.""" + task_metadata = dict(metadata or {}) + profile = str( + getattr(decision, "company_profile", "") + or task_metadata.get("company_profile", "") + or getattr(org_config, "company_profile", "") + or "" + ).strip().lower() + if profile != "custom": + return None + for candidate in ( + getattr(decision, "org_id", None), + task_metadata.get("org_id"), + task_metadata.get("organization_id"), + ): + normalized = str(candidate or "").strip() + if normalized: + return normalized + return None + @staticmethod def _shared_company_role_session_id( parent_session_id: str, @@ -3601,6 +3635,17 @@ class OPCEngine: root_session: bool = False, ) -> Task: assert self.store and self.memory + runtime_company_profile = str( + getattr(decision, "company_profile", "") + or (work_item.metadata or {}).get("company_profile", "") + or getattr(getattr(self.config, "org", None), "company_profile", "") + or "" + ).strip().lower() + runtime_org_id = self._runtime_org_id_for_identity( + decision, + getattr(work_item, "metadata", None), + getattr(getattr(self, "config", None), "org", None), + ) role_id = str(work_item.role_id or "").strip() seat_id = str((work_item.metadata or {}).get("seat_id", "") or "").strip() team_id = str((work_item.metadata or {}).get("team_id", "") or work_item.cell_id or "").strip() @@ -3653,6 +3698,42 @@ class OPCEngine: set_linked_work_item_id(existing, work_item.work_item_id) existing.session_id = session_id existing.metadata = dict(existing.metadata or {}) + if runtime_company_profile == "custom": + persisted_org_id = self._normalize_durable_org_id(getattr(existing, "org_id", None)) + incoming_org_id = self._normalize_durable_org_id(runtime_org_id) + if persisted_org_id and incoming_org_id and persisted_org_id != incoming_org_id: + from opc.plugins.office_ui.services.models import ServiceError + raise ServiceError( + "org_id_conflict", + "org_id_conflict", + { + "project_id": self.project_id or "default", + "task_id": str(getattr(work_item, "work_item_id", "") or ""), + "persisted_org_id": persisted_org_id, + "incoming_org_id": incoming_org_id, + "reason": "custom_company_run_org_id_conflict", + }, + ) + resolved_org_id = persisted_org_id or incoming_org_id + if not resolved_org_id: + from opc.plugins.office_ui.services.models import ServiceError + raise ServiceError( + "org_id_required", + "org_id_required", + { + "project_id": self.project_id or "default", + "task_id": str(getattr(work_item, "work_item_id", "") or ""), + "reason": "custom_company_run_requires_durable_org_id", + }, + ) + runtime_org_id = resolved_org_id + existing.org_id = resolved_org_id + existing.metadata["org_id"] = resolved_org_id + existing.metadata["organization_id"] = resolved_org_id + elif runtime_org_id: + existing.org_id = runtime_org_id + existing.metadata["org_id"] = runtime_org_id + existing.metadata["organization_id"] = runtime_org_id existing.metadata["shared_role_session"] = True existing.metadata["shared_role_id"] = role_id existing.metadata["company_runtime_root_session_id"] = parent_session_id @@ -3694,6 +3775,17 @@ class OPCEngine: ) await self.store.save_task(existing) return existing + if runtime_company_profile == "custom" and not runtime_org_id: + from opc.plugins.office_ui.services.models import ServiceError + raise ServiceError( + "org_id_required", + "org_id_required", + { + "project_id": self.project_id or "default", + "task_id": str(getattr(work_item, "work_item_id", "") or ""), + "reason": "custom_company_run_requires_durable_org_id", + }, + ) employee_assignment = dict(topology_seat.get("employee_assignment", {}) or {}) if not employee_assignment and self.org_engine and role_id: preferred_employee_id = str(topology_seat.get("employee_id", "") or "").strip() or None @@ -3744,6 +3836,18 @@ class OPCEngine: owner_execution_copy = build_work_item_owner_execution_copy(work_item) owner_execution_copy.setdefault("delegation_role_session_id", role_session_id) owner_execution_copy["work_kind"] = work_item_turn_type + runtime_identity_metadata = ( + { + "org_id": runtime_org_id or "", + "organization_id": runtime_org_id or "", + } + if runtime_company_profile == "custom" + else { + "organization_id": str( + getattr(getattr(self.config, "org", None), "organization_id", "") or "" + ).strip(), + } + ) task = Task( title=str(work_item.title or work_item_projection_ref or "Runtime Work Item").strip(), description=( @@ -3757,6 +3861,7 @@ class OPCEngine: session_id=session_id, parent_session_id=parent_session_id, assigned_external_agent=assigned_external_agent, + org_id=runtime_org_id, metadata=mark_work_item_projection(mark_work_item_runtime({ "mode": "company", "execution_mode": decision.mode.value, @@ -3765,7 +3870,6 @@ class OPCEngine: "original_message": original_message, "router_preferred_agent": decision.preferred_agent, "company_profile": decision.company_profile or getattr(self.config.org, "company_profile", "corporate"), - "organization_id": getattr(self.config.org, "organization_id", ""), "organization_name": getattr(self.config.org, "organization_name", ""), "organization_config_file": getattr(self.config.org, "organization_config_file", ""), "delegation_playbook": dict(delegation_playbook), @@ -3780,6 +3884,7 @@ class OPCEngine: ), "runtime_topology": copy.deepcopy(runtime_topology), **owner_execution_copy, + **runtime_identity_metadata, "work_item_projection_ref": work_item_projection_ref, "seat_manager_role_id": str(topology_seat.get("manager_role_id", "") or "").strip(), "manager_role_id": str(topology_seat.get("manager_role_id", "") or "").strip(), @@ -12649,7 +12754,12 @@ class OPCEngine: seen_employee_ids.add(employee_id) history = "" if self.memory: - organization_id = str(getattr(getattr(self.config, "org", None), "organization_id", "") or "").strip() + organization_id = str( + getattr(delivery_task, "org_id", "") + or (delivery_task.metadata or {}).get("org_id") + or (delivery_task.metadata or {}).get("organization_id") + or "" + ).strip() history = self.memory.employee_evolution.build_employee_delta_context( employee_id, project_id=task.project_id, @@ -13191,13 +13301,19 @@ class OPCEngine: await self._mark_company_runtime_checkpoint_status(checkpoint, status="invalid") return "Could not run self-evolution because the runtime task set could not be restored." + from opc.plugins.office_ui.execution_identity import resolve_delivery_task_org_identity + + organization_id, identity_error = resolve_delivery_task_org_identity( + waiting_task, + payload=payload, + active_org_id=getattr(getattr(self.config, "org", None), "organization_id", ""), + default_org_id=DEFAULT_ORGANIZATION_ID, + ) + if identity_error: + await self._mark_company_runtime_checkpoint_status(checkpoint, status="invalid") + return f"Could not run self-evolution because {identity_error}." + plan = deserialize_company_work_item_runtime_plan(payload.get("company_work_item_plan") or payload.get("plan", {})) - organization_id = str( - getattr(waiting_task, "org_id", "") - or payload.get("organization_id") - or getattr(getattr(self.config, "org", None), "organization_id", "") - or DEFAULT_ORGANIZATION_ID - ).strip() or DEFAULT_ORGANIZATION_ID root_role_id = str( getattr(plan, "final_decider_role_id", "") or plan.metadata.get("final_decider_role_id", "") diff --git a/opc/layer2_organization/company_mode.py b/opc/layer2_organization/company_mode.py index eafea48..9e8dbf2 100644 --- a/opc/layer2_organization/company_mode.py +++ b/opc/layer2_organization/company_mode.py @@ -22,7 +22,11 @@ from opc.core.active_task_runs import ( ActiveTaskRunAdmissionClosed, ActiveTaskRunRegistry, ) -from opc.core.config import DEFAULT_EXTERNAL_AGENT_STARTUP_TIMEOUT_SECONDS, DEFAULT_ORGANIZATION_ID +from opc.core.config import ( + DEFAULT_EXTERNAL_AGENT_STARTUP_TIMEOUT_SECONDS, + DEFAULT_ORGANIZATION_ID, + validate_organization_id, +) from opc.core.models import ( AdaptiveRoleProfile, AdaptiveSignalSpec, @@ -1326,6 +1330,26 @@ class CompanyRuntimeSpecBuilder(CompanyRuntimeWorkItemHelper): or "corporate" ).strip() or "corporate" org_config = getattr(self.org_engine.config, "org", None) + selected_org_id = "" + if profile == "custom": + try: + selected_org_id = validate_organization_id(getattr(decision, "org_id", None)) + except ValueError: + selected_org_id = "" + if not selected_org_id: + # A custom-organization run must carry a durable org_id on the + # decision. Never derive it from the process-wide active + # config; fail closed before any work items are created. + from opc.plugins.office_ui.services.models import ServiceError + raise ServiceError( + "org_id_required", + "org_id_required", + { + "company_profile": profile, + "reason": "custom_company_run_requires_durable_org_id", + }, + ) + decision.org_id = selected_org_id metadata: dict[str, Any] = { "source": "work_item_runtime", "execution_mode": "company_mode", @@ -1333,7 +1357,11 @@ class CompanyRuntimeSpecBuilder(CompanyRuntimeWorkItemHelper): "runtime_model": "multi_team_org", "work_item_driven": True, "company_profile": profile, - "organization_id": str(getattr(org_config, "organization_id", "") or "").strip(), + "organization_id": ( + selected_org_id + if profile == "custom" + else str(getattr(org_config, "organization_id", "") or "").strip() + ), "organization_name": str(getattr(org_config, "organization_name", "") or "").strip(), "organization_config_file": str(getattr(org_config, "organization_config_file", "") or "").strip(), "original_request": original_message, @@ -1341,7 +1369,7 @@ class CompanyRuntimeSpecBuilder(CompanyRuntimeWorkItemHelper): "domains": list(getattr(decision, "domains", []) or []), "preferred_agent": getattr(decision, "preferred_agent", None), "requested_sub_tasks": list(getattr(decision, "sub_tasks", []) or []), - "org_id": getattr(decision, "org_id", None), + "org_id": selected_org_id if profile == "custom" else getattr(decision, "org_id", None), } return CompanyRuntimeSpec( profile=profile, @@ -4247,7 +4275,21 @@ class CompanyWorkItemExecutor: existing_task_ids = {str(task.id or "").strip() for task in existing_tasks if str(task.id or "").strip()} existing_work_item_ids = set(task_by_linked_work_item_id(existing_tasks)) root_task = sorted(existing_tasks, key=lambda item: (item.created_at, item.id))[0] - runtime_topology = dict((root_task.metadata or {}).get("runtime_topology", {}) or {}) + root_metadata = dict(root_task.metadata or {}) + custom_runtime = str(root_metadata.get("company_profile", "") or "").strip().lower() == "custom" + runtime_org_id = str( + getattr(root_task, "org_id", "") + or root_metadata.get("org_id") + or root_metadata.get("organization_id") + or "" + ).strip() or None + if not custom_runtime: + runtime_org_id = None + if runtime_org_id: + for existing_task in existing_tasks: + if self._sync_runtime_org_identity(existing_task, runtime_org_id): + await self.store.save_task(existing_task) + runtime_topology = dict(root_metadata.get("runtime_topology", {}) or {}) root_parent_session_id = str( root_task.parent_session_id or root_task.session_id @@ -4283,6 +4325,8 @@ class CompanyWorkItemExecutor: persisted = await get_runtime_task(work_item_id) if persisted is not None: set_linked_work_item_id(persisted, work_item_id) + if self._sync_runtime_org_identity(persisted, runtime_org_id): + await self.store.save_task(persisted) self._raise_for_runtime_projection_issues(persisted, work_item, work_item_by_id) if persisted.id not in existing_task_ids: existing_tasks.append(persisted) @@ -4413,6 +4457,9 @@ class CompanyWorkItemExecutor: task_metadata.update(copy_work_item_execution_metadata(work_item)) task_metadata.update(owner_execution_copy) task_metadata[WORK_ITEM_TURN_TYPE_KEY] = turn_type + if custom_runtime: + task_metadata["org_id"] = runtime_org_id or "" + task_metadata["organization_id"] = runtime_org_id or "" temp_task = Task( id=str(uuid.uuid4()), title=str(getattr(work_item, "title", "") or projection_id or "Runtime Work Item").strip(), @@ -4423,6 +4470,7 @@ class CompanyWorkItemExecutor: session_id=session_id, parent_session_id=root_parent_session_id, assigned_external_agent=assigned_external_agent, + org_id=runtime_org_id, metadata=task_metadata, ) dependency_projection_ids: list[str] = [] @@ -4454,6 +4502,7 @@ class CompanyWorkItemExecutor: parent_session_id=temp_task.parent_session_id, assigned_external_agent=temp_task.assigned_external_agent, dependencies=dependency_projection_ids, + org_id=runtime_org_id, metadata=task_metadata, ) set_linked_work_item_id(task, work_item_id) @@ -4471,6 +4520,8 @@ class CompanyWorkItemExecutor: "failed to link new runtime Task " f"{task.id} for WorkItem {work_item_id}" ) + if self._sync_runtime_org_identity(task, runtime_org_id): + await self.store.save_task(task) set_linked_work_item_id(task, work_item_id) self._raise_for_runtime_projection_issues(task, work_item, work_item_by_id) if self.memory is not None and task.session_id: @@ -4510,6 +4561,23 @@ class CompanyWorkItemExecutor: await self.save_task(task) return existing_tasks + @staticmethod + def _sync_runtime_org_identity(task: Task, org_id: str | None) -> bool: + normalized_org_id = str(org_id or "").strip() + if not normalized_org_id: + return False + metadata = dict(task.metadata or {}) + changed = str(getattr(task, "org_id", "") or "").strip() != normalized_org_id + changed = changed or metadata.get("org_id") != normalized_org_id + changed = changed or metadata.get("organization_id") != normalized_org_id + if not changed: + return False + task.org_id = normalized_org_id + metadata["org_id"] = normalized_org_id + metadata["organization_id"] = normalized_org_id + task.metadata = metadata + return True + @staticmethod def _runtime_work_kind_to_work_item_turn_type(work_kind: str) -> str: return canonical_work_item_turn_type_for_kind(work_kind) diff --git a/opc/layer2_organization/custom_runtime.py b/opc/layer2_organization/custom_runtime.py index 24aa5e8..192c8e9 100644 --- a/opc/layer2_organization/custom_runtime.py +++ b/opc/layer2_organization/custom_runtime.py @@ -64,8 +64,22 @@ class CustomRuntimeRunner: ) -> str: from opc.engine import OPCEngine from opc.layer2_organization.phase_hooks import unregister_dispatcher_wake + from opc.plugins.office_ui.services.models import ServiceError - org_config, resolved_org_id = self._build_org_config(org_id) + normalized_org_id = str(org_id or "").strip() + if not normalized_org_id: + # Isolated org mode must carry a durable org_id; resolving the + # active index here would silently route the run to whichever + # organization is currently loaded. + raise ServiceError( + "org_id_required", + "org_id_required", + { + "project_id": project_id or self.parent.project_id or "default", + "reason": "custom_company_run_requires_durable_org_id", + }, + ) + org_config, resolved_org_id = self._build_org_config(normalized_org_id) normalized_project_id = str(project_id or self.parent.project_id or "default").strip() or "default" shared_store = getattr(self.parent, "store", None) runtime = OPCEngine( diff --git a/opc/plugins/office_ui/execution_identity.py b/opc/plugins/office_ui/execution_identity.py index aaab91a..b902130 100644 --- a/opc/plugins/office_ui/execution_identity.py +++ b/opc/plugins/office_ui/execution_identity.py @@ -14,9 +14,10 @@ for that identity: from __future__ import annotations from dataclasses import dataclass -from typing import Any +from typing import Any, Mapping from opc.core.config import validate_organization_id +from opc.layer2_organization.company_runtime_identity import is_company_runtime_task PREFERRED_AGENTS: frozenset[str] = frozenset({ "native", @@ -154,11 +155,12 @@ def execution_identity_from_task( company_profile = metadata.get("company_profile") metadata_profile = str(company_profile or "").strip().lower() metadata_org_id = ( - metadata.get("org_id") + getattr(task, "org_id", None) + or metadata.get("org_id") or metadata.get("organization_id") - or getattr(task, "org_id", None) or "" ) + mode_hint = str(metadata.get("mode", "") or "").strip().lower() if raw_exec_mode: exec_mode = raw_exec_mode @@ -169,6 +171,15 @@ def execution_identity_from_task( elif execution_mode == "company_mode" or metadata_profile: exec_mode = "company" explicit = True + elif ( + mode_hint in {"company", "org", "custom"} + or is_company_runtime_task(task) + ): + # Older company/runtime rows may only retain a mode marker or the + # runtime marker itself. Treat those rows as explicit company + # identity so they cannot fall through to task-mode/global defaults. + exec_mode = "org" if metadata_org_id else "company" + explicit = True elif metadata_org_id: exec_mode = "org" explicit = True @@ -187,3 +198,46 @@ def execution_identity_from_task( default_preferred_agent=default_preferred_agent, explicit_exec_mode=explicit, ) + + +def resolve_delivery_task_org_identity( + task: Any | None, + *, + payload: Mapping[str, Any] | None = None, + active_org_id: Any = "", + default_org_id: Any = "", +) -> tuple[str, str]: + """Validate the org identity of a delivery self-evolution task. + + Returns ``(organization_id, error)`` with at most one non-empty. Prefers + ``Task.org_id``, then task metadata org fields; conflicting sources are + rejected. Checkpoint-payload org fields are a last-resort legacy fallback + and never override task/metadata identity. The active configuration org + is only consulted for a confirmed corporate task; custom-org deliveries + without a durable org id fail closed. + """ + metadata = task_metadata(task) + candidates: list[str] = [] + for value in ( + getattr(task, "org_id", None), + metadata.get("org_id"), + metadata.get("organization_id"), + ): + normalized = normalize_org_id(value) + if normalized and normalized not in candidates: + candidates.append(normalized) + if len(candidates) > 1: + return "", "the delivery task org identity conflicts across task and metadata sources" + task_org_id = candidates[0] if candidates else "" + if task_org_id: + return task_org_id, "" + payload_org_id = normalize_org_id( + (payload or {}).get("org_id") + or (payload or {}).get("organization_id") + ) + if payload_org_id: + return payload_org_id, "" + identity = execution_identity_from_task(task) + if identity.is_company: + return normalize_org_id(active_org_id) or normalize_org_id(default_org_id), "" + return "", "the custom-organization delivery task has no durable org identity" diff --git a/opc/plugins/office_ui/services/session.py b/opc/plugins/office_ui/services/session.py index f980f19..778b52e 100644 --- a/opc/plugins/office_ui/services/session.py +++ b/opc/plugins/office_ui/services/session.py @@ -1163,6 +1163,12 @@ class SessionService: else: checkpoint = None org_id = self.resolve_task_org_id(config_task) if engine_mode == "org" else "" + if engine_mode == "org" and not org_id: + raise ServiceError( + "org_id_required", + "org_id_required", + {"project_id": project_id, "task_id": resolved_task_id}, + ) message_metadata: dict[str, Any] = {"ui_force_resume": True} if checkpoint is not None: message_metadata.update({ diff --git a/opc/plugins/office_ui/ws_handler.py b/opc/plugins/office_ui/ws_handler.py index b3faa54..51e9e7a 100644 --- a/opc/plugins/office_ui/ws_handler.py +++ b/opc/plugins/office_ui/ws_handler.py @@ -4967,6 +4967,50 @@ class WSHandler: def _resolve_task_org_id(self, task: Any | None) -> str: return self._ensure_office_services().session.resolve_task_org_id(task) + async def _resolve_session_runtime_config_task( + self, + task_id: str, + task: Any | None, + *, + engine: Any, + ) -> Any | None: + """Resolve company-session config from the durable runtime identity.""" + if task is None: + return None + exec_mode, _ = self._resolve_task_session_config(task) + runtime_bound = is_company_runtime_task(task) + if not runtime_bound and not self._is_company_session_exec_mode(exec_mode): + return task + try: + target = await self._resolve_company_runtime_target(task_id, engine=engine) + except ServiceError: + if runtime_bound: + raise + logger.opt(exception=True).debug( + "failed to resolve durable session config task" + ) + return task + except Exception as exc: + if runtime_bound: + raise ServiceError( + "company_runtime_identity_mismatch", + "Company runtime identity could not be resolved", + {"task_id": str(task_id or "").strip()}, + ) from exc + logger.opt(exception=True).debug( + "failed to resolve durable session config task" + ) + return task + if target is not None: + return target.get("config_task") or task + if runtime_bound: + raise ServiceError( + "company_runtime_identity_mismatch", + "Company runtime identity could not be resolved", + {"task_id": str(task_id or "").strip()}, + ) + return task + @staticmethod def _is_company_session_exec_mode(exec_mode: Any) -> bool: return str(exec_mode or "").strip().lower() in {"company", "org", "custom"} @@ -5037,14 +5081,24 @@ class WSHandler: # Look up session_id from task session_id: str | None = None task = None + config_task = None preferred_agent = self._task_preferred_agent session_org_id = self._normalize_session_org_id(org_id) if task_id and getattr(engine, "store", None): task = await engine.store.get_task(task_id) if task: session_id = task.session_id - identity = self._resolve_task_identity( + + company_runtime_target: dict[str, Any] | None = None + try: + if task is not None: + config_task = await self._resolve_session_runtime_config_task( + task_id, task, + engine=engine, + ) + identity = self._resolve_task_identity( + config_task, default_exec_mode=mode, default_company_profile=profile, default_preferred_agent=preferred_agent, @@ -5054,9 +5108,12 @@ class WSHandler: profile = identity.company_profile session_org_id = identity.org_id preferred_agent = identity.preferred_agent - - company_runtime_target: dict[str, Any] | None = None - try: + if identity.is_custom_org and not identity.org_id: + raise ServiceError( + "org_id_required", + "org_id_required", + {"project_id": pid, "task_id": task_id}, + ) content = f"{title}\n{description}".strip() engine_mode, company_profile = self._resolve_engine_mode(mode, profile) engine_preferred_agent = preferred_agent if engine_mode == "project" else None @@ -7797,6 +7854,16 @@ class WSHandler: parent_task = await run_engine.store.get_task(parent_task_id) except Exception: logger.opt(exception=True).debug("failed to load parent task for delivery feedback reply") + if parent_task is None: + raise ServiceError( + "org_id_required", + "org_id_required", + { + "project_id": pid, + "task_id": parent_task_id, + "reason": "delivery_feedback_requires_durable_parent_task", + }, + ) session_exec_mode = self._normalize_session_exec_mode(self._exec_mode) session_company_profile = self._normalize_session_company_profile(self._company_profile) session_org_id = "" @@ -8102,17 +8169,6 @@ class WSHandler: pid = self._normalize_project_id(run_project_id or getattr(run_engine, "project_id", None)) async with lock: try: - try: - await self._set_company_runtime_control( - target, - state="resuming", - checkpoint_id=str( - getattr(checkpoint, "checkpoint_id", "") or "" - ).strip(), - ) - except Exception: - logger.opt(exception=True).debug("failed to broadcast company suspend reply routing state") - config_task = target.get("config_task") session_exec_mode = self._normalize_session_exec_mode(self._exec_mode) session_company_profile = self._normalize_session_company_profile(self._company_profile) @@ -8124,6 +8180,27 @@ class WSHandler: session_exec_mode, session_company_profile, ) + if engine_mode == "org" and not session_org_id: + raise ServiceError( + "org_id_required", + "org_id_required", + { + "project_id": pid, + "task_id": str(target.get("config_source_task_id", "") or "").strip(), + }, + ) + + try: + await self._set_company_runtime_control( + target, + state="resuming", + checkpoint_id=str( + getattr(checkpoint, "checkpoint_id", "") or "" + ).strip(), + ) + except Exception: + logger.opt(exception=True).debug("failed to broadcast company suspend reply routing state") + engine_message_metadata = dict(message_metadata or {}) engine_message_metadata.update({ "response_to_checkpoint_id": str(getattr(checkpoint, "checkpoint_id", "") or ""), @@ -8590,13 +8667,37 @@ class WSHandler: item for item in pending or [] if str(getattr(item, "checkpoint_id", "") or "").strip() == checkpoint_id - and str(getattr(item, "checkpoint_type", "") or "").strip() - in self._LOCK_FREE_CHECKPOINT_ANSWER_TYPES + and str(getattr(item, "checkpoint_type", "") or "").strip() == checkpoint_type ), None, ) if checkpoint is None: return False + checkpoint_payload = dict(getattr(checkpoint, "payload", {}) or {}) + checkpoint_task_id = str( + getattr(checkpoint, "task_id", "") + or checkpoint_payload.get("waiting_task_id") + or checkpoint_payload.get("task_id") + or "" + ).strip() + checkpoint_session_id = str( + getattr(checkpoint, "session_id", "") + or checkpoint_payload.get("session_id") + or "" + ).strip() + payload_task_ids = { + str(item).strip() + for item in list(checkpoint_payload.get("task_ids", []) or []) + if str(item).strip() + } + if ( + not checkpoint_task_id + or checkpoint_task_id != str(task_id or "").strip() + or not checkpoint_session_id + or checkpoint_session_id != str(session_id or "").strip() + or (payload_task_ids and str(task_id or "").strip() not in payload_task_ids) + ): + return False logger.info( f"Lock-free checkpoint answer: task lock for {task_id} is held by a " f"live turn; delivering reply to pending checkpoint {checkpoint_id} " @@ -8694,14 +8795,32 @@ class WSHandler: session_preferred_agent = self._task_preferred_agent session_org_id = "" task = None + config_task = None store = engine.store if self._store_is_ready(store): from opc.core.models import TaskStatus task = await store.get_task(task_id) if task: - session_exec_mode, session_company_profile = self._resolve_task_session_config(task) - session_org_id = self._resolve_task_org_id(task) - session_preferred_agent = self._resolve_task_preferred_agent(task) + config_task = await self._resolve_session_runtime_config_task( + task_id, + task, + engine=engine, + ) + session_exec_mode, session_company_profile = self._resolve_task_session_config( + config_task + ) + session_org_id = self._resolve_task_org_id(config_task) + session_preferred_agent = self._resolve_task_preferred_agent(config_task) + if ( + session_exec_mode in {"org", "custom"} + and not session_org_id + and is_company_runtime_task(task) + ): + raise ServiceError( + "org_id_required", + "org_id_required", + {"project_id": pid, "task_id": task_id}, + ) if await self._try_lock_free_parked_checkpoint_answer( task_id=task_id, @@ -8729,9 +8848,26 @@ class WSHandler: from opc.core.models import TaskStatus task = await store.get_task(task_id) if task: - session_exec_mode, session_company_profile = self._resolve_task_session_config(task) - session_org_id = self._resolve_task_org_id(task) - session_preferred_agent = self._resolve_task_preferred_agent(task) + config_task = await self._resolve_session_runtime_config_task( + task_id, + task, + engine=engine, + ) + session_exec_mode, session_company_profile = self._resolve_task_session_config( + config_task + ) + session_org_id = self._resolve_task_org_id(config_task) + session_preferred_agent = self._resolve_task_preferred_agent(config_task) + if ( + session_exec_mode in {"org", "custom"} + and not session_org_id + and is_company_runtime_task(task) + ): + raise ServiceError( + "org_id_required", + "org_id_required", + {"project_id": pid, "task_id": task_id}, + ) if task.status == TaskStatus.DONE and self._is_company_session_exec_mode(session_exec_mode): task.status = TaskStatus.IDLE task.metadata = dict(getattr(task, "metadata", {}) or {}) @@ -8757,12 +8893,17 @@ class WSHandler: except Exception: logger.opt(exception=True).debug("failed to mark company session runtime running") try: - engine_mode, company_profile = self._resolve_engine_mode( - session_exec_mode, - session_company_profile, + config_task_id = str(getattr(config_task, "id", "") or "").strip() + selected_task_id = str(getattr(task, "id", "") or "").strip() + should_persist_selected_config = bool( + task is not None + and ( + not config_task_id + or config_task_id == selected_task_id + or (session_exec_mode == "org" and not session_org_id) + ) ) - engine_preferred_agent = session_preferred_agent if session_exec_mode == "task" else None - if task is not None: + if should_persist_selected_config: await self._persist_session_config( task, exec_mode=session_exec_mode, @@ -8771,6 +8912,22 @@ class WSHandler: org_id=session_org_id, engine=engine, ) + persisted_identity = self._resolve_task_identity( + task, + default_exec_mode=session_exec_mode, + default_company_profile=session_company_profile, + default_preferred_agent=session_preferred_agent, + default_org_id=session_org_id, + ) + session_exec_mode = persisted_identity.exec_mode + session_company_profile = persisted_identity.company_profile + session_preferred_agent = persisted_identity.preferred_agent + session_org_id = persisted_identity.org_id + engine_mode, company_profile = self._resolve_engine_mode( + session_exec_mode, + session_company_profile, + ) + engine_preferred_agent = session_preferred_agent if session_exec_mode == "task" else None engine_message_metadata = dict(message_metadata or {}) engine_message_metadata.update(_ui_message_identity_metadata( message_id=user_message_id, diff --git a/tests/test_company_recruiter.py b/tests/test_company_recruiter.py index 6a4bf6a..28b8fb0 100644 --- a/tests/test_company_recruiter.py +++ b/tests/test_company_recruiter.py @@ -2072,6 +2072,7 @@ class CompanyRecruiterFlowTests(unittest.IsolatedAsyncioTestCase): decision = RouterDecision( mode=ExecutionMode.COMPANY_MODE, company_profile="custom", + org_id="test-org", domains=[], ) runtime_spec = engine.company_runtime_spec_builder.build_spec( @@ -2140,6 +2141,7 @@ class CompanyRecruiterFlowTests(unittest.IsolatedAsyncioTestCase): decision = RouterDecision( mode=ExecutionMode.COMPANY_MODE, company_profile="custom", + org_id="test-org", domains=[], ) runtime_spec = engine.company_runtime_spec_builder.build_spec( @@ -2199,6 +2201,7 @@ class CompanyRecruiterFlowTests(unittest.IsolatedAsyncioTestCase): decision = RouterDecision( mode=ExecutionMode.COMPANY_MODE, company_profile="custom", + org_id="test-org", domains=[], ) runtime_spec = engine.company_runtime_spec_builder.build_spec( @@ -2312,6 +2315,7 @@ class CompanyRecruiterFlowTests(unittest.IsolatedAsyncioTestCase): decision = RouterDecision( mode=ExecutionMode.COMPANY_MODE, company_profile="custom", + org_id="test-org", domains=[], ) runtime_spec = engine.company_runtime_spec_builder.build_spec( diff --git a/tests/test_company_runtime_identity.py b/tests/test_company_runtime_identity.py index abf82fe..e3d7a4b 100644 --- a/tests/test_company_runtime_identity.py +++ b/tests/test_company_runtime_identity.py @@ -260,6 +260,212 @@ def test_work_item_chat_resume_uses_canonical_ui_anchor_as_engine_origin() -> No asyncio.run(scenario()) +def test_company_suspend_reply_routes_selected_org_to_engine() -> None: + async def scenario() -> None: + _tasks, checkpoint = _runtime_records() + handler = WSHandler.__new__(WSHandler) + handler._exec_mode = "task" + handler._company_profile = "corporate" + handler._shutting_down = False + handler._active_runtime_children = {} + handler._session_to_task = {} + handler._task_bg_context = {} + handler._company_suspend_reply_locks = {"runtime-session": asyncio.Lock()} + handler.chat_store = None + handler._set_company_runtime_control = AsyncMock() + handler._normalize_session_exec_mode = MagicMock(return_value="task") + handler._normalize_session_company_profile = MagicMock(return_value="corporate") + handler._resolve_task_session_config = MagicMock(return_value=("org", "custom")) + handler._resolve_task_org_id = MagicMock(return_value="selected-org") + handler._extract_checkpoint_metadata = AsyncMock(return_value=None) + handler._sync_task_transcript_messages = AsyncMock() + handler.on_kanban_changed = AsyncMock() + handler._flush_progress = AsyncMock() + run_engine = SimpleNamespace( + project_id="project-a", + process_message=AsyncMock(return_value="resumed"), + ) + target = { + "ui_anchor_task_id": "ui-anchor", + "config_task": SimpleNamespace( + metadata={"exec_mode": "org", "company_profile": "custom"}, + org_id="selected-org", + ), + } + + await handler._process_company_suspend_reply( + ui_task_id="final-decider", + runtime_session_id="runtime-session", + content="continue", + attachment_refs=None, + message_metadata={"ui_force_resume": True}, + user_message_id=None, + user_message_created_at=None, + run_engine=run_engine, + run_project_id="project-a", + target=target, + checkpoint=checkpoint, + lock=handler._company_suspend_reply_locks["runtime-session"], + ) + + call = run_engine.process_message.await_args + assert call.kwargs["org_id"] == "selected-org" + + asyncio.run(scenario()) + + +def test_delivery_feedback_reply_fails_closed_without_durable_parent_task() -> None: + async def scenario() -> None: + handler = WSHandler.__new__(WSHandler) + handler._exec_mode = "task" + handler._company_profile = "corporate" + handler._shutting_down = False + handler._active_runtime_children = {} + handler._session_to_task = {} + handler._task_bg_context = {} + handler._company_delivery_feedback_reply_locks = {} + handler.chat_store = None + handler._store_is_ready = MagicMock(return_value=True) + handler._normalize_session_exec_mode = MagicMock(return_value="task") + handler._normalize_session_company_profile = MagicMock(return_value="corporate") + handler._resolve_task_session_config = MagicMock(return_value=("org", "custom")) + handler._resolve_task_org_id = MagicMock(return_value="") + handler._chat_store_is_ready = MagicMock(return_value=False) + handler._flush_progress = AsyncMock() + run_engine = SimpleNamespace( + project_id="project-a", + store=SimpleNamespace(get_task=AsyncMock(return_value=None)), + run_company_delivery_self_evolution_checkpoint=AsyncMock(return_value="ran"), + ) + + await handler._process_company_delivery_feedback_reply( + parent_task_id="delivery-task", + parent_session_id="delivery-session", + reply_channel_id="session:delivery-task", + content="approved", + attachment_refs=None, + message_metadata=None, + user_message_id=None, + user_message_created_at=None, + run_engine=run_engine, + run_project_id="project-a", + checkpoint=SimpleNamespace(checkpoint_id="cp-delivery", payload={}), + waiting_task_id="delivery-task", + lock=asyncio.Lock(), + ) + + run_engine.run_company_delivery_self_evolution_checkpoint.assert_not_awaited() + handler._resolve_task_org_id.assert_not_called() + + asyncio.run(scenario()) + + +def test_delivery_feedback_reply_proceeds_with_durable_parent_task() -> None: + async def scenario() -> None: + handler = WSHandler.__new__(WSHandler) + handler._exec_mode = "task" + handler._company_profile = "corporate" + handler._shutting_down = False + handler._active_runtime_children = {} + handler._session_to_task = {} + handler._task_bg_context = {} + handler._company_delivery_feedback_reply_locks = {} + handler.chat_store = None + handler._store_is_ready = MagicMock(return_value=True) + handler._normalize_session_exec_mode = MagicMock(return_value="task") + handler._normalize_session_company_profile = MagicMock(return_value="corporate") + handler._resolve_task_session_config = MagicMock(return_value=("org", "custom")) + handler._resolve_task_org_id = MagicMock(return_value="selected-org") + handler._chat_store_is_ready = MagicMock(return_value=False) + handler._flush_progress = AsyncMock() + parent_task = SimpleNamespace(id="delivery-task") + run_engine = SimpleNamespace( + project_id="project-a", + store=SimpleNamespace(get_task=AsyncMock(return_value=parent_task)), + run_company_delivery_self_evolution_checkpoint=AsyncMock(return_value="ran"), + ) + + await handler._process_company_delivery_feedback_reply( + parent_task_id="delivery-task", + parent_session_id="delivery-session", + reply_channel_id="session:delivery-task", + content="approved", + attachment_refs=None, + message_metadata=None, + user_message_id=None, + user_message_created_at=None, + run_engine=run_engine, + run_project_id="project-a", + checkpoint=SimpleNamespace(checkpoint_id="cp-delivery", payload={}), + waiting_task_id="delivery-task", + lock=asyncio.Lock(), + ) + + call = run_engine.run_company_delivery_self_evolution_checkpoint.await_args + assert call is not None + assert call.kwargs["action"] == "approve" + handler._resolve_task_org_id.assert_called_once_with(parent_task) + + asyncio.run(scenario()) + + +def test_suspend_reply_missing_custom_org_id_fails_closed_before_engine_call() -> None: + async def scenario() -> None: + _tasks, checkpoint = _runtime_records() + handler = WSHandler.__new__(WSHandler) + handler._exec_mode = "task" + handler._company_profile = "corporate" + handler._shutting_down = False + handler.chat_store = None + handler._company_suspend_reply_locks = { + "runtime-session": asyncio.Lock(), + } + handler._task_bg_context = {} + handler._active_runtime_children = {} + handler._session_to_task = {} + handler._set_company_runtime_control = AsyncMock() + handler._normalize_session_exec_mode = MagicMock(return_value="task") + handler._normalize_session_company_profile = MagicMock(return_value="corporate") + handler._resolve_task_session_config = MagicMock(return_value=("org", "custom")) + handler._resolve_task_org_id = MagicMock(return_value="") + handler._extract_checkpoint_metadata = AsyncMock(return_value=None) + handler._sync_task_transcript_messages = AsyncMock() + handler.on_kanban_changed = AsyncMock() + handler._flush_progress = AsyncMock() + run_engine = SimpleNamespace( + project_id="project-a", + process_message=AsyncMock(), + ) + config_task = SimpleNamespace( + metadata={"exec_mode": "org", "company_profile": "custom"}, + org_id=None, + ) + target = { + "ui_anchor_task_id": "ui-anchor", + "config_task": config_task, + } + + await handler._process_company_suspend_reply( + ui_task_id="final-decider", + runtime_session_id="runtime-session", + content="continue", + attachment_refs=None, + message_metadata=None, + user_message_id=None, + user_message_created_at=None, + run_engine=run_engine, + run_project_id="project-a", + target=target, + checkpoint=checkpoint, + lock=handler._company_suspend_reply_locks["runtime-session"], + ) + + run_engine.process_message.assert_not_awaited() + handler._set_company_runtime_control.assert_not_awaited() + + asyncio.run(scenario()) + + def test_delivery_feedback_rejects_missing_canonical_identity_without_first_task_fallback() -> None: async def scenario() -> None: tasks, checkpoint = _runtime_records() diff --git a/tests/test_lock_free_checkpoint_answer.py b/tests/test_lock_free_checkpoint_answer.py index 09f0f35..82663b9 100644 --- a/tests/test_lock_free_checkpoint_answer.py +++ b/tests/test_lock_free_checkpoint_answer.py @@ -49,11 +49,20 @@ class _EngineStub: return self.reply -def _pending_checkpoint(checkpoint_id: str, checkpoint_type: str = "task_user_input") -> Any: +def _pending_checkpoint( + checkpoint_id: str, + checkpoint_type: str = "task_user_input", + *, + task_id: str = "chat-task", + session_id: str = "session-1", +) -> Any: return SimpleNamespace( checkpoint_id=checkpoint_id, checkpoint_type=checkpoint_type, status="pending", + task_id=task_id, + session_id=session_id, + payload={"task_ids": [task_id], "waiting_task_id": task_id, "session_id": session_id}, ) @@ -189,6 +198,30 @@ class LockFreeCheckpointAnswerTests(unittest.IsolatedAsyncioTestCase): holder.release_event.set() # type: ignore[attr-defined] await holder + async def test_lock_free_requires_exact_checkpoint_type_and_owner(self) -> None: + engine = _EngineStub( + _StoreStub([ + _pending_checkpoint( + "ckpt-park", + "company_work_item_gate", + task_id="other-task", + session_id="other-session", + ) + ]) + ) + handler = _make_handler(engine) + holder = await self._hold_lock(handler, "chat-task") + try: + handled = await handler._try_lock_free_parked_checkpoint_answer( + engine=engine, + **_answer_kwargs(), + ) + self.assertFalse(handled) + self.assertEqual(engine.calls, []) + finally: + holder.release_event.set() # type: ignore[attr-defined] + await holder + async def test_engine_failure_surfaces_error_without_queueing(self) -> None: engine = _EngineStub( _StoreStub([_pending_checkpoint("ckpt-park")]), diff --git a/tests/test_office_execution_identity.py b/tests/test_office_execution_identity.py index 5b1e4fa..e98bd89 100644 --- a/tests/test_office_execution_identity.py +++ b/tests/test_office_execution_identity.py @@ -112,6 +112,21 @@ def test_task_org_id_field_is_org_identity_fallback() -> None: assert identity.org_id == "quantum_harbor" +def test_durable_task_org_id_wins_over_stale_metadata_org_id() -> None: + task = SimpleNamespace( + metadata={ + "exec_mode": "org", + "company_profile": "custom", + "organization_id": "active-org", + }, + org_id="selected-org", + ) + + identity = execution_identity_from_task(task) + + assert identity.org_id == "selected-org" + + def test_default_org_id_applies_only_when_task_has_no_persisted_identity() -> None: task = SimpleNamespace(metadata={}, org_id=None) @@ -146,3 +161,42 @@ def test_explicit_company_identity_ignores_default_org_id() -> None: assert identity.exec_mode == "company" assert identity.company_profile == "corporate" assert identity.org_id == "" + + +def test_company_mode_marker_without_profile_is_company_identity() -> None: + task = SimpleNamespace( + metadata={"mode": "company"}, + org_id=None, + ) + + identity = execution_identity_from_task(task) + + assert identity.exec_mode == "company" + assert identity.company_profile == "corporate" + assert identity.org_id == "" + + +def test_work_item_runtime_marker_without_profile_is_company_identity() -> None: + task = SimpleNamespace( + metadata={"work_item_runtime": True}, + org_id=None, + ) + + identity = execution_identity_from_task(task) + + assert identity.exec_mode == "company" + assert identity.company_profile == "corporate" + assert identity.org_id == "" + + +def test_company_runtime_marker_with_custom_profile_is_org_identity() -> None: + task = SimpleNamespace( + metadata={"mode": "company", "work_item_runtime": True, "company_profile": "custom"}, + org_id=None, + ) + + identity = execution_identity_from_task(task) + + assert identity.exec_mode == "org" + assert identity.company_profile == "custom" + assert identity.org_id == "" diff --git a/tests/test_office_session_org_fallback.py b/tests/test_office_session_org_fallback.py new file mode 100644 index 0000000..1f4fbd1 --- /dev/null +++ b/tests/test_office_session_org_fallback.py @@ -0,0 +1,161 @@ +from __future__ import annotations + +import unittest +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock + +from opc.plugins.office_ui.services.context import OfficeServiceContext +from opc.plugins.office_ui.services.models import ServiceError +from opc.plugins.office_ui.services.session import SessionService + + +class _Store: + def __init__(self) -> None: + self.saved: list[Any] = [] + + async def save_task(self, task: Any) -> None: + self.saved.append(task) + + +def _task(**overrides: Any) -> SimpleNamespace: + task = SimpleNamespace( + id="task-role-1", + session_id="sess-1", + project_id="demo", + title="Role task", + parent_session_id=None, + metadata={}, + org_id=None, + ) + for key, value in overrides.items(): + setattr(task, key, value) + return task + + +def _context(*, hook: Any | None = None) -> OfficeServiceContext: + engine = SimpleNamespace(project_id="demo", store=_Store(), memory=None) + context = OfficeServiceContext(engine=engine, agent_store=None, chat_store=None, event_adapter=None) + if hook is not None: + context.get_active_saved_org_name = hook + return context + + +class TestPersistSessionConfigOrgIdentity(unittest.IsolatedAsyncioTestCase): + async def _persist( + self, + context: OfficeServiceContext, + task: Any, + *, + exec_mode: str = "org", + company_profile: str = "custom", + org_id: str = "", + ) -> None: + await SessionService(context).persist_session_config( + task, + exec_mode=exec_mode, + company_profile=company_profile, + preferred_agent="native", + org_id=org_id, + ) + + async def test_rejects_missing_org_id_without_active_org_fallback(self) -> None: + active_org = AsyncMock(return_value="vc-investment-firm") + context = _context(hook=active_org) + task = _task() + + with self.assertRaises(ServiceError) as ctx: + await self._persist(context, task) + + assert ctx.exception.code == "org_id_required" + active_org.assert_not_awaited() + assert context.engine.store.saved == [] + + async def test_explicit_org_id_still_used_when_present(self) -> None: + active_org = AsyncMock(return_value="other-org") + context = _context(hook=active_org) + task = _task() + + await self._persist(context, task, org_id="vc-investment-firm") + + assert task.metadata["org_id"] == "vc-investment-firm" + assert task.org_id == "vc-investment-firm" + active_org.assert_not_awaited() + + async def test_raises_when_no_active_org_available(self) -> None: + empty_org = AsyncMock(return_value="") + context = _context(hook=empty_org) + task = _task() + + with self.assertRaises(ServiceError) as ctx: + await self._persist(context, task) + assert ctx.exception.code == "org_id_required" + empty_org.assert_not_awaited() + + async def test_raises_when_hook_unset(self) -> None: + context = _context() + task = _task() + + with self.assertRaises(ServiceError) as ctx: + await self._persist(context, task) + assert ctx.exception.code == "org_id_required" + + async def test_raises_when_hook_fails(self) -> None: + broken = AsyncMock(side_effect=RuntimeError("org index unreadable")) + context = _context(hook=broken) + task = _task() + + with self.assertRaises(ServiceError) as ctx: + await self._persist(context, task) + assert ctx.exception.code == "org_id_required" + broken.assert_not_awaited() + + async def test_company_mode_clears_org_fields_without_fallback(self) -> None: + fallback_called = False + + async def active_org() -> str: + nonlocal fallback_called + fallback_called = True + return "vc-investment-firm" + + context = _context(hook=active_org) + task = _task(metadata={"org_id": "stale-org"}) + + await self._persist( + context, + task, + exec_mode="company", + company_profile="corporate", + ) + + assert not fallback_called + assert "org_id" not in task.metadata + assert "organization_id" not in task.metadata + assert task.org_id is None + + async def test_task_mode_ignores_org_entirely(self) -> None: + fallback_called = False + + async def active_org() -> str: + nonlocal fallback_called + fallback_called = True + return "vc-investment-firm" + + context = _context(hook=active_org) + task = _task() + + await self._persist( + context, + task, + exec_mode="task", + company_profile="corporate", + ) + + assert not fallback_called + assert task.metadata["execution_mode"] == "task_mode" + assert "org_id" not in task.metadata + assert task.org_id is None + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_org_mode_entrypoint.py b/tests/test_org_mode_entrypoint.py index f372e32..509a228 100644 --- a/tests/test_org_mode_entrypoint.py +++ b/tests/test_org_mode_entrypoint.py @@ -13,6 +13,7 @@ from opc.core.org_config import ( ) from opc.engine import OPCEngine from opc.layer2_organization.custom_runtime import CustomRuntimeRunner +from opc.plugins.office_ui.services.models import ServiceError def test_requested_mode_normalization_keeps_core_company_router_main_compatible() -> None: @@ -61,6 +62,34 @@ def test_custom_runtime_runner_loads_org_storage_without_mutating_parent_config( assert (config_dir / "company_orgs" / "org_lab_config.yaml").exists() +def test_process_message_rejects_org_mode_without_org_id() -> None: + engine = OPCEngine.__new__(OPCEngine) + engine.opc_home = None + engine.config = OPCConfig() + runner = CustomRuntimeRunner(engine) + caught: ServiceError | None = None + + async def _run() -> None: + await runner.process_message( + "run org", + project_id="default", + session_id="session-1", + org_id=None, + preferred_agent=None, + domains=None, + origin_task_id=None, + attachment_refs=None, + message_metadata=None, + ) + + try: + asyncio.run(_run()) + except ServiceError as exc: + caught = exc + assert caught is not None + assert caught.code == "org_id_required" + + def test_process_message_routes_org_mode_to_custom_runner(monkeypatch) -> None: engine = OPCEngine.__new__(OPCEngine) engine._initialized = True diff --git a/tests/test_runtime_config_enforcement.py b/tests/test_runtime_config_enforcement.py index db0c969..55789ac 100644 --- a/tests/test_runtime_config_enforcement.py +++ b/tests/test_runtime_config_enforcement.py @@ -11,9 +11,15 @@ from unittest.mock import AsyncMock, MagicMock, patch import yaml from pydantic import ValidationError -from opc.core.config import AgentsConfig, ExternalAgentConfig, OPCConfig +from opc.core.config import ( + AgentsConfig, + DEFAULT_ORGANIZATION_ID, + ExternalAgentConfig, + OPCConfig, +) from opc.core.models import ( DelegationWorkItem, + ExecutionCheckpoint, ExecutionMode, RouterDecision, SessionMessageRecord, @@ -24,7 +30,8 @@ from opc.core.models import ( WorkItemExecutionStrategy, ) from opc.engine import OPCEngine -from opc.layer2_organization.company_mode import CompanyWorkItemExecutor +from opc.layer2_organization.company_mode import CompanyRuntimeSpecBuilder, CompanyWorkItemExecutor +from opc.plugins.office_ui.services.models import ServiceError from opc.layer3_agent.adapters.claude_code import ClaudeCodeAdapter from opc.layer3_agent.adapters.codex_adapter import CodexAdapter from opc.layer3_agent.adapters.cursor_adapter import CursorAdapter @@ -323,6 +330,650 @@ class RuntimeConfigEnforcementTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(task.metadata["preferred_external_agent"], "opencode") self.assertEqual(task.metadata["work_item_execution_strategy"], WorkItemExecutionStrategy.EXTERNAL.value) + async def test_company_root_task_persists_decision_org_over_active_config(self) -> None: + engine = OPCEngine(config=OPCConfig(), project_id="proj1") + engine.config.org.company_profile = "custom" + engine.config.org.organization_id = "active-org" + engine.store = SimpleNamespace( + get_runtime_task_for_work_item=AsyncMock(return_value=None), + save_delegation_work_item=AsyncMock(), + save_task=AsyncMock(), + link_work_item_runtime_task=AsyncMock(return_value=True), + ) + engine.memory = SimpleNamespace(ensure_session=AsyncMock()) + engine.org_engine = SimpleNamespace( + current_org_version=MagicMock(return_value=1), + current_runtime_topology_version=MagicMock(return_value=1), + ) + engine._requests_explicit_project_knowledge = MagicMock(return_value=False) + work_item = DelegationWorkItem( + work_item_id="wi-selected-org", + run_id="run-selected-org", + cell_id="team::engineering", + team_instance_id="team-instance-1", + role_id="engineer", + seat_id="seat-engineer", + title="Engineering execution", + summary="Implement the requested change.", + kind="execute", + projection_id="engineering-execute", + metadata={"seat_id": "seat-engineer", "team_id": "team::engineering"}, + ) + + task = await engine._ensure_runtime_work_item_task( + work_item=work_item, + parent_session_id="sess-company", + original_message="Build the thing.", + decision=RouterDecision( + mode=ExecutionMode.COMPANY_MODE, + domains=[], + company_profile="custom", + org_id="selected-org", + ), + runtime_topology={ + "final_decider_role_id": "lead", + "seats": [ + { + "seat_id": "seat-engineer", + "team_id": "team::engineering", + "role_id": "engineer", + "employee_assignment": {"employee_id": "eng-1", "name": "Engineer"}, + "metadata": {"role_name": "Engineer"}, + } + ], + }, + delegation_playbook={}, + secretary_context="", + target_output_dir=None, + origin_channel="cli", + origin_chat_id="", + origin_thread_id="", + origin_task_id=None, + attachment_refs=[], + attachment_context="", + force_native_execution=False, + ) + + self.assertEqual(task.org_id, "selected-org") + self.assertEqual(task.metadata["org_id"], "selected-org") + self.assertEqual(task.metadata["organization_id"], "selected-org") + + def test_build_spec_rejects_custom_run_without_durable_org_id(self) -> None: + builder = CompanyRuntimeSpecBuilder( + org_engine=SimpleNamespace( + get_company_profile=lambda: "custom", + config=SimpleNamespace( + org=SimpleNamespace( + organization_id="active-org", + organization_name="Active Org", + organization_config_file="org_active-org_config.yaml", + company_profile="custom", + ) + ), + ) + ) + with self.assertRaises(ServiceError) as ctx: + builder.build_spec( + RouterDecision( + mode=ExecutionMode.COMPANY_MODE, + company_profile="custom", + org_id=None, + ), + original_message="Run the company.", + ) + assert ctx.exception.code == "org_id_required" + + def test_build_spec_serializes_selected_org_not_active_config(self) -> None: + builder = CompanyRuntimeSpecBuilder( + org_engine=SimpleNamespace( + get_company_profile=lambda: "custom", + config=SimpleNamespace( + org=SimpleNamespace( + organization_id="active-org", + organization_name="Active Org", + organization_config_file="org_active-org_config.yaml", + company_profile="custom", + ) + ), + ) + ) + decision = RouterDecision( + mode=ExecutionMode.COMPANY_MODE, + company_profile="custom", + org_id="selected-org", + ) + spec = builder.build_spec(decision, original_message="Run the company.") + assert spec.metadata["org_id"] == "selected-org" + assert spec.metadata["organization_id"] == "selected-org" + + def test_runtime_org_id_for_identity_never_derives_from_active_config(self) -> None: + engine = OPCEngine(config=OPCConfig(), project_id="proj1") + engine.config.org.company_profile = "custom" + engine.config.org.organization_id = "active-org" + decision = RouterDecision( + mode=ExecutionMode.COMPANY_MODE, + company_profile="custom", + org_id=None, + ) + resolved = OPCEngine._runtime_org_id_for_identity( + decision, + {}, + engine.config.org, + ) + assert resolved is None + + async def test_ensure_runtime_work_item_task_rejects_custom_without_durable_org(self) -> None: + engine = OPCEngine(config=OPCConfig(), project_id="proj1") + engine.config.org.company_profile = "custom" + engine.config.org.organization_id = "active-org" + engine.store = SimpleNamespace( + get_runtime_task_for_work_item=AsyncMock(return_value=None), + save_delegation_work_item=AsyncMock(), + save_task=AsyncMock(), + link_work_item_runtime_task=AsyncMock(return_value=True), + ) + engine.memory = SimpleNamespace(ensure_session=AsyncMock()) + engine.org_engine = SimpleNamespace( + current_org_version=MagicMock(return_value=1), + current_runtime_topology_version=MagicMock(return_value=1), + ) + engine._requests_explicit_project_knowledge = MagicMock(return_value=False) + work_item = DelegationWorkItem( + work_item_id="wi-no-durable-org", + run_id="run-no-durable-org", + cell_id="team::engineering", + team_instance_id="team-instance-1", + role_id="engineer", + seat_id="seat-engineer", + title="Engineering execution", + summary="Implement the requested change.", + kind="execute", + projection_id="engineering-execute", + metadata={"seat_id": "seat-engineer", "team_id": "team::engineering"}, + ) + with self.assertRaises(ServiceError) as ctx: + await engine._ensure_runtime_work_item_task( + work_item=work_item, + parent_session_id="sess-company", + original_message="Build the thing.", + decision=RouterDecision( + mode=ExecutionMode.COMPANY_MODE, + company_profile="custom", + org_id=None, + ), + runtime_topology={ + "final_decider_role_id": "lead", + "seats": [ + { + "seat_id": "seat-engineer", + "team_id": "team::engineering", + "role_id": "engineer", + "employee_assignment": {"employee_id": "eng-1", "name": "Engineer"}, + "metadata": {"role_name": "Engineer"}, + } + ], + }, + delegation_playbook={}, + secretary_context="", + target_output_dir=None, + origin_channel="cli", + origin_chat_id="", + origin_thread_id="", + origin_task_id=None, + attachment_refs=[], + attachment_context="", + force_native_execution=False, + ) + assert ctx.exception.code == "org_id_required" + engine.store.save_task.assert_not_awaited() + + async def test_ensure_runtime_work_item_task_rejects_existing_custom_task_without_durable_org(self) -> None: + existing = Task( + id="existing-no-org", + title="Engineering execution", + project_id="proj1", + session_id="sess-company:wi-existing-no-org", + assigned_to="engineer", + metadata={ + "execution_mode": "company_mode", + "runtime_model": "multi_team_org", + "work_item_runtime": True, + "work_item_projection_id": "engineering-execute", + "work_item_turn_type": "execute", + "company_profile": "custom", + "delegation_seat_id": "seat-engineer", + }, + ) + engine = OPCEngine(config=OPCConfig(), project_id="proj1") + engine.config.org.company_profile = "custom" + engine.config.org.organization_id = "active-org" + engine.store = SimpleNamespace( + get_runtime_task_for_work_item=AsyncMock(return_value=existing), + save_delegation_work_item=AsyncMock(), + save_task=AsyncMock(), + link_work_item_runtime_task=AsyncMock(return_value=True), + ) + engine.memory = SimpleNamespace(ensure_session=AsyncMock()) + work_item = DelegationWorkItem( + work_item_id="wi-existing-no-org", + run_id="run-existing-no-org", + cell_id="team::engineering", + team_instance_id="team-instance-1", + role_id="engineer", + seat_id="seat-engineer", + title="Engineering execution", + summary="Implement the requested change.", + kind="execute", + projection_id="engineering-execute", + metadata={"seat_id": "seat-engineer", "team_id": "team::engineering"}, + ) + with self.assertRaises(ServiceError) as ctx: + await engine._ensure_runtime_work_item_task( + work_item=work_item, + parent_session_id="sess-company", + original_message="Build the thing.", + decision=RouterDecision( + mode=ExecutionMode.COMPANY_MODE, + company_profile="custom", + org_id=None, + ), + runtime_topology={ + "final_decider_role_id": "lead", + "seats": [ + { + "seat_id": "seat-engineer", + "team_id": "team::engineering", + "role_id": "engineer", + "employee_assignment": {"employee_id": "eng-1", "name": "Engineer"}, + "metadata": {"role_name": "Engineer"}, + } + ], + }, + delegation_playbook={}, + secretary_context="", + target_output_dir=None, + origin_channel="cli", + origin_chat_id="", + origin_thread_id="", + origin_task_id=None, + attachment_refs=[], + attachment_context="", + force_native_execution=False, + ) + assert ctx.exception.code == "org_id_required" + engine.store.save_task.assert_not_awaited() + + async def test_ensure_runtime_work_item_task_rejects_conflicting_org_ids(self) -> None: + existing = Task( + id="existing-conflict-org", + title="Engineering execution", + project_id="proj1", + session_id="sess-company:wi-existing-conflict-org", + assigned_to="engineer", + org_id="persisted-org", + metadata={ + "execution_mode": "company_mode", + "runtime_model": "multi_team_org", + "work_item_runtime": True, + "work_item_projection_id": "engineering-execute", + "work_item_turn_type": "execute", + "company_profile": "custom", + "organization_id": "persisted-org", + "delegation_seat_id": "seat-engineer", + }, + ) + engine = OPCEngine(config=OPCConfig(), project_id="proj1") + engine.config.org.company_profile = "custom" + engine.config.org.organization_id = "active-org" + engine.store = SimpleNamespace( + get_runtime_task_for_work_item=AsyncMock(return_value=existing), + save_delegation_work_item=AsyncMock(), + save_task=AsyncMock(), + link_work_item_runtime_task=AsyncMock(return_value=True), + ) + engine.memory = SimpleNamespace(ensure_session=AsyncMock()) + work_item = DelegationWorkItem( + work_item_id="wi-existing-conflict-org", + run_id="run-existing-conflict-org", + cell_id="team::engineering", + team_instance_id="team-instance-1", + role_id="engineer", + seat_id="seat-engineer", + title="Engineering execution", + summary="Implement the requested change.", + kind="execute", + projection_id="engineering-execute", + metadata={"seat_id": "seat-engineer", "team_id": "team::engineering"}, + ) + with self.assertRaises(ServiceError) as ctx: + await engine._ensure_runtime_work_item_task( + work_item=work_item, + parent_session_id="sess-company", + original_message="Build the thing.", + decision=RouterDecision( + mode=ExecutionMode.COMPANY_MODE, + company_profile="custom", + org_id="selected-org", + ), + runtime_topology={ + "final_decider_role_id": "lead", + "seats": [ + { + "seat_id": "seat-engineer", + "team_id": "team::engineering", + "role_id": "engineer", + "employee_assignment": {"employee_id": "eng-1", "name": "Engineer"}, + "metadata": {"role_name": "Engineer"}, + } + ], + }, + delegation_playbook={}, + secretary_context="", + target_output_dir=None, + origin_channel="cli", + origin_chat_id="", + origin_thread_id="", + origin_task_id=None, + attachment_refs=[], + attachment_context="", + force_native_execution=False, + ) + assert ctx.exception.code == "org_id_conflict" + engine.store.save_task.assert_not_awaited() + + async def test_ensure_runtime_work_item_task_repairs_existing_custom_task_from_persisted_durable_org(self) -> None: + existing = Task( + id="existing-durable-org", + title="Engineering execution", + project_id="proj1", + session_id="sess-company:wi-existing-durable-org", + assigned_to="engineer", + org_id="persisted-org", + metadata={ + "execution_mode": "company_mode", + "runtime_model": "multi_team_org", + "work_item_runtime": True, + "work_item_projection_id": "engineering-execute", + "work_item_turn_type": "execute", + "company_profile": "custom", + "organization_id": "stale-org", + "delegation_seat_id": "seat-engineer", + }, + ) + engine = OPCEngine(config=OPCConfig(), project_id="proj1") + engine.config.org.company_profile = "custom" + engine.config.org.organization_id = "active-org" + engine.store = SimpleNamespace( + get_runtime_task_for_work_item=AsyncMock(return_value=existing), + save_delegation_work_item=AsyncMock(), + save_task=AsyncMock(), + link_work_item_runtime_task=AsyncMock(return_value=True), + ) + engine.memory = SimpleNamespace(ensure_session=AsyncMock()) + work_item = DelegationWorkItem( + work_item_id="wi-existing-durable-org", + run_id="run-existing-durable-org", + cell_id="team::engineering", + team_instance_id="team-instance-1", + role_id="engineer", + seat_id="seat-engineer", + title="Engineering execution", + summary="Implement the requested change.", + kind="execute", + projection_id="engineering-execute", + metadata={"seat_id": "seat-engineer", "team_id": "team::engineering"}, + ) + repaired = await engine._ensure_runtime_work_item_task( + work_item=work_item, + parent_session_id="sess-company", + original_message="Build the thing.", + decision=RouterDecision( + mode=ExecutionMode.COMPANY_MODE, + company_profile="custom", + org_id=None, + ), + runtime_topology={ + "final_decider_role_id": "lead", + "seats": [ + { + "seat_id": "seat-engineer", + "team_id": "team::engineering", + "role_id": "engineer", + "employee_assignment": {"employee_id": "eng-1", "name": "Engineer"}, + "metadata": {"role_name": "Engineer"}, + } + ], + }, + delegation_playbook={}, + secretary_context="", + target_output_dir=None, + origin_channel="cli", + origin_chat_id="", + origin_thread_id="", + origin_task_id=None, + attachment_refs=[], + attachment_context="", + force_native_execution=False, + ) + self.assertEqual(repaired.id, "existing-durable-org") + self.assertEqual(repaired.org_id, "persisted-org") + self.assertEqual(repaired.metadata["org_id"], "persisted-org") + self.assertEqual(repaired.metadata["organization_id"], "persisted-org") + engine.store.save_task.assert_awaited_with(existing) + + async def test_delivery_self_evolution_custom_without_durable_org_fails_closed(self) -> None: + engine = OPCEngine(config=OPCConfig(), project_id="proj1") + engine.config.org.company_profile = "custom" + engine.config.org.organization_id = "active-org" + waiting_task = Task( + id="waiting-custom", + project_id="proj1", + session_id="sess-delivery", + metadata={ + "execution_mode": "company_mode", + "company_profile": "custom", + "work_item_runtime": True, + "work_item_projection_id": "delivery", + "work_item_turn_type": "deliver", + }, + ) + engine.store = SimpleNamespace(get_task=AsyncMock(return_value=waiting_task)) + engine._mark_company_runtime_checkpoint_status = AsyncMock() + engine._create_company_self_evolution_root_work_item = AsyncMock() + checkpoint = ExecutionCheckpoint( + checkpoint_id="cp-delivery", + project_id="proj1", + session_id="sess-delivery", + task_id="waiting-custom", + checkpoint_type="company_delivery_feedback", + payload={"waiting_task_id": "waiting-custom", "task_ids": ["waiting-custom"]}, + ) + result = await engine._run_company_delivery_self_evolution_consumed( + checkpoint, + action="approve", + ) + assert "durable org identity" in result + engine._mark_company_runtime_checkpoint_status.assert_awaited_once_with( + checkpoint, + status="invalid", + ) + engine._create_company_self_evolution_root_work_item.assert_not_awaited() + + async def test_delivery_self_evolution_custom_uses_durable_org_over_payload_and_config(self) -> None: + engine = OPCEngine(config=OPCConfig(), project_id="proj1") + engine.config.org.company_profile = "custom" + engine.config.org.organization_id = "active-org" + waiting_task = Task( + id="waiting-custom-2", + project_id="proj1", + session_id="sess-delivery", + org_id="selected-org", + metadata={ + "execution_mode": "company_mode", + "company_profile": "custom", + "work_item_runtime": True, + "work_item_projection_id": "delivery", + "work_item_turn_type": "deliver", + }, + ) + engine.store = SimpleNamespace(get_task=AsyncMock(return_value=waiting_task)) + engine._mark_company_runtime_checkpoint_status = AsyncMock() + engine.org_engine = None + engine._company_followup_target_task = MagicMock( + return_value=SimpleNamespace(assigned_to="lead", metadata={}), + ) + engine.company_executor = SimpleNamespace() + engine._self_evolution_assignments_by_role = MagicMock(return_value={}) + engine._create_company_self_evolution_root_work_item = AsyncMock(return_value=None) + checkpoint = ExecutionCheckpoint( + checkpoint_id="cp-delivery-2", + project_id="proj1", + session_id="sess-delivery", + task_id="waiting-custom-2", + checkpoint_type="company_delivery_feedback", + payload={ + "waiting_task_id": "waiting-custom-2", + "task_ids": ["waiting-custom-2"], + "organization_id": "stale-org", + }, + ) + await engine._run_company_delivery_self_evolution_consumed( + checkpoint, + action="approve", + ) + call = engine._create_company_self_evolution_root_work_item.await_args + assert call is not None + assert call.kwargs["organization_id"] == "selected-org" + + async def test_delivery_self_evolution_corporate_still_uses_config_default(self) -> None: + engine = OPCEngine(config=OPCConfig(), project_id="proj1") + waiting_task = Task( + id="waiting-corporate", + project_id="proj1", + session_id="sess-delivery", + metadata={ + "execution_mode": "company_mode", + "company_profile": "corporate", + "work_item_runtime": True, + "work_item_projection_id": "delivery", + "work_item_turn_type": "deliver", + }, + ) + engine.store = SimpleNamespace(get_task=AsyncMock(return_value=waiting_task)) + engine._mark_company_runtime_checkpoint_status = AsyncMock() + engine.org_engine = None + engine._company_followup_target_task = MagicMock( + return_value=SimpleNamespace(assigned_to="lead", metadata={}), + ) + engine.company_executor = SimpleNamespace() + engine._self_evolution_assignments_by_role = MagicMock(return_value={}) + engine._create_company_self_evolution_root_work_item = AsyncMock(return_value=None) + checkpoint = ExecutionCheckpoint( + checkpoint_id="cp-delivery-corp", + project_id="proj1", + session_id="sess-delivery", + task_id="waiting-corporate", + checkpoint_type="company_delivery_feedback", + payload={"waiting_task_id": "waiting-corporate", "task_ids": ["waiting-corporate"]}, + ) + await engine._run_company_delivery_self_evolution_consumed( + checkpoint, + action="approve", + ) + call = engine._create_company_self_evolution_root_work_item.await_args + assert call is not None + assert call.kwargs["organization_id"] == DEFAULT_ORGANIZATION_ID + + async def test_delivery_self_evolution_metadata_only_legacy_custom_uses_metadata_org(self) -> None: + engine = OPCEngine(config=OPCConfig(), project_id="proj1") + engine.config.org.company_profile = "custom" + engine.config.org.organization_id = "active-org" + waiting_task = Task( + id="waiting-legacy-metadata", + project_id="proj1", + session_id="sess-delivery", + metadata={ + "work_item_runtime": True, + "work_item_projection_id": "delivery", + "work_item_turn_type": "deliver", + "org_id": "selected-org", + }, + ) + engine.store = SimpleNamespace(get_task=AsyncMock(return_value=waiting_task)) + engine._mark_company_runtime_checkpoint_status = AsyncMock() + engine.org_engine = None + engine._company_followup_target_task = MagicMock( + return_value=SimpleNamespace(assigned_to="lead", metadata={}), + ) + engine.company_executor = SimpleNamespace() + engine._self_evolution_assignments_by_role = MagicMock(return_value={}) + engine._create_company_self_evolution_root_work_item = AsyncMock(return_value=None) + checkpoint = ExecutionCheckpoint( + checkpoint_id="cp-delivery-legacy", + project_id="proj1", + session_id="sess-delivery", + task_id="waiting-legacy-metadata", + checkpoint_type="company_delivery_feedback", + payload={ + "waiting_task_id": "waiting-legacy-metadata", + "task_ids": ["waiting-legacy-metadata"], + }, + ) + await engine._run_company_delivery_self_evolution_consumed( + checkpoint, + action="approve", + ) + call = engine._create_company_self_evolution_root_work_item.await_args + assert call is not None + assert call.kwargs["organization_id"] == "selected-org" + + async def test_delivery_self_evolution_conflicting_org_ids_fail_closed(self) -> None: + engine = OPCEngine(config=OPCConfig(), project_id="proj1") + engine.config.org.company_profile = "custom" + engine.config.org.organization_id = "active-org" + waiting_task = Task( + id="waiting-conflict", + project_id="proj1", + session_id="sess-delivery", + org_id="selected-org", + metadata={ + "work_item_runtime": True, + "work_item_projection_id": "delivery", + "work_item_turn_type": "deliver", + "org_id": "other-org", + }, + ) + engine.store = SimpleNamespace(get_task=AsyncMock(return_value=waiting_task)) + engine._mark_company_runtime_checkpoint_status = AsyncMock() + engine.org_engine = None + engine._company_followup_target_task = MagicMock( + return_value=SimpleNamespace(assigned_to="lead", metadata={}), + ) + engine.company_executor = SimpleNamespace() + engine._self_evolution_assignments_by_role = MagicMock(return_value={}) + engine._create_company_self_evolution_root_work_item = AsyncMock(return_value=None) + checkpoint = ExecutionCheckpoint( + checkpoint_id="cp-delivery-conflict", + project_id="proj1", + session_id="sess-delivery", + task_id="waiting-conflict", + checkpoint_type="company_delivery_feedback", + payload={ + "waiting_task_id": "waiting-conflict", + "task_ids": ["waiting-conflict"], + }, + ) + result = await engine._run_company_delivery_self_evolution_consumed( + checkpoint, + action="approve", + ) + assert "org identity" in result + engine._mark_company_runtime_checkpoint_status.assert_awaited_once_with( + checkpoint, + status="invalid", + ) + engine._create_company_self_evolution_root_work_item.assert_not_awaited() + async def test_company_materialized_work_item_uses_selected_agent_over_template_preference(self) -> None: saved_tasks: list[Task] = [] @@ -389,6 +1040,132 @@ class RuntimeConfigEnforcementTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(created.metadata["work_item_execution_strategy"], WorkItemExecutionStrategy.EXTERNAL.value) self.assertEqual(saved_tasks[0].assigned_external_agent, "cursor") + async def test_company_materialization_persists_durable_org_on_new_role_task(self) -> None: + executor = CompanyWorkItemExecutor( + org_engine=SimpleNamespace(), + communication=SimpleNamespace(), + approval_engine=SimpleNamespace(), + memory=SimpleNamespace(ensure_session=AsyncMock()), + execute_task=AsyncMock(), + save_task=AsyncMock(), + ) + executor.store = SimpleNamespace( + get_runtime_task_for_work_item=AsyncMock(return_value=None), + save_delegation_work_item=AsyncMock(), + save_task=AsyncMock(), + link_work_item_runtime_task=AsyncMock(return_value=True), + ) + root_task = Task( + id="root-custom-org", + title="Root", + project_id="proj1", + session_id="sess-company", + org_id="selected-org", + metadata={ + "execution_mode": "company_mode", + "runtime_model": "multi_team_org", + "company_profile": "custom", + "organization_id": "active-org", + "runtime_topology": { + "seats": [ + { + "seat_id": "seat-engineer", + "team_id": "team::engineering", + "role_id": "engineer", + "metadata": {"role_name": "Engineer"}, + } + ] + }, + }, + ) + work_item = DelegationWorkItem( + work_item_id="wi-new-custom-org", + run_id="run-custom-org", + cell_id="team::engineering", + team_instance_id="team-instance-1", + role_id="engineer", + seat_id="seat-engineer", + title="Engineering execution", + summary="Implement the requested change.", + kind="execute", + projection_id="engineering-execute", + metadata={"seat_id": "seat-engineer", "team_id": "team::engineering"}, + ) + + tasks = await executor._materialize_work_item_tasks([root_task], [work_item]) + created = next(task for task in tasks if task.id != root_task.id) + + self.assertEqual(created.org_id, "selected-org") + self.assertEqual(created.metadata["org_id"], "selected-org") + self.assertEqual(created.metadata["organization_id"], "selected-org") + + async def test_company_materialization_repairs_existing_role_task_org_identity(self) -> None: + existing = Task( + id="existing-custom-role", + title="Engineering execution", + project_id="proj1", + session_id="sess-company:wi-existing-custom-org", + assigned_to="engineer", + org_id="active-org", + metadata={ + "execution_mode": "company_mode", + "runtime_model": "multi_team_org", + "work_item_runtime": True, + "work_item_projection_id": "engineering-execute", + "work_item_turn_type": "execute", + "company_profile": "custom", + "organization_id": "active-org", + "delegation_seat_id": "seat-engineer", + }, + ) + executor = CompanyWorkItemExecutor( + org_engine=SimpleNamespace(), + communication=SimpleNamespace(), + approval_engine=SimpleNamespace(), + memory=SimpleNamespace(ensure_session=AsyncMock()), + execute_task=AsyncMock(), + save_task=AsyncMock(), + ) + executor.store = SimpleNamespace( + get_runtime_task_for_work_item=AsyncMock(return_value=existing), + save_delegation_work_item=AsyncMock(), + save_task=AsyncMock(), + ) + root_task = Task( + id="root-custom-org-existing", + title="Root", + project_id="proj1", + session_id="sess-company", + org_id="selected-org", + metadata={ + "execution_mode": "company_mode", + "runtime_model": "multi_team_org", + "company_profile": "custom", + "organization_id": "active-org", + }, + ) + work_item = DelegationWorkItem( + work_item_id="wi-existing-custom-org", + run_id="run-custom-org", + cell_id="team::engineering", + team_instance_id="team-instance-1", + role_id="engineer", + seat_id="seat-engineer", + title="Engineering execution", + summary="Implement the requested change.", + kind="execute", + projection_id="engineering-execute", + metadata={"seat_id": "seat-engineer", "team_id": "team::engineering"}, + ) + + tasks = await executor._materialize_work_item_tasks([root_task], [work_item]) + repaired = next(task for task in tasks if task.id == existing.id) + + self.assertEqual(repaired.org_id, "selected-org") + self.assertEqual(repaired.metadata["org_id"], "selected-org") + self.assertEqual(repaired.metadata["organization_id"], "selected-org") + executor.store.save_task.assert_any_await(existing) + async def test_task_mode_external_followup_reuses_primary_session_external_agent_session(self) -> None: engine = OPCEngine(config=OPCConfig(), project_id="proj1") engine.store = SimpleNamespace( diff --git a/tests/test_session_integration.py b/tests/test_session_integration.py index 4ace294..4d2e3b0 100644 --- a/tests/test_session_integration.py +++ b/tests/test_session_integration.py @@ -1998,6 +1998,192 @@ class TestWSHandlerSessionSend(unittest.IsolatedAsyncioTestCase): self.assertIn("company_session_reopened_at", refreshed.metadata) self.engine.process_message.assert_called_once() + async def test_process_session_message_uses_durable_org_for_role_task(self) -> None: + runtime_session_id = "runtime-org-session" + anchor = await self.store.get_task(self.task_id) + assert anchor is not None + anchor.session_id = runtime_session_id + anchor.metadata = { + "exec_mode": "org", + "mode": "company", + "company_profile": "custom", + "org_id": "wrong-active-org", + "organization_id": "wrong-active-org", + } + anchor.org_id = "vc-investment-firm" + await self.store.save_task(anchor) + + role_task = Task( + id="role-task-without-org-id", + title="Sector Analyst", + project_id="test-project", + session_id=f"{runtime_session_id}:role:sector-analyst", + parent_session_id=runtime_session_id, + org_id="vc-investment-firm", + metadata={ + "exec_mode": "org", + "mode": "company", + "company_profile": "custom", + "org_id": "wrong-active-org", + "organization_id": "wrong-active-org", + "shared_role_session": True, + "shared_role_id": "sector_analyst", + "company_runtime_root_session_id": runtime_session_id, + }, + ) + await self.store.save_task(role_task) + self.handler.services_context.get_active_saved_org_name = AsyncMock( + return_value="wrong-active-org" + ) + + await self.handler._process_session_message( + role_task.id, + "approve", + session_id=role_task.session_id, + ) + + self.engine.process_message.assert_called_once() + call_kwargs = self.engine.process_message.call_args.kwargs + self.assertEqual(call_kwargs["mode"], "org") + self.assertEqual(call_kwargs["company_profile"], "custom") + self.assertEqual(call_kwargs["org_id"], "vc-investment-firm") + + async def test_process_session_message_fails_closed_when_runtime_identity_is_missing(self) -> None: + task = await self.store.get_task(self.task_id) + assert task is not None + task.metadata = { + "mode": "company", + "company_profile": "custom", + } + await self.store.save_task(task) + self.handler._resolve_company_runtime_target = AsyncMock(return_value=None) + self.handler.services_context.get_active_saved_org_name = AsyncMock( + return_value="wrong-active-org" + ) + + with self.assertRaises(ServiceError) as context: + await self.handler._process_session_message( + self.task_id, + "approve", + session_id=self.session_id, + ) + + self.engine.process_message.assert_not_called() + self.handler.services_context.get_active_saved_org_name.assert_not_awaited() + self.assertEqual(context.exception.code, "company_runtime_identity_mismatch") + + async def test_process_session_message_rejects_runtime_org_without_durable_org_id(self) -> None: + runtime_session_id = "runtime-org-missing-id-session" + anchor = await self.store.get_task(self.task_id) + assert anchor is not None + anchor.session_id = runtime_session_id + anchor.metadata = { + "exec_mode": "org", + "mode": "company", + "company_profile": "custom", + } + await self.store.save_task(anchor) + + role_task = Task( + id="role-task-missing-org-id", + title="Sector Analyst", + project_id="test-project", + session_id=f"{runtime_session_id}:role:sector-analyst", + parent_session_id=runtime_session_id, + metadata={ + "exec_mode": "org", + "mode": "company", + "company_profile": "custom", + "shared_role_session": True, + "shared_role_id": "sector_analyst", + "company_runtime_root_session_id": runtime_session_id, + }, + ) + await self.store.save_task(role_task) + self.handler.services_context.get_active_saved_org_name = AsyncMock( + return_value="wrong-active-org" + ) + + with self.assertRaises(ServiceError) as context: + await self.handler._process_session_message( + role_task.id, + "approve", + session_id=role_task.session_id, + ) + + self.engine.process_message.assert_not_called() + self.handler.services_context.get_active_saved_org_name.assert_not_awaited() + self.assertEqual(context.exception.code, "org_id_required") + + async def test_lock_free_process_session_message_uses_durable_org_for_role_task(self) -> None: + runtime_session_id = "runtime-org-lock-free-session" + anchor = await self.store.get_task(self.task_id) + assert anchor is not None + anchor.session_id = runtime_session_id + anchor.metadata = { + "exec_mode": "org", + "mode": "company", + "company_profile": "custom", + "org_id": "vc-investment-firm", + } + await self.store.save_task(anchor) + + role_task = Task( + id="role-task-lock-free-without-org-id", + title="Sector Analyst", + project_id="test-project", + session_id=f"{runtime_session_id}:role:sector-analyst", + parent_session_id=runtime_session_id, + metadata={ + "exec_mode": "org", + "mode": "company", + "company_profile": "custom", + "shared_role_session": True, + "shared_role_id": "sector_analyst", + "company_runtime_root_session_id": runtime_session_id, + }, + ) + await self.store.save_task(role_task) + checkpoint = ExecutionCheckpoint( + checkpoint_id="org-lock-free-gate", + project_id="test-project", + session_id=role_task.session_id, + checkpoint_type="company_work_item_gate", + status="pending", + task_id=role_task.id, + payload={ + "waiting_task_id": role_task.id, + "task_ids": [role_task.id], + "session_id": role_task.session_id, + }, + ) + await self.store.save_execution_checkpoint(checkpoint) + self.engine._load_execution_checkpoint_by_id = AsyncMock(return_value=checkpoint) + self.handler.services_context.get_active_saved_org_name = AsyncMock( + return_value="wrong-active-org" + ) + + lock = self.handler._get_task_lock(role_task.id) + await lock.acquire() + try: + await self.handler._process_session_message( + role_task.id, + "approve", + session_id=role_task.session_id, + message_metadata={ + "response_to_checkpoint_id": checkpoint.checkpoint_id, + "response_to_checkpoint_type": checkpoint.checkpoint_type, + }, + ) + finally: + lock.release() + + self.engine.process_message.assert_called_once() + call_kwargs = self.engine.process_message.call_args.kwargs + self.assertEqual(call_kwargs["mode"], "org") + self.assertEqual(call_kwargs["company_profile"], "custom") + self.assertEqual(call_kwargs["org_id"], "vc-investment-firm") + async def test_session_send_reuses_task_session_without_is_ready_flag(self) -> None: """Session replies should reuse the task session even for simple stub stores.""" ws = MagicMock() @@ -5296,6 +5482,41 @@ class TestOfficeServiceExecutionIdentity(unittest.IsolatedAsyncioTestCase): assert persisted is not None self.assertEqual(persisted.status, TaskStatus.RUNNING) + async def test_continue_rejects_runtime_custom_org_without_durable_org_id(self) -> None: + task = Task( + id="service-continue-org-missing-id", + title="Broken custom runtime", + session_id="service-continue-org-runtime", + project_id="test-project", + metadata={"exec_mode": "org", "company_profile": "custom"}, + ) + checkpoint = ExecutionCheckpoint( + checkpoint_id="service-continue-org-checkpoint", + project_id="test-project", + session_id=task.session_id, + checkpoint_type="company_runtime_suspended", + status="pending", + task_id=task.id, + ) + await self.store.save_task(task) + self.session_service._resolve_company_runtime_target = AsyncMock(return_value={ + "runtime_session_id": task.session_id, + "ui_anchor_task_id": task.id, + "config_task": task, + "checkpoint": checkpoint, + "affected_task_ids": [task.id], + }) + + with self.assertRaises(ServiceError) as raised: + await self.session_service.continue_run( + project_id="test-project", + task_id=task.id, + content="continue", + ) + + self.assertEqual(raised.exception.code, "org_id_required") + self.engine.process_message.assert_not_called() + async def test_session_send_from_work_item_uses_runtime_checkpoint_identity(self) -> None: anchor = Task( id="service-send-anchor", @@ -5726,6 +5947,32 @@ class TestWSHandlerRunTask(unittest.IsolatedAsyncioTestCase): self.assertEqual(call_kwargs["company_profile"], "custom") self.assertEqual(call_kwargs["org_id"], "quantum_harbor") + async def test_run_task_fails_closed_for_custom_identity_without_org_id(self) -> None: + task_id = str(uuid.uuid4()) + session_id = str(uuid.uuid4()) + task = Task( + id=task_id, + title="Missing Org Runtime", + session_id=session_id, + project_id="test-project", + metadata={"exec_mode": "org", "company_profile": "custom"}, + ) + await self.store.save_task(task) + self.handler.services_context.get_active_saved_org_name = AsyncMock( + return_value="wrong-active-org" + ) + + await self.handler._run_task( + "Missing Org Runtime", + "Description", + "org", + "custom", + task_id, + ) + + self.engine.process_message.assert_not_called() + self.handler.services_context.get_active_saved_org_name.assert_not_awaited() + async def test_run_task_explicit_company_clears_stale_custom_fields(self) -> None: task_id = str(uuid.uuid4()) session_id = str(uuid.uuid4())