Merge pull request #27 from cgycorey/fix/org-id-required-fallback

fix(ui): durable custom-org identity for role tasks, replies, approvals, and followups

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", "")

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)

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(

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"

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({

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,