Merge pull request #29 from CatJuly/fix/ci-install-dev-deps

PR #27 scoped the lock-free parked-checkpoint answer to exact checkpoint
task/session equality. Company gate cards are raised by role work-item
tasks but answered from the run's anchor chat, whose task id only
appears in payload["task_ids"] — the exact-match guard silently
disabled the fast path for precisely the answers it exists for and
re-opened the project-0012 late-approval lock wedge. Scope by the same
linkage set _find_parked_checkpoint_for_deferred_resume uses (checkpoint
task/session plus payload waiting_task_id/task_ids), keep rejecting
unrelated channels, and keep legacy checkpoints without linkage
deliverable.

The new fail-closed identity errors in _process_session_message raised
out of fire-and-forget background tasks (_track_session), where they are
only logged and the user's message silently vanishes. Surface them as a
visible system chat error and stop instead of raising.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

.github/workflows/external-agent-smoke.yml

@@ -22,7 +22,7 @@ jobs:
         with:
           python-version: "3.11"
       - name: Install package
-        run: python -m pip install -e .
+        run: python -m pip install -e ".[dev]"
       - name: Run external-agent smoke checks
         run: >
           python -m pytest -q

config/llm_config.yaml

@@ -7,6 +7,8 @@ llm:
 
   temperature: 1
   max_tokens: 32768
+  # reasoning_effort: "high"  # optional; use a provider-supported level
+                              # such as low, medium, or high
   routing: {}
   fallback: {}
 

opc/core/config.py

@@ -273,6 +273,7 @@ class LLMConfig(BaseModel):
     fallback: dict[str, Any] = Field(default_factory=dict)
     temperature: float = 0.3
     max_tokens: int = 32768
+    reasoning_effort: str | None = None
     # Total input context window (tokens) for the active model. Set this when
     # the model is not mapped in litellm (e.g. proxy/self-hosted models like
     # doubao/minimax/glm), so the context-usage ring and compaction thresholds

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/llm/provider.py

@@ -569,6 +569,8 @@ class LLMProvider:
             "max_tokens": max_tok,
             **kwargs,
         }
+        if self.config.reasoning_effort and "reasoning_effort" not in call_kwargs:
+            call_kwargs["reasoning_effort"] = self.config.reasoning_effort
         if self._api_base:
             call_kwargs["api_base"] = self._api_base
         if self._api_key:
@@ -715,6 +717,8 @@ class LLMProvider:
             "stream": True,
             **kwargs,
         }
+        if self.config.reasoning_effort and "reasoning_effort" not in call_kwargs:
+            call_kwargs["reasoning_effort"] = self.config.reasoning_effort
         if self._api_base:
             call_kwargs["api_base"] = self._api_base
         if self._api_key:

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/frontend_dist/index.html

@@ -5,7 +5,7 @@
     <meta name="viewport" content="width=device-width, initial-scale=1.0" />
     <link rel="icon" href="data:," />
     <title>OpenOPC Pixel Office</title>
-    <script type="module" crossorigin src="./assets/index-Dw3OK9za.js"></script>
+    <script type="module" crossorigin src="./assets/index-Cdi_JV6w.js"></script>
     <link rel="modulepreload" crossorigin href="./assets/phaser-DFK5Ua9d.js">
     <link rel="stylesheet" crossorigin href="./assets/index-B6-ikSHW.css">
   </head>

opc/plugins/office_ui/frontend_src/components/ProjectSelector.tsx

@@ -87,13 +87,14 @@ export function ProjectSelector({ projects, activeId, onSelect, onCreate, onDele
           background: 'rgba(0,0,0,0.5)',
         }}>
           <div style={{
-            background: 'var(--bg-surface, #1e1e2e)', borderRadius: 12, padding: '24px 32px',
+            background: 'var(--bg-elevated)', borderRadius: 12, padding: '24px 32px',
+            border: '1px solid var(--border)', color: 'var(--text)',
             maxWidth: 400, boxShadow: '0 8px 32px rgba(0,0,0,0.4)', textAlign: 'center',
           }}>
-            <p style={{ margin: '0 0 8px', fontWeight: 600, fontSize: 15 }}>
+            <p style={{ margin: '0 0 8px', fontWeight: 600, fontSize: 15, color: 'var(--text)' }}>
               {t('project.confirmTitle', { name: confirmDelete })}
             </p>
