feat(delegation): add same-turn background result injection

agent/conversation_loop.py

@@ -1179,6 +1179,10 @@ def run_conversation(
     active_system_prompt = _ctx.active_system_prompt
     effective_task_id = _ctx.effective_task_id
     turn_id = _ctx.turn_id
+    # delegate_task snapshots this id at dispatch. A late inject result whose
+    # originating turn already ended must stay on the normal synthetic-turn
+    # path instead of leaking into a later user turn.
+    agent._active_turn_id = turn_id
     current_turn_user_idx = _ctx.current_turn_user_idx
     _should_review_memory = _ctx.should_review_memory
     _plugin_user_context = _ctx.plugin_user_context
@@ -1255,7 +1259,31 @@ def run_conversation(
             should_review_memory=_should_review_memory,
         )
 
+    def _release_unconsumed_injects() -> None:
+        try:
+            from agent.delegation_inject import release_pending_injects
+
+            release_pending_injects(agent, messages, turn_id=turn_id)
+        except Exception:
+            logger.debug("Failed to settle unconsumed inject claims", exc_info=True)
+
     while (api_call_count < agent.max_iterations and agent.iteration_budget.remaining > 0) or agent._budget_grace_call:
+        # Safe boundary: the previous assistant tool-call block (if any) and
+        # every corresponding tool result have already been appended. Drain
+        # only already-ready results from this foreground turn.
+        try:
+            from agent.delegation_inject import (
+                drain_ready_injects,
+                ensure_pending_inject_heartbeat,
+            )
+
+            drain_ready_injects(agent, messages, turn_id)
+            # Start the lease before context assembly/compression; those steps
+            # can themselves be slow for very large parent histories.
+            ensure_pending_inject_heartbeat(agent)
+        except Exception:
+            logger.debug("Same-turn delegation inject drain failed", exc_info=True)
+
         _redirect_text = agent._drain_pending_redirect()
         if _redirect_text:
             _apply_active_turn_redirect(agent, messages, _redirect_text)
@@ -1271,6 +1299,12 @@ def run_conversation(
 
         # Check for interrupt request (e.g., user sent new message)
         if agent._interrupt_requested:
+            try:
+                from agent.delegation_inject import release_pending_injects
+
+                release_pending_injects(agent, messages, turn_id=turn_id)
+            except Exception:
+                logger.debug("Failed to release interrupted inject claims", exc_info=True)
             interrupted = True
             _turn_exit_reason = "interrupted_by_user"
             if not agent.quiet_mode:
@@ -5020,6 +5054,10 @@ def _perform_api_call(next_api_kwargs):
                             force=True,
                         )
                     logger.error(f"{agent.log_prefix}Non-retryable client error: {api_error}")
+                    # The provider did not consume a normalized response. Remove
+                    # RAM-only inject markers and release their durable claims
+                    # before any terminal-error persistence can mark them saved.
+                    _release_unconsumed_injects()
                     # Skip session persistence when the error is likely
                     # context-overflow related (status 400 + large session).
                     # Persisting the failed user message would make the
@@ -5230,6 +5268,7 @@ def _perform_api_call(next_api_kwargs):
                         agent._dump_api_request_debug(
                             api_kwargs, reason="max_retries_exhausted", error=api_error,
                         )
+                    _release_unconsumed_injects()
                     agent._persist_session(messages, conversation_history)
                     _billing_block = None
                     if classified.reason == FailoverReason.billing:
@@ -5350,6 +5389,15 @@ def _perform_api_call(next_api_kwargs):
                         agent._vprint(f"{agent.log_prefix}⚡ Interrupt detected during retry wait, aborting.", force=True)
                         _interrupt_text = f"Operation interrupted: retrying API call after error (retry {retry_count}/{max_retries})."
                         close_interrupted_tool_sequence(messages, _interrupt_text)
+                        try:
+                            from agent.delegation_inject import release_pending_injects
+
+                            release_pending_injects(agent, messages, turn_id=turn_id)
+                        except Exception:
+                            logger.debug(
+                                "Failed to release backoff-interrupted inject claims",
+                                exc_info=True,
+                            )
                         agent._persist_session(messages, conversation_history)
                         agent.clear_interrupt()
                         return {
@@ -5453,6 +5501,15 @@ def _perform_api_call(next_api_kwargs):
             normalized = _transport.normalize_response(response, **_normalize_kwargs)
             assistant_message = normalized
             finish_reason = normalized.finish_reason
+            # A normalized provider response is the acceptance boundary for any
+            # synthetic delegation result in this request.  Until here its
+            # durable row remains pending+claimed so a crash can recover it.
+            try:
+                from agent.delegation_inject import acknowledge_pending_injects
+
+                acknowledge_pending_injects(agent, turn_id=turn_id)
+            except Exception:
+                logger.debug("Failed to acknowledge consumed inject claims", exc_info=True)
 
             # Normalize content to string — some OpenAI-compatible servers
             # (llama-server, etc.) return content as a dict or list instead
@@ -6304,6 +6361,16 @@ def _perform_api_call(next_api_kwargs):
                 # gateway kills the session before the next activity
                 # touch fires (#69559, #69131).
                 agent._touch_activity(f"tool results posted, continuing iteration #{api_call_count}")
+                # Safe boundary even when this was the nominal last iteration:
+                # every tool result is present and post-tool compression is done.
+                # Drain once without waiting; a ready inject grants exactly one
+                # reconciliation call only if the normal budget is exhausted.
+                try:
+                    from agent.delegation_inject import drain_ready_injects
+
+                    drain_ready_injects(agent, messages, turn_id)
+                except Exception:
+                    logger.debug("Post-tool delegation inject drain failed", exc_info=True)
                 # Continue loop for next response
                 continue
 
@@ -6911,8 +6978,28 @@ def _perform_api_call(next_api_kwargs):
                     final_response = None
                     continue
 
-                messages.append(final_msg)
-
+                # Treat this answer as provisional until one final
+                # non-blocking inject drain has run. If an auditor/dependency
+                # completed at this boundary, append its report after the
+                # provisional assistant message and reconcile once more.
+                try:
+                    from agent.delegation_inject import reconcile_provisional_final
+
+                    _ready_injects = reconcile_provisional_final(
+                        agent, messages, final_msg, turn_id=turn_id
+                    )
+                except Exception:
+                    logger.debug("Final delegation inject drain failed", exc_info=True)
+                    # Preserve the original final even if the optional drain
+                    # machinery itself fails before appending it.
+                    if not messages or messages[-1] is not final_msg:
+                        messages.append(final_msg)
+                    _ready_injects = False
+                if _ready_injects:
+                    final_response = None
+                    agent._session_messages = messages
+                    continue
+
                 _turn_exit_reason = f"text_response(finish_reason={finish_reason})"
                 if not agent.quiet_mode:
                     agent._safe_print(f"🎉 Conversation completed after {api_call_count} OpenAI-compatible API call(s)")
@@ -7013,10 +7100,22 @@ def _perform_api_call(next_api_kwargs):
                 messages.append({"role": "assistant", "content": final_response})
                 break
 
+    # Any claim left here was never consumed by a normalized provider response.
+    # Drop its RAM-only synthetic message and return the durable event to the
+    # queue.  If a persistence boundary already saved it, that transcript row
+    # is the durable handoff and the helper acknowledges rather than requeues.
+    try:
+        from agent.delegation_inject import release_pending_injects
+
+        release_pending_injects(agent, messages, turn_id=turn_id)
+    except Exception:
+        logger.debug("Failed to settle unconsumed inject claims", exc_info=True)
+
     # Post-loop turn finalization extracted to agent/turn_finalizer.finalize_turn
     # (god-file decomposition Phase 1 step 4). Behavior-neutral: the assembled
     # result dict is returned exactly as before.
     from agent.turn_finalizer import finalize_turn
+
     return finalize_turn(
         agent,
         final_response=final_response,