stdin closes after a delegated task settles before Result, dropping a later continuation control response

[starvex]

Summary

The task-lifecycle mitigation from #1088 keeps stdin open when a Result
arrives while a local_agent is still active. A residual ordering remains when
the same task becomes terminal immediately before that Result:

task_started(local_agent)
task_notification(completed)
Result
late parent continuation control request
final Result

At the first Result, _inflight_tasks is empty, so
_first_result_event.set() wakes wait_for_result_and_end_input() and stdin is
closed. The later permission/hook/SDK-MCP response can no longer be written.

The source comment in Query._track_task_lifecycle() already describes this
specific ambiguity as outside the current mitigation.

Versions reproduced

• claude-agent-sdk==0.2.127 (bundled Claude Code 2.1.219)
• claude-agent-sdk==0.2.131 (bundled Claude Code 2.1.223)

Minimal reproduction

The repro.py below uses only a synthetic custom Transport; it makes no
API or model call. Run it in a clean environment:

python -m venv .venv
. .venv/bin/activate
pip install claude-agent-sdk==0.2.131
python repro.py

Actual behavior

stdin_closed_before_late_control=True
late_control_response_delivered=False
late_response_write_after_close=True
end_input_calls=1

The script exits 1 because the response cannot cross the already-closed input
channel.

Expected behavior

stdin should remain available until the continuation that follows delegated
task completion reaches its final Result, so the late control response is
delivered. end_input() should then run exactly once.

Related

• SDK closes stdin on first result frame, breaking bidirectional control from nested Task/Agent subagents #1088
• fix(query): don't close stdin on a result frame while tasks are in flight #1103

Would you accept a CLI run-boundary signal that distinguishes a per-turn
Result from a run-ending Result? Task bookkeeping alone cannot distinguish
"settled task with a pending continuation" from "no remaining work" without
that signal.

repro.py

"""Synthetic reproduction for a Claude Agent SDK stdin lifecycle gap."""

from __future__ import annotations

import asyncio
import json
from typing import Any, AsyncIterator

from claude_agent_sdk import (
    ClaudeAgentOptions,
    HookMatcher,
    PermissionResultAllow,
    ResultMessage,
    Transport,
    query,
)


class SyntheticTransport(Transport):
    def __init__(self) -> None:
        self.ready = False
        self.closed_for_input = False
        self.incoming: asyncio.Queue[dict[str, Any] | None] = asyncio.Queue()
        self.end_input_calls = 0
        self.late_response_delivered = asyncio.Event()
        self.late_write_after_close = False

    async def connect(self) -> None:
        self.ready = True

    async def write(self, data: str) -> None:
        payload = json.loads(data)
        if (
            payload.get("type") == "control_request"
            and payload.get("request", {}).get("subtype") == "initialize"
        ):
            await self.incoming.put(
                {
                    "type": "control_response",
                    "response": {
                        "subtype": "success",
                        "request_id": payload["request_id"],
                        "response": {},
                    },
                }
            )
            return

        if payload.get("type") == "user":
            await self.incoming.put(
                {
                    "type": "system",
                    "subtype": "task_started",
                    "task_id": "agent-1",
                    "task_type": "local_agent",
                    "description": "synthetic task",
                    "uuid": "task-start",
                    "session_id": "session-1",
                }
            )
            await self.incoming.put(
                {
                    "type": "system",
                    "subtype": "task_notification",
                    "task_id": "agent-1",
                    "status": "completed",
                    "output_file": "/tmp/synthetic-task.output",
                    "summary": "synthetic task completed",
                    "uuid": "task-complete",
                    "session_id": "session-1",
                }
            )
            await self.incoming.put(result_frame("before-continuation"))
            return

        if (
            payload.get("type") == "control_response"
            and payload.get("response", {}).get("request_id") == "late-control"
        ):
            if self.closed_for_input:
                self.late_write_after_close = True
                raise BrokenPipeError("stdin already closed")
            self.late_response_delivered.set()

    async def read_messages(self) -> AsyncIterator[dict[str, Any]]:
        while True:
            item = await self.incoming.get()
            if item is None:
                return
            yield item

    async def end_input(self) -> None:
        self.end_input_calls += 1
        self.closed_for_input = True

    async def close(self) -> None:
        self.ready = False

    def is_ready(self) -> bool:
        return self.ready

    async def send_late_control(self) -> None:
        await self.incoming.put(
            {
                "type": "control_request",
                "request_id": "late-control",
                "request": {
                    "subtype": "can_use_tool",
                    "tool_name": "Read",
                    "input": {"file_path": "/tmp/synthetic.txt"},
                    "tool_use_id": "late-tool",
                },
            }
        )

    async def finish(self) -> None:
        await self.incoming.put(result_frame("final"))
        await self.incoming.put(None)