-            <p style={{ margin: '0 0 20px', fontSize: 13, opacity: 0.7 }}>
+            <p style={{ margin: '0 0 20px', fontSize: 13, color: 'var(--text-secondary)' }}>
               {t('project.confirmBody')}
             </p>
             <div style={{ display: 'flex', gap: 8, justifyContent: 'center' }}>

opc/plugins/office_ui/services/session.py

@@ -618,25 +618,6 @@ class SessionService:
             default_preferred_agent=self.context.mode_state.task_preferred_agent,
             explicit_exec_mode=True,
         )
-        if identity.is_custom_org and not identity.org_id:
-            # Role-task rows may lack org_id; mirror create()'s active-org fallback
-            fallback_org_id = ""
-            if self.context.get_active_saved_org_name is not None:
-                try:
-                    fallback_org_id = await self.context.get_active_saved_org_name()
-                except Exception:
-                    logger.opt(exception=True).debug(
-                        "persist_session_config: failed to resolve active saved org"
-                    )
-            if fallback_org_id:
-                identity = canonicalize_execution_identity(
-                    exec_mode=exec_mode,
-                    company_profile=company_profile,
-                    preferred_agent=preferred_agent,
-                    org_id=fallback_org_id,
-                    default_preferred_agent=self.context.mode_state.task_preferred_agent,
-                    explicit_exec_mode=True,
-                )
         if identity.is_custom_org and not identity.org_id:
             raise ServiceError("org_id_required", "org_id_required", {
                 "task_id": str(getattr(task, "id", "") or ""),
@@ -1182,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

@@ -4978,16 +4978,38 @@ class WSHandler:
         if task is None:
             return None
         exec_mode, _ = self._resolve_task_session_config(task)
-        if not self._is_company_session_exec_mode(exec_mode):
+        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 Exception:
+        except ServiceError:
+            if runtime_bound:
+                raise
             logger.opt(exception=True).debug(
                 "failed to resolve durable session config task"
             )
             return task
-        return (target or {}).get("config_task") or 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:
@@ -5059,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,
@@ -5076,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
@@ -7819,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 = ""
@@ -8124,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)
@@ -8146,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 ""),
@@ -8612,13 +8667,43 @@ 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
+        # Ownership scoping must accept every channel the card legitimately
+        # reaches, not just the task that raised the checkpoint: company gate
+        # cards raised by role work items are answered from the run's anchor
+        # chat, whose task id only appears in payload["task_ids"] (the same
+        # linkage set _find_parked_checkpoint_for_deferred_resume uses). An
+        # exact task/session equality check would silently disable this fast
+        # path for those answers and re-open the late-approval lock wedge.
+        checkpoint_payload = dict(getattr(checkpoint, "payload", {}) or {})
+        requester_task_id = str(task_id or "").strip()
+        requester_session_id = str(session_id or "").strip()
+        linked_task_ids = {
+            str(checkpoint_payload.get("task_id") or "").strip(),
+            str(checkpoint_payload.get("waiting_task_id") or "").strip(),
+            str(getattr(checkpoint, "task_id", "") or "").strip(),
+        }
+        linked_task_ids.update(
+            str(item or "").strip()
+            for item in list(checkpoint_payload.get("task_ids", []) or [])
+        )
+        linked_task_ids.discard("")
+        linked_session_ids = {
+            str(getattr(checkpoint, "session_id", "") or "").strip(),
+            str(checkpoint_payload.get("session_id") or "").strip(),
+        }
+        linked_session_ids.discard("")
+        has_linkage = bool(linked_task_ids or linked_session_ids)
+        if has_linkage and (
+            requester_task_id not in linked_task_ids
+            and (not requester_session_id or requester_session_id not in linked_session_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} "
@@ -8722,6 +8807,11 @@ class WSHandler:
             from opc.core.models import TaskStatus
             task = await store.get_task(task_id)
             if task:
+                # This coroutine usually runs as a fire-and-forget background
+                # task (_track_session), where a raised ServiceError is only
+                # logged and the user's message silently vanishes. Fail closed
+                # with a visible chat error instead of raising.
+                try:
                     config_task = await self._resolve_session_runtime_config_task(
                         task_id,
                         task,
@@ -8732,6 +8822,35 @@ class WSHandler:
                     )
                     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},
