fix: route internal-turn approval cards to visible channels; feed provider errors back to the model

Approval cards raised by company-mode internal scheduling turns (review/report
work items, session ids like `<root>:review::<wid>::vN`) were posted to the
turn's own session channel, which the UI deliberately hides. The card silently
timed out after 300s and the work item parked on AWAITING_HUMAN, so users saw
only the gate card and never the approval prompt. ws_handler now detects these
internal turns and routes their escalation cards to origin_task_id, the root
session's primary task channel, or the activity channel — never the hidden one.
Also unblocks the previously dead origin/session fallbacks in the resolver.

Unclassified LLM stream failures (e.g. provider content-filter rejections like
"input may contain sensitive information") used to hit a blind truncate-retry
loop that replayed the identical payload for a dozen-plus consecutive failures.
runtime_v2 now feeds the provider's verbatim error text back into the
conversation as a "[runtime notice]" system message so the model can adapt
(rephrase, drop quotes, change tack), bounded at 2 feedback retries (counter
resets on any successful stream) plus one context-reset attempt, then fails
honestly with the real error. The blind truncate path remains only for
classified tool-protocol errors.

Verified: new end-to-end tests for recover-after-notice and bounded-failure;
runtime_v2 + ws_handler + escalation/approval + company-mode suites all green.

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

opc/layer3_agent/runtime_v2/runtime.py

@@ -270,6 +270,13 @@ class NativeRuntimeV2:
             1,
             int(self.config.system.native_runtime.reactive_compaction.max_overflow_retries or 1),
         ) if self.config.system.native_runtime.reactive_compaction.enabled else 1
+        # Unclassified provider failures (content filters, transient rejects)
+        # get bounded retries with the provider's error text fed back into the
+        # conversation so the model can adapt; the counter resets after every
+        # successful stream so long runs are not penalized for sporadic blips.
+        stream_error_feedback_retries = 0
+        max_stream_error_feedback_retries = 2
+        stream_error_context_reset_attempted = False
         compaction_boundaries: list[dict[str, Any]] = []
 
         await self._save_runtime_session(
@@ -566,6 +573,7 @@ class NativeRuntimeV2:
                     )
                 else:
                     await self._cancel_early_tool_runs(early_tool_runs)
+                    if self.llm.is_tool_protocol_error(exc):
                         truncated = self._truncate_to_last_clean_user_turn(messages, base_prefix_len)
                         if truncated and len(truncated) > base_prefix_len:
                             await self._emit_runtime_event(
@@ -580,6 +588,40 @@ class NativeRuntimeV2:
                             )
                             messages = truncated
                             continue
+                    elif stream_error_feedback_retries < max_stream_error_feedback_retries:
+                        stream_error_feedback_retries += 1
+                        messages.append(self._provider_error_feedback_message(exc))
+                        await self._emit_runtime_event(
+                            runtime_session_id,
+                            task,
+                            "tool_protocol_retry",
+                            {
+                                "iteration": iteration + 1,
+                                "strategy": "provider_error_feedback",
+                                "attempt": stream_error_feedback_retries,
+                                "message": str(exc),
+                            },
+                        )
+                        continue
+                    elif not stream_error_context_reset_attempted:
+                        stream_error_context_reset_attempted = True
+                        truncated = self._truncate_to_last_clean_user_turn(messages, base_prefix_len)
+                        if truncated and base_prefix_len < len(truncated) < len(messages):
+                            truncated.append(
+                                self._provider_error_feedback_message(exc, context_reset=True)
+                            )
+                            await self._emit_runtime_event(
+                                runtime_session_id,
+                                task,
+                                "tool_protocol_retry",
+                                {
+                                    "iteration": iteration + 1,
+                                    "strategy": "provider_error_context_reset",
+                                    "message": str(exc),
+                                },
+                            )
+                            messages = truncated
+                            continue
                     await self._emit_runtime_event(
                         runtime_session_id,
                         task,
@@ -607,6 +649,7 @@ class NativeRuntimeV2:
                         token_usage=total_usage,
                     )
 
+            stream_error_feedback_retries = 0
             tool_calls = self._finalize_tool_calls(tool_call_chunks)
             assistant_message = {"role": "assistant", "content": assistant_text}
             if tool_calls:
@@ -1697,6 +1740,41 @@ class NativeRuntimeV2:
                     break
         return None
 
