fix(agents): scope delegation ledger IDs to parent runs (#5898)

* fix(agents): scope delegation ledger IDs to parent runs

* fix(agents): preserve checkpointed replies from resumed runs

README.md

@@ -1654,6 +1654,9 @@ After Stop interrupts a delegated task before it returns a reply, the next user
 turn marks that earlier task as cancelled in the agent's durable context so it
 can retry. Existing replies are preserved. Older replies without status metadata
 may still appear in progress; their outcome is not inferred from their text.
+The durable delegation ledger distinguishes calls by run and provider tool-call
+ID, so a later user turn can reuse an ID without replacing earlier work or
+losing that turn's per-run delegation count.
 
 The lead agent can spawn sub-agents on the fly — each with its own scoped context, tools, and termination conditions — when delegation has clear net benefit from real parallel latency, specialist capability, or context isolation. It keeps interdependent scopes and overlapping side effects out of parallel dispatch; a bounded sequential chain can still run in one sub-agent when specialist or context-isolation benefit clearly wins. The lead uses the fewest useful sub-agents and re-evaluates later batches instead of fanning out solely because a task is large or multi-step. Sub-agents report back structured results, and the lead agent verifies and synthesizes them into a coherent output. Deterministic tool receipts cover both direct tool messages and state-updating `Command` results such as delegated `task` responses; when the receipt ledger reaches its context budget, it retains the newest actions and their original receipt IDs. Operators can disable this provenance layer with `verification.receipts_enabled: false`. Their configured skills are resolved from the same user-scoped catalog as the lead agent, so user-owned custom skills remain available without exposing another user's version. Their internal AI and tool messages stay scoped to the delegated graph instead of entering the parent chat stream. Reloaded thread history enforces the same boundary: callback-captured sub-agent AI responses remain available in run-event diagnostics but are excluded from the parent transcript, while the parent `task` result remains attached to its subtask card. Long-running sub-agents compact older history when summarization is enabled and re-inject the summary as guarded, hidden durable context before continuing, so recent assistant/tool activity remains grounded in the task. Their system instructions, including the role and report contract, survive compaction; if only those instructions and the current request would be summarized, compaction is skipped. Provider/model request failures are reported as failed sub-agent tasks rather than successful results, so the lead agent and Web UI can react to them correctly. Concurrent parent runs also receive independent server-side sub-agent execution IDs, so a provider that reuses a tool-call ID cannot make one run poll, cancel, or clean up another run's background task. Collapsed sub-agent cards show the effective model and, when the provider returns usage metadata, a cumulative token total that updates after each completed sub-agent LLM call and persists after a reload. When token usage tracking is enabled, completed sub-agent usage is attributed back to the dispatching step from that run's terminal tool-message metadata rather than a process-global provider-ID cache.
 

backend/packages/harness/deerflow/agents/AGENTS.md

@@ -28,7 +28,7 @@ same resulting prompt. Empty overlays preserve bytes.
 
 **ThreadState** (`packages/harness/deerflow/agents/thread_state.py`):
 - Extends `AgentState` with: `sandbox`, `thread_data`, `title`, `artifacts`, `todos`, `uploaded_files`, `viewed_images`, `goal`, `promoted`, `delegations`, `skill_context`, `summary_text`
-- Uses custom reducers: `merge_artifacts` (deduplicate), `merge_viewed_images` (merge/clear), `merge_goal` (preserve the active goal across ordinary state updates unless the goal writer replaces it), `merge_promoted` (catalog-hash-scoped deferred tool promotions), `merge_delegations` (append task delegation entries, same id latest wins, terminal status never downgraded, capped to the most recent entries), and `merge_skill_context` (dedupe active-skill references by path, keep the most recently read entries; entries store a name/path/description reference, not the SKILL.md body). `summary_text` is a LastValue channel updated by summarization and projected into model requests as durable context data instead of being stored as a `messages` item.
+- Uses custom reducers: `merge_artifacts` (deduplicate), `merge_viewed_images` (merge/clear), `merge_goal` (preserve the active goal across ordinary state updates unless the goal writer replaces it), `merge_promoted` (catalog-hash-scoped deferred tool promotions), `merge_delegations` (append task delegation entries, same (run,id) wins, terminal status never downgraded, capped to the most recent entries), and `merge_skill_context` (dedupe active-skill references by path, keep the most recently read entries; entries store a name/path/description reference, not the SKILL.md body). `summary_text` is a LastValue channel updated by summarization and projected into model requests as durable context data instead of being stored as a `messages` item.
 - Delta-mode `merge_message_writes` normalizes the current message state once,
   then folds normalized writes in order with message-ID position indexes and
   deferred tombstone compaction. It preserves public `add_messages` behavior,

backend/packages/harness/deerflow/agents/middlewares/AGENTS.md

@@ -14,7 +14,7 @@ ones, and treat completed work as reusable evidence rather than acceptance.
 
 On new user turns, DurableContext cancels earlier-run unanswered delegations.
 It preserves resumes, same-run continuations, and entries without `run_id`.
-Any reply prevents cancellation; legacy replies without status metadata may
+A pre-turn reply prevents cancellation; legacy replies without status metadata may
 stay `in_progress`. Never infer status from reply text.
 
 Assembly order: `tool_error_handling_middleware.py::_build_runtime_middlewares` (exposed as `build_lead_runtime_middlewares`), then `../lead_agent/agent.py::build_middlewares` appends lead-only entries. Optional entries require their config/runtime condition.
@@ -115,6 +115,7 @@ Before changing a later authorization phase, read the [authorization RFC](../../
 27. **DeferredToolFilterMiddleware** - *(optional, if `tool_search.enabled`)* Hides deferred (MCP) tool schemas from the bound model until `tool_search` or `McpRoutingMiddleware` promotes them (reads per-thread promotions from `ThreadState.promoted`, hash-scoped)
 28. **SystemMessageCoalescingMiddleware** - Merges every SystemMessage into a single leading SystemMessage per request; provider-agnostic fix for strict backends (vLLM/SGLang/Qwen/Anthropic) that reject non-leading system messages. Touches the per-request payload only (checkpoint state unchanged); on midnight crossings only the latest `dynamic_context_reminder` SystemMessage survives. The subagent builder places its date-only context middleware immediately before this coalescer, so the built-in subagent prompt and hidden date reminder still reach providers as one leading system block
 29. **SubagentLimitMiddleware** - *(optional, if `subagent_enabled`)* Truncates excess ordinary `task` tool calls to enforce both the per-response concurrency limit (`max_concurrent_subagents`, resolved against startup `subagent_runtime.max_running` and the 1-64 safety range before construction) and the per-run total delegation cap (`max_total_subagents` runtime override or `subagents.max_total_per_run`, default 6, clamped to 1-50). The total cap counts current-run entries in the durable delegation ledger (entries are tagged with `run_id` when captured), so repeated planning checkpoints in one run cannot keep launching legal-sized batches indefinitely, while later user turns in the same thread get a fresh run budget. Explicit durable `batch_task` calls are a separate mode with persisted total/live/running limits and are not rewritten into ordinary ledger entries. If the ordinary cap is exhausted, the middleware strips remaining `task` calls, forces `finish_reason="stop"`, and appends a visible limit note so the run can synthesize existing results instead of ending with an empty tool-call response.
+   Delegations use `(run_id, tool_call_id)` for capture and retention. Later-run replies cannot close older orphans; missing run IDs match conservatively.
 30. **LoopDetectionMiddleware** - *(optional, if `loop_detection.enabled`)* Detects repeated tool-call loops; hard-stop clears structured, raw, and content-block tool calls before forcing a final text answer; stamps `loop_capped` via `consume_stop_reason` (#3875 Phase 2), symmetric to `TokenBudgetMiddleware`; persists warned-state transitions (first per call hash or per tool-frequency burst) and hard stops as `middleware:loop_detection`, attributed with `is_subagent` and the optional `agent_id`, without tool arguments, message content, tool results, or argument-derived hashes. Ordinary task subagents get dedicated recorder keys through a parent-loop proxy; never pass `RunJournal` into their isolated loop. Durable batch subagents have no parent run journal and do not persist these transitions
    State is run-scoped: new user runs get fresh budgets; same-run goal
    continuations share history. Keep sibling warnings isolated and lifecycle

backend/packages/harness/deerflow/agents/middlewares/durable_context_middleware.py

@@ -46,6 +46,10 @@ _AUTHORITY_CONTRACT = "\n".join(
 _DELEGATION_STABLE_FIELDS = ("description", "subagent_type", "status", "run_id", "result_brief", "result_sha256", "result_ref")
 
 
+def _delegation_identity(entry: dict) -> tuple[str | None, str | None]:
+    return entry.get("run_id") or None, entry.get("id")
+
+
 def _normalize_skills_root(skills_container_path: str | None) -> str:
     return posixpath.normpath(skills_container_path or DEFAULT_SKILLS_CONTAINER_PATH)
 
@@ -93,10 +97,12 @@ def _retained_delegation_window(delegations: list[dict], existing: list[dict]) -
     if len(existing) < _DELEGATION_LEDGER_MAX_ENTRIES or not existing:
         return delegations
 
-    earliest_retained_id = existing[0].get("id") if isinstance(existing[0], dict) else None
-    if earliest_retained_id is not None:
+    earliest = existing[0] if isinstance(existing[0], dict) else None
+    if earliest is not None:
+        earliest_run_id, earliest_id = _delegation_identity(earliest)
         for index, entry in enumerate(delegations):
-            if entry.get("id") == earliest_retained_id:
+            entry_run_id, entry_id = _delegation_identity(entry)
+            if entry_id == earliest_id and (entry_run_id is None or entry_run_id == earliest_run_id):
                 return delegations[index:]
 
     return delegations[-_DELEGATION_LEDGER_MAX_ENTRIES:]
@@ -104,10 +110,12 @@ def _retained_delegation_window(delegations: list[dict], existing: list[dict]) -
 
 def _filter_changed_delegations(delegations: list[dict], existing: list[dict]) -> list[dict]:
     comparable_delegations = _retained_delegation_window(delegations, existing)
+    existing_by_identity = {_delegation_identity(entry): entry for entry in existing if isinstance(entry, dict)}
     existing_by_id = {entry.get("id"): entry for entry in existing if isinstance(entry, dict)}
     changed: list[dict] = []
     for entry in comparable_delegations:
-        previous = existing_by_id.get(entry.get("id"))
+        entry_run_id, entry_id = _delegation_identity(entry)
+        previous = existing_by_identity.get((entry_run_id, entry_id)) if entry_run_id is not None else existing_by_id.get(entry_id)
         if previous is None:
             changed.append(entry)
             continue
@@ -175,7 +183,7 @@ def _run_opening_human_index(messages: list[AnyMessage], run_id: str, pre_existi
     return None
 
 
-def _current_run_messages(messages: list[AnyMessage], run_id: str | None, pre_existing_message_ids: frozenset[str]) -> list[AnyMessage]:
+def _current_run_messages(messages: list[AnyMessage], run_id: str | None, pre_existing_message_ids: frozenset[str], opening_index: int | None) -> list[AnyMessage]:
     """Return the message tail where this invocation may have emitted tasks.
 
     The worker supplies the message ids that existed before this run, so a
@@ -184,13 +192,12 @@ def _current_run_messages(messages: list[AnyMessage], run_id: str | None, pre_ex
     """
     if run_id is None:
         return messages
-    index = _run_opening_human_index(messages, run_id, pre_existing_message_ids)
-    if index is not None:
-        return messages[index + 1 :]
+    if opening_index is not None:
+        return messages[opening_index + 1 :]
     return _messages_after_pre_existing_boundary(messages, pre_existing_message_ids)
 
 
-def _close_delegations_left_by_earlier_runs(messages: list[AnyMessage], existing: list[dict], run_id: str) -> list[dict]:
+def _close_delegations_left_by_earlier_runs(messages: list[AnyMessage], existing: list[dict], run_id: str, opening_index: int) -> list[dict]:
     """Mark delegations that an earlier run left in_progress without a result as cancelled.
 
     A ``task`` call waits for its subagent, so an entry that is still
@@ -198,33 +205,49 @@ def _close_delegations_left_by_earlier_runs(messages: list[AnyMessage], existing
     run that was stopped while the subagent ran. Nothing else will ever update
     it, and the ledger would keep telling the model not to delegate again.
 
-    Any recorded reply excludes this inference, including legacy ToolMessages
-    without subagent status metadata. Their outcome is unknown, not evidence
-    of cancellation. Conservatively leave those entries unchanged, even if
-    they remain in_progress: this repairs missing replies, not legacy results.
+    Any reply before the current run's opening HumanMessage excludes this
+    inference, including legacy ToolMessages without subagent status metadata.
+    A resumed run has no opening HumanMessage, so a saved reply cannot safely
+    be assigned to the preceding user's run. Preserve unmarked runs with a
+    matching reply even across repeated IDs; the reply's owner is ambiguous.
     Current task producers stamp metadata for extract_delegations to capture.
     """
-    answered = {str(message.tool_call_id) for message in messages if isinstance(message, ToolMessage) and message.tool_call_id}
-    return [{**entry, "status": "cancelled"} for entry in existing if isinstance(entry, dict) and entry.get("status") == "in_progress" and entry.get("run_id") not in (None, run_id) and entry.get("id") not in answered]
+    answered: set[tuple[str | None, str]] = set()
+    marked_run_ids: set[str] = set()
+    replied_ids: set[str] = set()
+    message_run_id: str | None = None
+    for message in messages[:opening_index]:
+        if isinstance(message, HumanMessage):
+            marker = message.additional_kwargs.get("run_id")
+            message_run_id = str(marker) if marker else None
+            if message_run_id is not None:
+                marked_run_ids.add(message_run_id)
+        elif isinstance(message, ToolMessage) and message.tool_call_id:
+            tool_call_id = str(message.tool_call_id)
+            answered.add((message_run_id, tool_call_id))
+            replied_ids.add(tool_call_id)
+
+    cancelled = []
+    for entry in existing:
+        if not isinstance(entry, dict) or entry.get("status") != "in_progress":
+            continue
+        entry_run_id = entry.get("run_id")
+        entry_id = entry.get("id")
+        if entry_run_id in (None, run_id) or (entry_run_id, entry_id) in answered or (None, entry_id) in answered:
+            continue
+        # Command(resume=...) can checkpoint a reply before ledger capture,
+        # without a HumanMessage carrying that run's id. Its owner is unknown.
+        if entry_run_id not in marked_run_ids and entry_id in replied_ids:
+            continue
+        cancelled.append({**entry, "status": "cancelled"})
+    return cancelled
 
 
-def _with_run_id(delegations: list[dict], run_id: str | None, existing: list[dict]) -> list[dict]:
-    """Tag only new delegation ids with the current run_id."""
+def _with_run_id(delegations: list[dict], run_id: str | None) -> list[dict]:
+    """Tag delegations from the current run's bounded message window."""
     if run_id is None:
         return delegations
-    existing_by_id = {entry.get("id"): entry for entry in existing if isinstance(entry, dict)}
-    tagged: list[dict] = []
-    for entry in delegations:
-        previous = existing_by_id.get(entry.get("id"))
-        if previous is not None:
-            previous_run_id = previous.get("run_id")
-            if previous_run_id:
-                tagged.append({**entry, "run_id": previous_run_id})
-            else:
-                tagged.append({key: value for key, value in entry.items() if key != "run_id"})
-            continue
-        tagged.append({**entry, "run_id": run_id})
-    return tagged
+    return [{**entry, "run_id": run_id} for entry in delegations]
 
 
 class DurableContextMiddleware(AgentMiddleware[AgentState]):
@@ -272,14 +295,15 @@ class DurableContextMiddleware(AgentMiddleware[AgentState]):
     def _capture_delegations(self, state: AgentState, runtime: Runtime | None) -> dict | None:
         run_id = _runtime_run_id(runtime)
         pre_existing_message_ids = _runtime_pre_existing_message_ids(runtime)
-        messages = _current_run_messages(state["messages"], run_id, pre_existing_message_ids)
+        opening_index = _run_opening_human_index(state["messages"], run_id, pre_existing_message_ids) if run_id is not None else None
+        messages = _current_run_messages(state["messages"], run_id, pre_existing_message_ids, opening_index)
         existing = state.get("delegations") or []
         delegations = _filter_changed_delegations(
-            _with_run_id(extract_delegations(messages), run_id, existing),
+            _with_run_id(extract_delegations(messages), run_id),
             existing,
         )
-        if run_id is not None and _run_opening_human_index(state["messages"], run_id, pre_existing_message_ids) is not None:
-            delegations = [*delegations, *_close_delegations_left_by_earlier_runs(state["messages"], existing, run_id)]
+        if run_id is not None and opening_index is not None:
+            delegations = [*delegations, *_close_delegations_left_by_earlier_runs(state["messages"], existing, run_id, opening_index)]
         if delegations:
             return {"delegations": delegations}
         return None

backend/packages/harness/deerflow/agents/thread_state.py

@@ -193,28 +193,34 @@ def merge_delegations(existing: list[DelegationEntry] | None, new: list[Delegati
     """Reducer for the delegation ledger.
 
     - new None/empty -> preserve existing.
-    - append entries, replacing same id with the latest version while preserving
-      first-seen order.
+    - append entries, replacing the same (run_id, tool-call id) with the latest
+      version while preserving first-seen order. Provider ids can repeat in
+      later runs of the same thread.
     - terminal status is never overwritten by a non-terminal status.
     """
     if not new:
         return existing or []
 
-    by_id: dict[str, DelegationEntry] = {}
-    order: list[str] = []
+    by_key: dict[tuple[str | None, str], DelegationEntry] = {}
+    order: list[tuple[str | None, str]] = []
     for entry in [*(existing or []), *new]:
         entry_id = entry["id"]
-        previous = by_id.get(entry_id)
+        key = (entry.get("run_id") or None, entry_id)
+        if key[0] is None:
+            # Legacy updates without run_id still update the most recent
+            # matching entry, as they did before run-scoped identities.
+            key = next((prior for prior in reversed(order) if prior[1] == entry_id), key)
+        previous = by_key.get(key)
         if previous is not None and previous["status"] in TERMINAL_STATUSES and entry["status"] not in TERMINAL_STATUSES:
             continue
-        if entry_id not in by_id:
-            order.append(entry_id)
+        if key not in by_key:
+            order.append(key)
         elif previous.get("created_at"):
             entry = {**entry, "created_at": previous["created_at"]}
             if previous.get("run_id") and not entry.get("run_id"):
                 entry["run_id"] = previous["run_id"]
-        by_id[entry_id] = entry
-    merged = [by_id[entry_id] for entry_id in order]
+        by_key[key] = entry
+    merged = [by_key[key] for key in order]
     if len(merged) > _DELEGATION_LEDGER_MAX_ENTRIES:
         merged = merged[-_DELEGATION_LEDGER_MAX_ENTRIES:]
     return merged

backend/tests/test_delegation_ledger.py

@@ -69,6 +69,26 @@ class TestMergeDelegations:
 
         assert out[0]["run_id"] == "run-1"
 
+    def test_same_tool_call_id_in_different_runs_keeps_both_delegations(self):
+        existing = [{**_entry("call_1", "completed", "old task"), "run_id": "run-old"}]
+        new = [{**_entry("call_1", "in_progress", "new task"), "run_id": "run-new"}]
+
+        merged = merge_delegations(existing, new)
+
+        assert [(entry["run_id"], entry["description"], entry["status"]) for entry in merged] == [
+            ("run-old", "old task", "completed"),
+            ("run-new", "new task", "in_progress"),
+        ]
+
+    def test_same_tool_call_id_in_one_run_still_updates_one_delegation(self):
+        existing = [{**_entry("call_1", "in_progress"), "run_id": "run-1"}]
+        new = [{**_entry("call_1", "completed"), "run_id": "run-1"}]
+
+        merged = merge_delegations(existing, new)
+
+        assert len(merged) == 1
+        assert merged[0]["status"] == "completed"
+
     def test_over_cap_keeps_most_recent_entries(self):
         from deerflow.agents import thread_state as thread_state_module
 

backend/tests/test_durable_context_middleware.py

@@ -152,6 +152,235 @@ class TestBeforeModelCapture:
         assert out is not None
         assert out["delegations"][0]["run_id"] == "run-42"
 
+    def test_reused_tool_call_id_is_captured_for_new_run_and_counts_toward_limit(self):
+        middleware = DurableContextMiddleware()
+        runtime = SimpleNamespace(context={"run_id": "run-new"})
+        existing = [
+            {
+                "id": "call_1",
+                "run_id": "run-old",
+                "description": "old task",
+                "subagent_type": "general-purpose",
+                "status": "completed",
+                "created_at": "2026-07-11T00:00:00Z",
+            }
+        ]
+        current_call = AIMessage(
+            content="",
+            tool_calls=[{"name": "task", "args": {"description": "new task", "subagent_type": "general-purpose"}, "id": "call_1", "type": "tool_call"}],
+        )
+
+        update = middleware.after_model(
+            {"messages": [HumanMessage(content="new request", additional_kwargs={"run_id": "run-new"}), current_call], "delegations": existing},
+            runtime,
+        )
+
+        assert update is not None
+        ledger = merge_delegations(existing, update["delegations"])
+        assert [(entry["run_id"], entry["description"]) for entry in ledger] == [("run-old", "old task"), ("run-new", "new task")]
+
+        next_call = AIMessage(content="", tool_calls=[{"name": "task", "args": {"description": "extra task"}, "id": "call_2", "type": "tool_call"}])
+        blocked = SubagentLimitMiddleware(max_concurrent=2, max_total=1).after_model({"messages": [next_call], "delegations": ledger}, runtime)
+        assert blocked is not None
+        assert blocked["messages"][0].tool_calls == []
+        assert runtime.context["stop_reason"] == "subagent_limit_capped"
+
+    def test_reused_id_does_not_hide_new_work_at_full_ledger_window(self):
+        middleware = DurableContextMiddleware()
+        existing = [
+            {
+                "id": f"call_{index}",
+                "run_id": "run-old",
+                "description": f"old task {index}",
+                "subagent_type": "general-purpose",
+                "status": "completed",
+                "created_at": "2026-07-11T00:00:00Z",
+            }
+            for index in range(thread_state_module._DELEGATION_LEDGER_MAX_ENTRIES)
+        ]
+        current_call = AIMessage(
+            content="",
+            tool_calls=[
+                {"name": "task", "args": {"description": "first new task"}, "id": "new-call", "type": "tool_call"},
+                {"name": "task", "args": {"description": "second new task"}, "id": "call_0", "type": "tool_call"},
+            ],
+        )
+
+        update = middleware.after_model(
+            {"messages": [HumanMessage(content="new request", additional_kwargs={"run_id": "run-new"}), current_call], "delegations": existing},
+            SimpleNamespace(context={"run_id": "run-new"}),
+        )
+
+        assert update is not None
+        assert [(entry["run_id"], entry["id"]) for entry in update["delegations"]] == [("run-new", "new-call"), ("run-new", "call_0")]
+        ledger = merge_delegations(existing, update["delegations"])
+        assert len(ledger) == thread_state_module._DELEGATION_LEDGER_MAX_ENTRIES
+        assert [(entry["run_id"], entry["id"]) for entry in ledger[-2:]] == [("run-new", "new-call"), ("run-new", "call_0")]
+
+    def test_new_run_reply_with_reused_id_does_not_keep_old_orphan_in_progress(self):
+        middleware = DurableContextMiddleware()
+        existing = [
+            {
+                "id": "call_1",
+                "run_id": "run-old",
+                "description": "old task",
+                "subagent_type": "general-purpose",
+                "status": "in_progress",
+                "created_at": "2026-07-11T00:00:00Z",
+            }
+        ]
+        messages = [
+            HumanMessage(content="old request", additional_kwargs={"run_id": "run-old"}),
+            AIMessage(content="", tool_calls=[{"name": "task", "args": {"description": "old task"}, "id": "call_1", "type": "tool_call"}]),
+            HumanMessage(content="new request", additional_kwargs={"run_id": "run-new"}),
+            AIMessage(content="", tool_calls=[{"name": "task", "args": {"description": "new task"}, "id": "call_1", "type": "tool_call"}]),
+            ToolMessage(content="Task Succeeded. Result: done", tool_call_id="call_1", additional_kwargs=make_subagent_additional_kwargs("completed", result="done")),
+        ]
+
+        update = middleware.before_model({"messages": messages, "delegations": existing}, SimpleNamespace(context={"run_id": "run-new"}))
+
+        assert update is not None
+        ledger = merge_delegations(existing, update["delegations"])
+        assert [(entry["run_id"], entry["status"]) for entry in ledger] == [("run-old", "cancelled"), ("run-new", "completed")]
+
+    def test_unmarked_new_run_reply_with_reused_id_does_not_answer_old_orphan(self):
+        middleware = DurableContextMiddleware()
+        existing = [
+            {
+                "id": "call_1",
+                "run_id": "run-old",
+                "description": "old task",
+                "subagent_type": "general-purpose",
+                "status": "in_progress",
+                "created_at": "2026-07-11T00:00:00Z",
+            }
+        ]
+        messages = [
+            HumanMessage(id="old-human", content="old request"),
+            AIMessage(id="old-ai", content="", tool_calls=[{"name": "task", "args": {"description": "old task"}, "id": "call_1", "type": "tool_call"}]),
+            HumanMessage(id="new-human", content="new request"),
+            AIMessage(id="new-ai", content="", tool_calls=[{"name": "task", "args": {"description": "new task"}, "id": "call_1", "type": "tool_call"}]),
+            ToolMessage(id="new-result", content="Task Succeeded. Result: done", tool_call_id="call_1", additional_kwargs=make_subagent_additional_kwargs("completed", result="done")),
+        ]
+        runtime = SimpleNamespace(context={"run_id": "run-new", CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY: {"old-human", "old-ai"}})
+
+        update = middleware.before_model({"messages": messages, "delegations": existing}, runtime)
+
+        assert update is not None
+        ledger = merge_delegations(existing, update["delegations"])
+        assert [(entry["run_id"], entry["status"]) for entry in ledger] == [("run-old", "cancelled"), ("run-new", "completed")]
+
+    def test_unmarked_prior_reply_preserves_earlier_delegation(self):
+        middleware = DurableContextMiddleware()
+        existing = [
+            {
+                "id": "call_1",
+                "run_id": "run-old",
+                "description": "old task",
+                "subagent_type": "general-purpose",
+                "status": "in_progress",
+                "created_at": "2026-07-11T00:00:00Z",
+            }
+        ]
+        messages = [
+            HumanMessage(id="old-human", content="old request"),
+            AIMessage(id="old-ai", content="", tool_calls=[{"name": "task", "args": {"description": "old task"}, "id": "call_1", "type": "tool_call"}]),
+            ToolMessage(id="old-result", content="legacy result", tool_call_id="call_1"),
+            HumanMessage(id="new-human", content="new request"),
+        ]
+        runtime = SimpleNamespace(context={"run_id": "run-new", CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY: {"old-human", "old-ai", "old-result"}})
+
+        update = middleware.before_model({"messages": messages, "delegations": existing}, runtime)
+
+        assert update is None
+
+    def test_resumed_run_without_new_human_does_not_close_earlier_delegation(self):
+        middleware = DurableContextMiddleware()
+        existing = [
+            {
+                "id": "call_1",
+                "run_id": "run-old",
+                "description": "old task",
+                "subagent_type": "general-purpose",
+                "status": "in_progress",
+                "created_at": "2026-07-11T00:00:00Z",
+            }
+        ]
+        messages = [
+            HumanMessage(id="old-human", content="old request"),
+            AIMessage(id="old-ai", content="", tool_calls=[{"name": "task", "args": {"description": "old task"}, "id": "call_1", "type": "tool_call"}]),
+            AIMessage(id="new-ai", content="", tool_calls=[{"name": "task", "args": {"description": "new task"}, "id": "call_1", "type": "tool_call"}]),
+            ToolMessage(id="new-result", content="Task Succeeded. Result: done", tool_call_id="call_1", additional_kwargs=make_subagent_additional_kwargs("completed", result="done")),
+        ]
+        runtime = SimpleNamespace(context={"run_id": "run-new", CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY: {"old-human", "old-ai"}})
+
+        update = middleware.before_model({"messages": messages, "delegations": existing}, runtime)
+
+        assert update is not None
+        ledger = merge_delegations(existing, update["delegations"])
+        assert [(entry["run_id"], entry["status"]) for entry in ledger] == [("run-old", "in_progress"), ("run-new", "completed")]
+
+    @pytest.mark.parametrize("resumed_call_id", ["resumed-call", "shared-call"])
+    def test_next_turn_keeps_reply_checkpointed_by_resumed_run(self, resumed_call_id):
+        """A resumed run has no HumanMessage; Stop can leave its reply saved before ledger capture."""
+        middleware = DurableContextMiddleware()
+        existing = [
+            {"id": "shared-call", "run_id": "run-a", "description": "old task", "subagent_type": "general-purpose", "status": "in_progress", "created_at": "2026-07-11T00:00:00Z"},
+            {"id": resumed_call_id, "run_id": "run-b", "description": "resumed task", "subagent_type": "general-purpose", "status": "in_progress", "created_at": "2026-07-11T00:00:01Z"},
+        ]
+        messages = [
+            HumanMessage(content="original request", additional_kwargs={"run_id": "run-a"}),
+            AIMessage(content="", tool_calls=[{"name": "task", "args": {"description": "old task"}, "id": "shared-call", "type": "tool_call"}]),
+            AIMessage(content="", tool_calls=[{"name": "task", "args": {"description": "resumed task"}, "id": resumed_call_id, "type": "tool_call"}]),
+            ToolMessage(content="Task Succeeded. Result: done", tool_call_id=resumed_call_id, additional_kwargs=make_subagent_additional_kwargs("completed", result="done")),
+            HumanMessage(content="continue", additional_kwargs={"run_id": "run-c"}),
+        ]
+
+        update = middleware.before_model({"messages": messages, "delegations": existing}, SimpleNamespace(context={"run_id": "run-c"}))
+
+        ledger = merge_delegations(existing, update["delegations"] if update else [])
+        assert next(entry for entry in ledger if entry["run_id"] == "run-b")["status"] == "in_progress"
+        if resumed_call_id != "shared-call":
+            assert next(entry for entry in ledger if entry["run_id"] == "run-a")["status"] == "cancelled"
+        else:
+            assert next(entry for entry in ledger if entry["run_id"] == "run-a")["status"] == "in_progress"
+
+    def test_earlier_reply_conservatively_preserves_unmarked_reuse(self):
+        middleware = DurableContextMiddleware()
+        existing = [
+            {"id": "shared-call", "run_id": "run-a", "description": "old task", "subagent_type": "general-purpose", "status": "in_progress", "created_at": "2026-07-11T00:00:00Z"},
+            {"id": "shared-call", "run_id": "run-b", "description": "resumed task", "subagent_type": "general-purpose", "status": "in_progress", "created_at": "2026-07-11T00:00:01Z"},
+        ]
+        messages = [
+            HumanMessage(content="original request", additional_kwargs={"run_id": "run-a"}),
+            AIMessage(content="", tool_calls=[{"name": "task", "args": {"description": "old task"}, "id": "shared-call", "type": "tool_call"}]),
+            ToolMessage(content="Task Succeeded. Result: old", tool_call_id="shared-call", additional_kwargs=make_subagent_additional_kwargs("completed", result="old")),
+            AIMessage(content="", tool_calls=[{"name": "task", "args": {"description": "resumed task"}, "id": "shared-call", "type": "tool_call"}]),
+            HumanMessage(content="continue", additional_kwargs={"run_id": "run-c"}),
+        ]
+
+        update = middleware.before_model({"messages": messages, "delegations": existing}, SimpleNamespace(context={"run_id": "run-c"}))
+
+        assert update is None
+
+    def test_later_unanswered_reuse_does_not_cancel_earlier_resumed_reply(self):
+        middleware = DurableContextMiddleware()
+        existing = [
+            {"id": "shared-call", "run_id": "run-b", "description": "answered task", "subagent_type": "general-purpose", "status": "in_progress", "created_at": "2026-07-11T00:00:00Z"},
+            {"id": "shared-call", "run_id": "run-d", "description": "unanswered task", "subagent_type": "general-purpose", "status": "in_progress", "created_at": "2026-07-11T00:00:01Z"},
+        ]
+        messages = [
+            HumanMessage(content="original request", additional_kwargs={"run_id": "run-a"}),
+            AIMessage(content="", tool_calls=[{"name": "task", "args": {"description": "answered task"}, "id": "shared-call", "type": "tool_call"}]),
+            ToolMessage(content="Task Succeeded. Result: done", tool_call_id="shared-call", additional_kwargs=make_subagent_additional_kwargs("completed", result="done")),
+            AIMessage(content="", tool_calls=[{"name": "task", "args": {"description": "unanswered task"}, "id": "shared-call", "type": "tool_call"}]),
+            HumanMessage(content="continue", additional_kwargs={"run_id": "run-e"}),
+        ]
+
+        # With no HumanMessage from either resumed run, the saved reply cannot
+        # be assigned safely to B or D. Both entries must remain conservative.
+        assert middleware.before_model({"messages": messages, "delegations": existing}, SimpleNamespace(context={"run_id": "run-e"})) is None
+
     def test_runtime_run_id_capture_starts_at_current_run_message(self):
         middleware = DurableContextMiddleware()
         runtime = SimpleNamespace(context={"run_id": "run-new"})