+                        )
+                except ServiceError as exc:
+                    logger.warning(
+                        f"Session message for task {task_id} rejected during "
+                        f"runtime identity resolution: {exc.code}"
+                    )
+                    try:
+                        msg = await self.chat_store.insert_message(
+                            channel_id=channel_id,
+                            sender="system",
+                            sender_name="OPC",
+                            content=f"Error: {exc.message}",
+                            project_id=pid,
+                        )
+                        await self.broadcast({"type": "session_message", "payload": msg})
+                    except Exception:
+                        logger.opt(exception=True).debug(
+                            "failed to surface session identity error"
+                        )
+                    return
 
         if await self._try_lock_free_parked_checkpoint_answer(
             task_id=task_id,
@@ -8769,6 +8888,16 @@ class WSHandler:
                     )
                     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 {})

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(

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

tests/test_llm_provider_reasoning_effort.py

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+import unittest
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+from opc.core.config import LLMConfig
+from opc.llm.provider import LLMProvider
+
+
+def _completion_response() -> SimpleNamespace:
+    return SimpleNamespace(
+        choices=[
+            SimpleNamespace(
+                message=SimpleNamespace(content="ok", tool_calls=[]),
+                finish_reason="stop",
+            )
+        ],
+        usage=SimpleNamespace(prompt_tokens=1, completion_tokens=1),
+    )
+
+
+async def _completion_stream():
+    yield SimpleNamespace(
+        usage=None,
+        choices=[
+            SimpleNamespace(
+                delta=SimpleNamespace(
+                    content="ok",
+                    reasoning=None,
+                    reasoning_content=None,
+                    thinking=None,
+                    tool_calls=[],
+                ),
+                finish_reason="stop",
+            )
+        ],
+    )
+
+
+class TestLLMProviderReasoningEffort(unittest.IsolatedAsyncioTestCase):
+    def test_config_retains_reasoning_effort(self) -> None:
+        config = LLMConfig.model_validate({
+            "default_model": "openai/gpt-5.6-luna",
+            "reasoning_effort": "max",
+        })
+
+        assert config.reasoning_effort == "max"
+
+    async def test_chat_forwards_configured_reasoning_effort(self) -> None:
+        provider = LLMProvider(LLMConfig(
+            default_model="openai/gpt-5.6-luna",
+            reasoning_effort="max",
+        ))
+
+        with (
+            patch("opc.llm.provider._clamp_max_tokens", return_value=128),
+            patch(
+                "opc.llm.provider.litellm.acompletion",
+                new=AsyncMock(return_value=_completion_response()),
+            ) as completion,
+        ):
+            await provider.chat([{"role": "user", "content": "hello"}])
+
+        assert completion.await_args.kwargs["reasoning_effort"] == "max"
+
+    async def test_chat_allows_per_call_reasoning_effort_override(self) -> None:
+        provider = LLMProvider(LLMConfig(
+            default_model="openai/gpt-5.6-luna",
+            reasoning_effort="high",
+        ))
+
+        with (
+            patch("opc.llm.provider._clamp_max_tokens", return_value=128),
+            patch(
+                "opc.llm.provider.litellm.acompletion",
+                new=AsyncMock(return_value=_completion_response()),
+            ) as completion,
+        ):
+            await provider.chat(
+                [{"role": "user", "content": "hello"}],
+                reasoning_effort="low",
+            )
+
+        assert completion.await_args.kwargs["reasoning_effort"] == "low"
+
+    async def test_chat_stream_forwards_configured_reasoning_effort(self) -> None:
+        provider = LLMProvider(LLMConfig(
+            default_model="openai/gpt-5.6-luna",
+            reasoning_effort="max",
+        ))
+
+        with (
+            patch("opc.llm.provider._clamp_max_tokens", return_value=128),
+            patch(
+                "opc.llm.provider.litellm.acompletion",
+                new=AsyncMock(return_value=_completion_stream()),
+            ) as completion,
+        ):
+            events = [
+                event
+                async for event in provider.chat_stream([
+                    {"role": "user", "content": "hello"},
+                ])
+            ]
+
+        assert events
+        assert completion.await_args.kwargs["reasoning_effort"] == "max"
+        assert completion.await_args.kwargs["stream"] is True
+
+    async def test_chat_stream_allows_per_call_reasoning_effort_override(self) -> None:
+        provider = LLMProvider(LLMConfig(
+            default_model="openai/gpt-5.6-luna",
+            reasoning_effort="high",
+        ))
+
+        with (
+            patch("opc.llm.provider._clamp_max_tokens", return_value=128),
+            patch(
+                "opc.llm.provider.litellm.acompletion",
+                new=AsyncMock(return_value=_completion_stream()),
+            ) as completion,
+        ):
+            events = [
+                event
+                async for event in provider.chat_stream(
+                    [{"role": "user", "content": "hello"}],
+                    reasoning_effort="low",
+                )
+            ]
+
+        assert events
+        assert completion.await_args.kwargs["reasoning_effort"] == "low"
+
+    async def test_unset_reasoning_effort_is_not_added_to_requests(self) -> None:
+        provider = LLMProvider(LLMConfig(default_model="openai/gpt-5.6-luna"))
+
+        with (
+            patch("opc.llm.provider._clamp_max_tokens", return_value=128),
+            patch(
+                "opc.llm.provider.litellm.acompletion",
+                new=AsyncMock(return_value=_completion_response()),
+            ) as completion,
+        ):
+            await provider.chat([{"role": "user", "content": "hello"}])
+
+        assert "reasoning_effort" not in completion.await_args.kwargs

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")]),
@@ -230,6 +263,104 @@ class LockFreeCheckpointAnswerTests(unittest.IsolatedAsyncioTestCase):
         self.assertFalse(handled)
         self.assertEqual(engine.calls, [])
 
+    async def test_anchor_channel_answer_for_role_task_gate_is_handled(self) -> None:
+        # Project-0012 production shape: a company gate checkpoint is raised
+        # by a role work-item task, but the card is answered from the run's
+        # anchor chat channel. The anchor task id only appears in
+        # payload["task_ids"]; exact task/session equality would reject it
+        # and re-open the late-approval lock wedge.
+        checkpoint = SimpleNamespace(
+            checkpoint_id="ckpt-park",
+            checkpoint_type="company_work_item_gate",
+            status="pending",
+            task_id="role-task",
+            session_id="role-session",
+            payload={
+                "waiting_task_id": "role-task",
+                "session_id": "role-session",
+                "task_ids": ["role-task", "chat-task"],
+            },
+        )
+        engine = _EngineStub(_StoreStub([checkpoint]))
+        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(
+                    message_metadata={
+                        "response_to_checkpoint_id": "ckpt-park",
+                        "response_to_checkpoint_type": "company_work_item_gate",
+                    },
+                ),
+            )
+            self.assertTrue(handled)
+            self.assertEqual(len(engine.calls), 1)
+        finally:
+            holder.release_event.set()  # type: ignore[attr-defined]
+            await holder
+
+    async def test_legacy_checkpoint_without_linkage_is_still_handled(self) -> None:
+        # Checkpoints persisted before ownership fields existed carry no
+        # task/session linkage at all. They must keep the pre-scoping
+        # behavior (deliver by explicit checkpoint id) instead of silently
+        # falling back to the wedged serialized path.
+        checkpoint = SimpleNamespace(
+            checkpoint_id="ckpt-park",
+            checkpoint_type="task_user_input",
+            status="pending",
+            payload={},
+        )
+        engine = _EngineStub(_StoreStub([checkpoint]))
+        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.assertTrue(handled)
+            self.assertEqual(len(engine.calls), 1)
+        finally:
+            holder.release_event.set()  # type: ignore[attr-defined]
+            await holder
+
+
+class SessionIdentityErrorSurfacingTests(unittest.IsolatedAsyncioTestCase):
+    async def test_identity_service_error_surfaces_in_chat_instead_of_raising(self) -> None:
+        # _process_session_message mostly runs as a fire-and-forget background
+        # task; a ServiceError escaping it is only logged and the user's
+        # message silently vanishes. The pre-lock identity resolution must
+        # surface the failure as a visible chat error and stop.
+        from opc.plugins.office_ui.services.models import ServiceError
+
+        class _Store:
+            async def get_task(self, task_id: str) -> Any:
+                return SimpleNamespace(id=task_id, session_id="sess-1", metadata={})
+
+        engine = _EngineStub(_Store())
+        handler = _make_handler(engine)
+        handler.engine = engine
+        handler._session_to_task = {}
+        handler._exec_mode = "company"
+        handler._company_profile = "corporate"
+        handler._task_preferred_agent = "native"
+
+        async def _raise_identity_error(*args: Any, **kwargs: Any) -> Any:
+            raise ServiceError(
+                "company_runtime_identity_mismatch",
+                "Company runtime identity could not be resolved",
+                {"task_id": "chat-task"},
+            )
+
+        handler._resolve_session_runtime_config_task = _raise_identity_error
+
+        await handler._process_session_message("chat-task", "please continue")
+
+        self.assertEqual(engine.calls, [])
+        errors = [m for m in handler.chat_store.inserted if m.get("sender") == "system"]
+        self.assertEqual(len(errors), 1)
+        self.assertIn("Company runtime identity could not be resolved", errors[0]["content"])
+
 
 if __name__ == "__main__":
     unittest.main()

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 == ""

