anthropics / anthropics/claude-agent-sdk-python

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

Abierto
#1,190 2 comentarios 1 reacción 0 asignados Ver en GitHub
Lenguaje dominante
Python
Estrellas
8.1k
Forks
1.3k
Métricas de merge de PR
Métricas de PR pendientes

Descripción

## 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()))
```

Guía de contribución

No hay ninguna guía de contribución indexada para este repositorio

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.