+    @staticmethod
+    def _provider_error_feedback_message(
+        exc: Exception,
+        *,
+        context_reset: bool = False,
+    ) -> dict[str, str]:
+        """Conversation message telling the model why the last request failed.
+
+        Unclassified provider rejections (content filters, transient 4xx) never
+        produce model output, so without this the model has no way to know the
+        request failed or why. Feeding the provider's own error text back lets
+        the model decide how to proceed (rephrase, drop a quote, change tack)
+        instead of the runtime blindly replaying an identical payload.
+        """
+        error_text = " ".join(str(exc).split())[:600]
+        if context_reset:
+            detail = (
+                "The previous LLM request kept failing at the model provider, so the "
+                "intermediate steps of the current turn were dropped from the request."
+            )
+        else:
+            detail = (
+                "The previous LLM request failed at the model provider before any "
+                "output was produced."
+            )
+        return {
+            "role": "system",
+            "content": (
+                f"[runtime notice] {detail} Provider error: {error_text}. "
+                "This was not a user action. Adjust your next step accordingly — for "
+                "example rephrase sensitive wording, avoid quoting flagged content "
+                "verbatim, or choose another way to make progress — then continue the task."
+            ),
+        }
+
     def _truncate_to_last_clean_user_turn(
         self,
         messages: list[dict[str, Any]],

opc/plugins/office_ui/ws_handler.py

@@ -1960,6 +1960,9 @@ class WSHandler:
                 logger.warning(f"Failed to resolve escalation task mapping for {source_task_id}: {e}")
                 task = None
             if task is not None:
+                internal_turn_target = self._company_internal_turn_escalation_target(task)
+                if internal_turn_target is not None:
+                    return internal_turn_target or None
                 ui_task_id = self._ui_task_id_for_task(task)
                 if ui_task_id:
                     return ui_task_id
@@ -1975,6 +1978,45 @@ class WSHandler:
 
         return source_task_id
 
+    def _company_internal_turn_escalation_target(self, task: Any | None) -> str | None:
+        """Visible routing target for escalations raised by internal
+        company-mode scheduling turns.
+
+        Review/report turn work items get composite ids (``review::<wid>::vN``),
+        so their runtime tasks carry session ids shaped like
+        ``<root_session>:review::<wid>::vN``. The UI deliberately hides those
+        session channels, so an approval card posted to the turn's own channel
+        can never be seen or answered — it silently times out and the work item
+        parks on AWAITING_HUMAN.
+
+        Returns None when ``task`` is not such an internal turn (caller keeps
+        its normal resolution), the primary task id of the run's root session
+        when resolvable, or "" when the turn is internal but no visible session
+        is known (caller should fall back to the activity channel rather than
+        the hidden channel).
+        """
+        if task is None:
+            return None
+        session_id = str(getattr(task, "session_id", "") or "").strip()
+        root_session_id, sep, suffix = session_id.partition(":")
+        if not sep or "::" not in suffix:
+            return None
+        metadata = dict(getattr(task, "metadata", {}) or {})
+        origin_task_id = str(metadata.get("origin_task_id") or "").strip()
+        task_id = str(getattr(task, "id", "") or "").strip()
+        if origin_task_id and origin_task_id != task_id:
+            return origin_task_id
+        for candidate_session_id in (
+            root_session_id,
+            str(getattr(task, "parent_session_id", "") or "").strip(),
+        ):
+            if not candidate_session_id:
+                continue
+            mapped_task_id = str(self._session_to_task.get(candidate_session_id) or "").strip()
+            if mapped_task_id and mapped_task_id != task_id:
+                return mapped_task_id
+        return ""
+
     @staticmethod
     def _pending_escalation_matches_task(record: dict[str, Any], task_id: str | None) -> bool:
         task_key = str(task_id or "").strip()

tests/test_native_runtime_v2.py

@@ -1227,6 +1227,120 @@ class NativeRuntimeV2Tests(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(result.status, TaskStatus.DONE)
         self.assertIn("Recovered after provider tool protocol fallback.", result.content)
 
+    @staticmethod
+    def _make_provider_reject_llm(fail_times: int):
+        """LLM stub whose stream fails ``fail_times`` times with an
+        unclassified provider rejection (content-filter style), then answers."""
+
+        class _ProviderRejectLLM:
+            def __init__(self) -> None:
+                self.config = type("Cfg", (), {"max_tokens": 2048})()
+                self.stream_calls = 0
+                self.seen_notice_payloads: list[list[str]] = []
+
+            def prepare_user_message_content(self, content: str, attachment_refs=None):
+                _ = attachment_refs
+                return content
+
+            def get_tool_definitions(self, tools):
+                return tools
+
+            def is_context_overflow_error(self, error: Exception) -> bool:
+                _ = error
+                return False
+
+            def is_tool_protocol_error(self, error: Exception) -> bool:
+                _ = error
+                return False
+
+            def sanitize_tool_call_history(self, messages):
+                return list(messages)
+
+            async def chat_stream(self, messages, tools=None):
+                _ = tools
+                self.stream_calls += 1
+                self.seen_notice_payloads.append([
+                    str(m.get("content", ""))
+                    for m in messages
+                    if m.get("role") == "system" and "[runtime notice]" in str(m.get("content", ""))
+                ])
+                if self.stream_calls <= fail_times:
+                    yield type("Evt", (), {"event_type": "message_start", "payload": {}, "model": "stub"})()
+                    raise RuntimeError(
+                        "litellm.BadRequestError: OpenAIException - The request failed "
+                        "because the input may contain sensitive information."
+                    )
+                yield type("Evt", (), {"event_type": "message_start", "payload": {}, "model": "stub"})()
+                yield type("Evt", (), {
+                    "event_type": "assistant_delta",
+                    "payload": {"text": "Rephrased and continued."},
+                    "model": "stub",
+                })()
+                yield type("Evt", (), {"event_type": "message_stop", "payload": {}, "model": "stub"})()
+
+            async def chat(self, messages, tools=None):
+                raise AssertionError("non-stream fallback must not be used for unclassified errors")
+
+        return _ProviderRejectLLM()
+
+    async def test_unclassified_provider_error_feeds_error_back_and_recovers(self) -> None:
+        llm = self._make_provider_reject_llm(fail_times=1)
+        runtime = NativeRuntimeV2(
+            llm=llm,
+            tool_registry=ToolRegistry(),
+            memory_manager=_StubMemoryManager(_StubStore()),
+            config=OPCConfig(),
+            max_iterations=8,
+        )
+
+        result = await runtime.run(
+            system_prompt="You are a resilient runtime.",
+            user_message="Complete the task.",
+            task=Task(
+                title="provider-reject-recover",
+                description="provider-reject-recover",
+                session_id="sess-provider-reject-recover",
+                project_id="proj1",
+                metadata={"mode": "task"},
+            ),
+        )
+
+        self.assertEqual(result.status, TaskStatus.DONE)
+        self.assertIn("Rephrased and continued.", result.content)
+        self.assertEqual(llm.stream_calls, 2)
+        # The retry request must contain the provider's error text as a notice
+        retry_notices = llm.seen_notice_payloads[1]
+        self.assertEqual(len(retry_notices), 1)
+        self.assertIn("sensitive information", retry_notices[0])
+        self.assertIn("not a user action", retry_notices[0])
+
+    async def test_unclassified_provider_error_retries_are_bounded_then_fail(self) -> None:
+        llm = self._make_provider_reject_llm(fail_times=99)
+        runtime = NativeRuntimeV2(
+            llm=llm,
+            tool_registry=ToolRegistry(),
+            memory_manager=_StubMemoryManager(_StubStore()),
+            config=OPCConfig(),
+            max_iterations=20,
+        )
+
+        result = await runtime.run(
+            system_prompt="You are a resilient runtime.",
+            user_message="Complete the task.",
+            task=Task(
+                title="provider-reject-bounded",
+                description="provider-reject-bounded",
+                session_id="sess-provider-reject-bounded",
+                project_id="proj1",
+                metadata={"mode": "task"},
+            ),
+        )
+
+        self.assertEqual(result.status, TaskStatus.FAILED)
+        self.assertIn("sensitive information", result.content)
+        # 1 initial + 2 feedback retries + 1 context-reset retry = 4, never 20
+        self.assertLessEqual(llm.stream_calls, 4)
+
     async def test_todo_write_normalizes_openopc_task_ledger_shape(self) -> None:
         runtime = NativeRuntimeV2(
             llm=_StubLLM(),