def result_frame(result: str) -> dict[str, Any]:
    return {
        "type": "result",
        "subtype": "success",
        "duration_ms": 1,
        "duration_api_ms": 1,
        "is_error": False,
        "num_turns": 1,
        "session_id": "session-1",
        "result": result,
    }


async def prompt_stream() -> AsyncIterator[dict[str, Any]]:
    yield {
        "type": "user",
        "message": {"role": "user", "content": "run the synthetic task"},
        "parent_tool_use_id": None,
    }


async def allow_tool(
    _tool_name: str, tool_input: dict[str, Any], _context: Any
) -> PermissionResultAllow:
    return PermissionResultAllow(updated_input=tool_input)


async def noop_hook(
    _hook_input: dict[str, Any],
    _tool_use_id: str | None,
    _context: dict[str, Any],
) -> dict[str, Any]:
    return {}


async def main() -> int:
    transport = SyntheticTransport()
    options = ClaudeAgentOptions(
        can_use_tool=allow_tool,
        hooks={"PreToolUse": [HookMatcher(matcher="Read", hooks=[noop_hook])]},
    )
    first_result_seen = asyncio.Event()

    async def collect() -> None:
        async for message in query(
            prompt=prompt_stream(), options=options, transport=transport
        ):
            if isinstance(message, ResultMessage) and not first_result_seen.is_set():
                first_result_seen.set()

    collector = asyncio.create_task(collect())
    await asyncio.wait_for(first_result_seen.wait(), timeout=1)
    for _ in range(10):
        await asyncio.sleep(0)

    closed_before_late_control = transport.closed_for_input
    await transport.send_late_control()
    try:
        await asyncio.wait_for(transport.late_response_delivered.wait(), timeout=0.2)
    except TimeoutError:
        pass

    await transport.finish()
    await asyncio.wait_for(collector, timeout=1)

    print(f"stdin_closed_before_late_control={closed_before_late_control}")
    print(f"late_control_response_delivered={transport.late_response_delivered.is_set()}")
    print(f"late_response_write_after_close={transport.late_write_after_close}")
    print(f"end_input_calls={transport.end_input_calls}")

    healthy = (
        not closed_before_late_control
        and transport.late_response_delivered.is_set()
        and not transport.late_write_after_close
        and transport.end_input_calls == 1
    )
    return 0 if healthy else 1


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))

[starvex]

Independently reproduced on 0.2.132 (current latest) as well — _internal/query.py is byte-identical to 0.2.131, so the residual is present through the latest release. Running the repro.py above against 0.2.132 prints the same four flags and exits 1; the discarded control response surfaces as a real BrokenPipeError("stdin already closed") raised from Query._handle_control_request at both write sites (query.py:528 success-response and :544 error-response), confirming the response cannot cross the already-closed input channel.

Positioning vs nearby open items, to save triage time:

• Distinct from can_use_tool callback never invoked without hooks/MCP servers: stdin closes before permission requests arrive ("Error: Stream closed") #1105 / fix: keep stdin open for can_use_tool permission responses #1106 / fix: keep stdin open for can_use_tool control responses #715 — those address the arming condition (the stdin-holding waiter isn't armed when only can_use_tool is set, so stdin closes at input-stream exhaustion). Here the waiter is already armed (hooks present) and stdin is closed by the settle-before-Result ordering in _track_task_lifecycle, not by the arming gap. Fixing the arming condition does not remove this residual.
• Distinct from Background processes reaped at turn boundary; no re-ping when long-running work completes #1030 — that's background-process reaping / turn-boundary re-ping (process lifecycle), a different subsystem from this control-channel stdin close.

