diff --git a/opc/layer3_agent/runtime_v2/runtime.py b/opc/layer3_agent/runtime_v2/runtime.py index 4b9df30..0b63beb 100644 --- a/opc/layer3_agent/runtime_v2/runtime.py +++ b/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,20 +573,55 @@ class NativeRuntimeV2: ) else: await self._cancel_early_tool_runs(early_tool_runs) - truncated = self._truncate_to_last_clean_user_turn(messages, base_prefix_len) - if truncated and len(truncated) > base_prefix_len: + 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( + runtime_session_id, + task, + "tool_protocol_retry", + { + "iteration": iteration + 1, + "strategy": "truncate", + "message": str(exc), + }, + ) + 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": "truncate", + "strategy": "provider_error_feedback", + "attempt": stream_error_feedback_retries, "message": str(exc), }, ) - messages = truncated 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]], diff --git a/opc/plugins/office_ui/ws_handler.py b/opc/plugins/office_ui/ws_handler.py index 7835105..03296aa 100644 --- a/opc/plugins/office_ui/ws_handler.py +++ b/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::::vN``), + so their runtime tasks carry session ids shaped like + ``:review::::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() diff --git a/tests/test_native_runtime_v2.py b/tests/test_native_runtime_v2.py index 142510d..72d7d0f 100644 --- a/tests/test_native_runtime_v2.py +++ b/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(),