tests/test_office_session_org_fallback.py

@@ -3,6 +3,7 @@ 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
@@ -40,7 +41,7 @@ def _context(*, hook: Any | None = None) -> OfficeServiceContext:
     return context
 
 
-class TestPersistSessionConfigOrgFallback(unittest.IsolatedAsyncioTestCase):
+class TestPersistSessionConfigOrgIdentity(unittest.IsolatedAsyncioTestCase):
     async def _persist(
         self,
         context: OfficeServiceContext,
@@ -58,26 +59,20 @@ class TestPersistSessionConfigOrgFallback(unittest.IsolatedAsyncioTestCase):
             org_id=org_id,
         )
 
-    async def test_falls_back_to_active_saved_org_when_task_lacks_org_id(self) -> None:
-        async def active_org() -> str:
-            return "vc-investment-firm"
-
+    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 task.metadata["org_id"] == "vc-investment-firm"
-        assert task.metadata["organization_id"] == "vc-investment-firm"
-        assert task.org_id == "vc-investment-firm"
-        assert task.metadata["exec_mode"] == "org"
-        assert task.metadata["company_profile"] == "custom"
-        assert context.engine.store.saved == [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:
-        async def active_org() -> str:
-            return "other-org"
-
+        active_org = AsyncMock(return_value="other-org")
         context = _context(hook=active_org)
         task = _task()
 
@@ -85,17 +80,17 @@ class TestPersistSessionConfigOrgFallback(unittest.IsolatedAsyncioTestCase):
 
         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:
-        async def empty_org() -> str:
-            return ""
-
+        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()
@@ -106,15 +101,14 @@ class TestPersistSessionConfigOrgFallback(unittest.IsolatedAsyncioTestCase):
         assert ctx.exception.code == "org_id_required"
 
     async def test_raises_when_hook_fails(self) -> None:
-        async def broken() -> str:
-            raise RuntimeError("org index unreadable")
-
+        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

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

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(

tests/test_session_integration.py

@@ -2007,8 +2007,10 @@ class TestWSHandlerSessionSend(unittest.IsolatedAsyncioTestCase):
             "exec_mode": "org",
             "mode": "company",
             "company_profile": "custom",
-            "org_id": "vc-investment-firm",
+            "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(
@@ -2017,10 +2019,13 @@ class TestWSHandlerSessionSend(unittest.IsolatedAsyncioTestCase):
             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,
@@ -2043,6 +2048,96 @@ class TestWSHandlerSessionSend(unittest.IsolatedAsyncioTestCase):
         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"
+        )
+
+        # Fail closed without raising: this coroutine usually runs as a
+        # fire-and-forget background task where an escaping ServiceError is
+        # only logged and the user's message silently vanishes. The rejection
+        # must instead surface as a visible chat error.
+        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()
+        errors = [
+            msg["payload"].get("content", "")
+            for msg in self.broadcasts
+            if msg.get("type") == "session_message"
+            and str(msg.get("payload", {}).get("sender", "")) == "system"
+        ]
+        self.assertTrue(
+            any("Company runtime identity could not be resolved" in text for text in errors),
+            errors,
+        )
+
+    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"
+        )
+
+        # Same fail-closed-without-raising contract as the identity-mismatch
+        # case above: reject visibly instead of raising out of a background
+        # task.
+        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()
+        errors = [
+            msg["payload"].get("content", "")
+            for msg in self.broadcasts
+            if msg.get("type") == "session_message"
+            and str(msg.get("payload", {}).get("sender", "")) == "system"
+        ]
+        self.assertTrue(
+            any("org_id_required" in text for text in errors),
+            errors,
+        )
+
     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)
@@ -2075,10 +2170,15 @@ class TestWSHandlerSessionSend(unittest.IsolatedAsyncioTestCase):
         checkpoint = ExecutionCheckpoint(
             checkpoint_id="org-lock-free-gate",
             project_id="test-project",
-            session_id=runtime_session_id,
+            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)
@@ -5405,6 +5505,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",
@@ -5835,6 +5970,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())