This matches the _track_task_lifecycle docstring, which already names "a task that settles before the turn's result frame leaves the set empty" as outside the current mitigation and points to a CLI run-boundary signal as the real fix.

[cwj2001]

We independently reproduced this residual ordering in a production Agent Service integration, including on the current SDK/CLI release. This matches the synthetic reproduction in this issue and the model-facing failure described in #1088.

Versions

• claude-agent-sdk==0.2.140, Claude Code CLI 2.1.235
• claude-agent-sdk==0.2.147, bundled Claude Code CLI 2.1.250

Production symptom

The parent launches two run_in_background=true native agents. Both children reach terminal state. A completion-triggered parent continuation then tries to update its plan with TaskUpdate, but the tool results are:

The user doesn't want to take this action right now.
STOP what you are doing and wait for the user to tell you how to proceed.

The model follows that tool result and tells the end user that it has paused and is waiting for further instructions.

The overall run is still reported as succeeded, so this presents as a semantic false success: child tasks are complete, but the required final parent aggregation is missing.

Instrumented ordering

We instrumented SubprocessCLITransport.end_input() and recorded message positions. A representative failing run was:

result indices: [206, 331, 378]
end_input at received-message count: 332
post-first-result TaskUpdate calls: 4
STOP-denied TaskUpdate calls: 4

The SDK closes stdin at the intermediate result. The later parent continuation's TaskUpdate control requests occur after that close, and a final result still arrives afterwards.

In other failing traces we observed the same pattern:

result indices: [207, 343, 395]
end_input at received-message count: 344

result indices: [117, 260, 309]
end_input at received-message count: 261

Differential isolation

We reproduced this without our Agent Service by running the Python SDK directly with production-equivalent SDK hooks:

Stop
PreToolUse
PostToolUse
PostToolUseFailure
SubagentStart
SubagentStop

Results:

SDK 0.2.140 + CLI 2.1.235:
  3/4 traced concurrent runs reproduced the STOP denial

SDK 0.2.140 + CLI 2.1.247:
  0/5 reproduced in this sample

SDK 0.2.147 + bundled CLI 2.1.250:
  1/5 reproduced

The latest-SDK failing run again showed the close on an intermediate result:

result indices: [193, 364, 447]
end_input at received-message count: 364
STOP-denied calls: 4

Direct CLI controls using the same CLI, provider, permission mode, and prompt shape did not produce the phantom permission denial:

text input:                       0 STOP denials
stream-json with immediate EOF:  0 STOP denials
stream-json with held stdin:     0 STOP denials

This isolates the user-visible failure to the SDK bidirectional-control lifecycle rather than application prompt handling, static tool permissions, or Agent Service output processing.

Permission/configuration checks

The affected sessions use:

permission_mode=bypassPermissions
no runtime allowed_tools restriction
no runtime disallowed_tools restriction

Earlier TaskCreate and TaskUpdate calls in the same run succeed. The denials begin only after the intermediate-result / stdin-close boundary.

Business impact

• Final parent aggregation is not reliable.
• A run may be structurally succeeded while the requested workflow is semantically incomplete.
• The transport failure is surfaced to the model as a false user permission denial.
• Retrying the same prompt may succeed because the trigger depends on task-terminal / result-frame ordering.
• This is especially confusing in UIs because all child tasks are visibly completed while the parent says it was told to stop.

This appears to be exactly the settle-before-Result residual documented in Query._track_task_lifecycle().

Would Anthropic consider adding an authoritative CLI run-boundary signal, such as system/run_terminal or an explicit final-result marker, so SDKs can distinguish a per-turn Result from a run-ending Result?

We would be happy to contribute or validate a Python SDK regression test once the protocol-level boundary is defined. The deterministic synthetic transport reproduction already attached to this issue looks like the correct unit-level seam; the evidence above adds a real CLI/provider reproduction of the user-visible failure.