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

## 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`:

```text
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:

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

## Actual behavior

```text
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

- #1088
- #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

```python
"""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()))
```