diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml index 460b5e7f..d402e038 100644 --- a/.github/workflows/integration.yml +++ b/.github/workflows/integration.yml @@ -33,6 +33,7 @@ on: - lib/claude_agent_sdk/cli_installer.rb - lib/claude_agent_sdk/subprocess_cli_transport.rb - lib/claude_agent_sdk/query.rb + - lib/claude_agent_sdk/query/** - lib/claude_agent_sdk/message_parser.rb - spec/integration/** - .github/workflows/integration.yml diff --git a/CHANGELOG.md b/CHANGELOG.md index 6653b169..cae88f91 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Internal +- The decision of when `query()` may close stdin moved out of `Query` into one private class, `Query::RunLifecycle` (`lib/claude_agent_sdk/query/run_lifecycle.rb`): the per-run end, the task ledger, the session state and the ceiling timer, with the orderings between them. `Query` hands it five events, including frames, and one wait, and supplies the timer. No behavior change; nothing public moved. Its specs now go through that interface (`spec/unit/query/run_lifecycle_spec.rb`) instead of reaching into `Query`'s instance variables. - Hash-form option interpretation moved into one private module (`OptionForms`); no behaviour change. ## [1.2.0] - 2026-10-03 diff --git a/CLAUDE.md b/CLAUDE.md index 062d7845..318857a5 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -26,6 +26,7 @@ User code - Permission callback handling (can_use_tool) - SDK MCP server request routing (tools/call, resources/read, prompts/get) - Message queue (Async::Queue) separating control messages from SDK messages + - When stdin may close: Query::RunLifecycle (lib/claude_agent_sdk/query/run_lifecycle.rb) │ ▼ SubprocessCLITransport (lib/claude_agent_sdk/subprocess_cli_transport.rb) @@ -40,10 +41,18 @@ User code ### Two API Entry Points -- **`query()`** — Simple function interface, run in its own `Async { }.wait`. Like `Client`, it always uses streaming mode with a `Query` handler and the control protocol, so hooks, `can_use_tool` and SDK MCP servers work (an optional `transport:` replaces the default `SubprocessCLITransport`). It writes the prompt (a String, or an Enumerator streamed in the background), then closes stdin. When hooks / `can_use_tool` / SDK MCP servers still need the channel it waits for the end of the run (Python #1279): the CLI's `session_state_changed` `idle` after a result (the transport sets `CLAUDE_CODE_SDK_READS_SESSION_STATE=1`, and `Query` drops the `sdk_host_only` frames that brings), bounded between turns by `CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS` (10 min; a main-thread turn, `requires_action` or a tracked agent stops the clock); from a CLI that sends no state, the first result with no tracked task in flight (the #1088 ledger). A run is a per-run `Query::RunEnd` that `reopen_run` swaps out, so a waiter the ended run woke still closes stdin. The ceiling sleeper is a child of the read task, never a `spawn_task` entry, and every exit path clears it. The difference from `Client` is lifecycle, not protocol: no follow-up queries and no outbound control calls (interrupt, set_model, ...). `ClaudeAgentSDK.ask` is a thin wrapper over `query()` that returns the last `ResultMessage` (raising `CLIConnectionError` if none arrives) and adds no error policy of its own. +- **`query()`** — Simple function interface, run in its own `Async { }.wait`. Like `Client`, it always uses streaming mode with a `Query` handler and the control protocol, so hooks, `can_use_tool` and SDK MCP servers work (an optional `transport:` replaces the default `SubprocessCLITransport`). It writes the prompt (a String, or an Enumerator streamed in the background), then closes stdin. When hooks / `can_use_tool` / SDK MCP servers still need the channel it waits for the end of the run (Python #1279), which `Query::RunLifecycle` decides (see Run Lifecycle below). The difference from `Client` is lifecycle, not protocol: no follow-up queries and no outbound control calls (interrupt, set_model, ...). `ClaudeAgentSDK.ask` is a thin wrapper over `query()` that returns the last `ResultMessage` (raising `CLIConnectionError` if none arrives) and adds no error policy of its own. - **`Client`** — Full-featured bidirectional sessions. Accepts optional `transport_class` (defaults to `SubprocessCLITransport`) and `transport_args` for custom transports. Creates transport, instantiates a `Query` handler that runs `read_messages` in an async task, routes control messages internally, and exposes SDK messages via `Async::Queue`. Supports hooks, permission callbacks, SDK MCP servers, interrupt, model switching, and file rewind. +### Run Lifecycle (when stdin may close) + +`Query::RunLifecycle` (`lib/claude_agent_sdk/query/run_lifecycle.rb`, `@api private`) owns the one decision *when may stdin close?*: the CLI's `session_state_changed` `idle` after a result (the transport sets `CLAUDE_CODE_SDK_READS_SESSION_STATE=1`, and `Query` drops the `sdk_host_only` frames that brings), bounded between turns by `CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS` (10 min; a main-thread turn, `requires_action` or a tracked agent stops the clock); from a CLI that sends no state, the first result with no tracked task in flight (the #1088 ledger). + +`Query` only feeds it: `frame` for every frame the read loop does not route elsewhere (after the mirror flush for a result, before the `sdk_host_only` drop), `message_will_write` (after serialization, before the write), `wait`, `stdin_closing`, `empty_input` (final only — the #1226 trade-off), `reader_gone`. It supplies the timer (`sleeper:`, by default `Query#sleep_on_read_task`: a child of the read task, never a `spawn_task` entry) and three live, non-suspending predicates (`bidirectional_needs:`, `closed:`, `control_requests_in_flight:`). `close_now` never calls into it — it can run off the reactor — and the lifecycle is reactor-only, with no mutex. + +The orderings live inside the module, and its class comment states them: a run is a per-run `RunEnd`, marked ended **before** its waiters are signalled and before the sleeper is stopped (each of those can hand the reactor to an input writer, whose message must then get a run of its own); a ceiling that wakes checks its generation, detaches, then acts, and re-arms instead of ending while a control request is in flight; every exit path clears the sleeper. `spec/unit/query/run_lifecycle_spec.rb` tables them through the interface with a `FakeSleeper` (`spec/support/fake_sleeper.rb`; `Query.new(sleeper:)` takes one too); specs read `query.run_lifecycle` (`ended?`, `final?`, `ceiling_armed?`, `inflight?`, `session_state`), never its instance variables. It is the next TLA+ target (`formal/tla/README.md`). + ### CLI Discovery / Installer `SubprocessCLITransport#find_cli` probes, in order: `CLAUDE_CLI_PATH` (absolutized against the process cwd before use — the CLI is later spawned with `chdir: options.cwd`, where a relative path would name a different file) → `CLIInstaller.installed_path` (project-local vendored binary, deliberately ahead of `PATH` so a pinned install is hermetic) → `claude` on the process `PATH`, searched by the SDK itself (`#executable_on_path`; no `which` — it would be found through `PATH` too, relative entries included, and its answer would have to be believed; with no `PATH` at all, `DEFAULT_DISCOVERY_SEARCH_PATH`, Ruby's default without its `.`) → common install locations. diff --git a/formal/tla/README.md b/formal/tla/README.md index 4ccfe5f7..d601b301 100644 --- a/formal/tla/README.md +++ b/formal/tla/README.md @@ -41,6 +41,17 @@ There are three properties. `EverySenderFinishes` says every control method retu | `bug_nocheck` | `CHECK_SLOT_FIRST = FALSE` | `EverySenderFinishes` violated: the fiber's write suspended after delivery, and the broadcast signal fired before it waited, so the signal was lost | `await_control_response`: "a signal arriving before the sender reaches wait would otherwise be dropped" | | `bug_edge_thread` | `LEVEL_TRIGGERED = FALSE` | `EverySenderFinishes` violated: the sender off the owning reactor checked the slot, the signal fired before it parked, and the signal was lost. A fiber on another thread's reactor was in exactly this position while the waiter was picked by "has an Async task" | `ThreadWaiter`: "closing the check-then-wait gap that an edge-triggered Condition would lose across threads" | +## Next: `RunLifecycle.tla` (not written yet) + +`Query::RunLifecycle` (`lib/claude_agent_sdk/query/run_lifecycle.rb`) decides when `query()` may close stdin: at the CLI's `idle` after a result, at a ceiling between turns, or at the first result with no tracked task in flight. Its fixes so far were all orderings between fibers — ending a run before stopping the ceiling sleeper, re-arming while a control request is in flight, the task ledger — and none of them is model-checked. The decision now sits in one Ruby file behind five events and one wait, so one model can cover it. Until it exists there is no model and no exhaustive check of the interleavings. What runs is RSpec: the lifecycle's table (`spec/unit/query/run_lifecycle_spec.rb`), and through a real `Query`, the end-before-clear ordering on the real scheduler (`spec/unit/query_end_run_order_spec.rb`, which CI also runs on the async 2.10 floor) and, in `spec/unit/query_run_end_spec.rb`, a hook parked under a passing ceiling, the mirror flush gated ahead of a result, and a close that begins while the read loop holds a result (from a reactor task and from another thread). Each covers only the interleavings its examples spell out. + +What the model has to hold: + +- **Per-run identity, not one `ended` flag.** A waiter keeps the run it started waiting on; a run that ended and was reopened still releases it. So: a sequence of runs, the current one, the set of ended ones, and the run each waiter captured. +- **Actions split at the handoffs.** Ending a run is three steps another fiber can run between: mark the run ended, signal its waiters, stop the sleeper. The input writer (`MessageWillWrite`) must be able to interleave with each. The other actions: a result, a session state, a main-thread turn, a task starting or settling, a control request starting or ending, the ceiling waking (armed, and its generation still current), stdin closing, empty input, the reader going away. +- **Switches**, one per rejected alternative: `END_BEFORE_CLEAR` (`FALSE`: the sleeper is stopped before the run is marked ended, and a message written in that gap joins the run that is about to end), `REARM_ON_CONTROL` (`FALSE`: a ceiling that passes under a control request ends the run), `TRACK_LEDGER` (`FALSE`: a result ends the run with a background agent still in flight). +- **Properties.** `NoCloseMidTurn`: stdin never closes during a turn or with a tracked task in flight, unless the reader is gone or the closing waiter's own run had ended. `MessageOwnsRun`: after a message is written, the current run is not ended until a later result, `idle` or the reader's exit — for single-message prompts; from a CLI that reports no state, a queued later message is a documented limitation, not a property. `EventuallyEnds`, under weak fairness and a ceiling above 0, given that timer-resetting work eventually stops (no endless `running` frames, tracked tasks or control requests after a result) and that an interval between turns eventually occurs. + ## What the models assume and leave out - File contents are abstracted to "the version they are the bytes of", so a checksum match is equality. Platform detection is not modelled. diff --git a/lib/claude_agent_sdk/query.rb b/lib/claude_agent_sdk/query.rb index 89f919c6..dac96a65 100644 --- a/lib/claude_agent_sdk/query.rb +++ b/lib/claude_agent_sdk/query.rb @@ -9,6 +9,7 @@ require_relative 'transport' require_relative 'errors' require_relative 'cancellation_signal' +require_relative 'query/run_lifecycle' module ClaudeAgentSDK # Handles bidirectional control protocol on top of Transport @@ -30,37 +31,22 @@ class Query # rubocop:disable Metrics/ClassLength -- control-protocol hub: routi # @api private attr_reader :initialization_result + # When stdin may close (RunLifecycle). Read by specs. + # + # @api private + attr_reader :run_lifecycle + CONTROL_REQUEST_TIMEOUT_ENV_VAR = 'CLAUDE_AGENT_SDK_CONTROL_REQUEST_TIMEOUT_SECONDS' DEFAULT_CONTROL_REQUEST_TIMEOUT_SECONDS = 1200.0 - # Task types whose completion runs a follow-up turn, and which therefore - # may still need the control channel after the turn's result frame. - # - # Mirrors the set the CLI itself holds a result back for, which is - # narrower than its notion of "delegated agent work". The types left out - # are left out on purpose: - # - background shells and monitors run indefinitely by design, so - # deferring the close on one withholds it forever rather than briefly; - # - teammates are long-lived too — their status stays running for their - # whole lifetime, so they never settle the ledger; - # - remote agents can be long-running monitors the CLI likewise refuses - # to wait on. - # Anything added here must be a type that reliably reaches a terminal - # status, or it will hang the query (see #track_task_lifecycle). - DEFERRING_TASK_TYPES = %w[local_agent local_workflow].freeze - # The CLI's own wait for background work once stdin is closed; the SDK # bounds its wait for the CLI's "idle" by the same value - # (#arm_run_end_ceiling, Python #1279). + # (RunLifecycle#arm, Python #1279). RUN_END_CEILING_ENV_VAR = 'CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS' DEFAULT_RUN_END_CEILING_MS = 600_000 # The longest ceiling honored (~24.8 days), as in the TypeScript and # Python SDKs, whose timers cannot run longer. MAX_RUN_END_CEILING_MS = (2**31) - 1 - # Frame types that mark a main-thread turn under way (when they carry no - # parent_tool_use_id), and the states that do not re-arm the ceiling. - TURN_FRAME_TYPES = %w[assistant stream_event].freeze - NON_RUNNING_SESSION_STATES = %w[idle requires_action].freeze # Read CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS from where the CLI gets it: # +options_env+ (ClaudeAgentOptions#env) overrides the inherited @@ -76,33 +62,6 @@ def self.run_end_ceiling_ms(options_env) digits.match?(/\A\d+\z/) ? Integer(digits, 10) : DEFAULT_RUN_END_CEILING_MS end - # One run's end, set at most once (the Python SDK's per-run anyio.Event). - # A waiter holds the object it started waiting on, so a run that ends and - # is then reopened (#reopen_run swaps in a fresh RunEnd) still releases the - # waiters the ended run woke, while later waits wait for the new run. - # Reactor-only, like every caller. - class RunEnd - def initialize - @ended = false - @condition = Async::Condition.new - end - - def ended? - @ended - end - - def end! - return if @ended - - @ended = true - @condition.signal - end - - def wait - @condition.wait until @ended - end - end - # Apply ClaudeAgentOptions#verbatim_prompts to one outgoing user message # (Python #1269's stamp_user_message). Off: returns +message+ unchanged. # On: returns a new Hash with `client_composed: true`, dropping any @@ -164,7 +123,7 @@ def initialize(transport:, is_streaming_mode:, can_use_tool: nil, hooks: nil, sd exclude_dynamic_sections: nil, system_prompt_snapshot: nil, skills: nil, forward_subagent_text: false, agent_progress_summaries: nil, callback_scheduling: :thread, callback_wrapper: nil, verbatim_prompts: false, - run_end_ceiling_ms: DEFAULT_RUN_END_CEILING_MS) + run_end_ceiling_ms: DEFAULT_RUN_END_CEILING_MS, sleeper: nil) @transport = transport @is_streaming_mode = is_streaming_mode @can_use_tool = can_use_tool @@ -179,7 +138,6 @@ def initialize(transport:, is_streaming_mode:, can_use_tool: nil, hooks: nil, sd @forward_subagent_text = forward_subagent_text @agent_progress_summaries = agent_progress_summaries @verbatim_prompts = verbatim_prompts - @run_end_ceiling_ms = run_end_ceiling_ms # Control protocol state @pending_control_responses = {} @@ -197,36 +155,14 @@ def initialize(transport:, is_streaming_mode:, can_use_tool: nil, hooks: nil, sd @message_queue = Async::Queue.new # Set once #receive_messages has consumed the read loop's end sentinel. @stream_ended = false - # Ends when the run is over, so the stdin-closing waiter can wake; see - # #read_messages and @inflight_tasks below (Python #1088, #1190/#1279). - # Work the CLI takes up after the run ended swaps in a fresh one - # (#reopen_run). - @run_end = RunEnd.new - @result_received = false - # The CLI's latest session_state_changed state, or nil while it sends - # none (a CLI too old to honor CLAUDE_CODE_SDK_READS_SESSION_STATE). A - # CLI that reports state stays "running" while a background agent is - # live or its completion is still to be handled, and reports "idle" once - # no further turn is owed. - @session_state = nil - # Ends the run if no new turn starts within the ceiling after a result - # (#arm_run_end_ceiling). The generation tells a sleeper that woke after - # it was cleared or re-armed to stand down. - @run_end_ceiling_task = nil - @run_end_ceiling_generation = 0 - # A main-thread turn is under way (its assistant/stream_event frames - # have started and its result has not arrived): the ceiling counts only - # the wait between turns, so it is not armed meanwhile. - @turn_in_progress = false - # Set once stdin is closed or the reader is gone: the run then stays - # ended, since nothing can wait on a reopened one. - @run_final = false - # Task IDs of started-but-not-finished deferring tasks. A result frame - # only ends one turn, not the run: a background task keeps running past - # it and still needs stdin for hook/SDK-MCP control responses (Python - # #1088/#1103), so a result that arrives while this set is non-empty - # must not close stdin. - @inflight_tasks = Set.new + # When stdin may close: the end of the run (Python #1088, #1190/#1279). + # +sleeper+ replaces the ceiling's timer in specs; the default reads @task + # when it is called, which is after #start. + @run_lifecycle = RunLifecycle.new( + ceiling_ms: run_end_ceiling_ms, sleeper: sleeper || method(:sleep_on_read_task), + bidirectional_needs: -> { bidirectional_needs? }, closed: -> { @closed }, + control_requests_in_flight: -> { !@inflight_control_request_tasks.empty? } + ) # Set to the result payload when the most recent message is a result # with is_error=true. Used to replace the generic "exit code 1" # ProcessError with a ResultError carrying what the CLI already @@ -488,43 +424,27 @@ def read_messages # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexit @transcript_mirror_batcher&.enqueue(message[:filePath] || message[:file_path], message[:entries] || []) next else - # Track task lifecycle frames so results can tell "one turn ended" - # apart from "the run is done" (Python #1088/#1103). - if msg_type == 'system' - had_tasks_in_flight = !@inflight_tasks.empty? - track_task_lifecycle(message) - # The ceiling left the last tracked agent alone; the wait between - # turns starts over now that it settled. - rearm_run_end_ceiling_between_turns if had_tasks_in_flight && @inflight_tasks.empty? - if message[:subtype] == 'session_state_changed' - on_session_state(message[:state]) - # Frames the CLI sent only because the transport asked for them - # (CLAUDE_CODE_SDK_READS_SESSION_STATE); the caller did not opt - # in, so they never reach the stream, observers or the parser. - next if message[:sdk_host_only] == true - end - end + session_state_frame = msg_type == 'system' && message[:subtype] == 'session_state_changed' + # Flush the mirror before signaling/yielding the result so a + # consumer observing the result sees an up-to-date store for the turn. + flush_transcript_mirror if msg_type == 'result' + # Every frame that gets here drives the run's end (task lifecycle, + # session state, results, main-thread turns), whether or not it goes + # on to the stream. + @run_lifecycle.frame(message) + # Frames the CLI sent only because the transport asked for them + # (CLAUDE_CODE_SDK_READS_SESSION_STATE); the caller did not opt in, + # so they never reach the stream, observers or the parser. + next if session_state_frame && message[:sdk_host_only] == true if msg_type == 'result' - # Flush the mirror before signaling/yielding the result so a - # consumer observing the result sees an up-to-date store for the turn. - flush_transcript_mirror - on_result @last_error_result = message[:is_error] ? message : nil - elsif !(msg_type == 'system' && message[:subtype] == 'session_state_changed') + elsif !session_state_frame # Anything other than the post-turn session_state_changed marker # means the conversation moved on; a ProcessError now is a fresh # crash, not the expected exit from a prior error result. Mirrors # the Python/TypeScript SDK reset logic. @last_error_result = nil - # A main-thread turn is under way, so the ceiling stops (it counts - # only the wait between turns, as the CLI's does) and the run - # reopens even if the ceiling ended it while no state changed. - if TURN_FRAME_TYPES.include?(msg_type) && message[:parent_tool_use_id].nil? - @turn_in_progress = true - reopen_run - clear_run_end_ceiling - end end # Regular SDK messages go to the queue @message_queue.enqueue(message) @@ -576,189 +496,12 @@ def read_messages # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexit # Unblock the stdin-closing waiter so it doesn't stall on early exit; # with the reader gone the run stays ended. Also stops a pending # ceiling sleeper, which would otherwise keep the reactor alive. - @run_final = true - end_run + @run_lifecycle.reader_gone # Always signal end of stream @message_queue.enqueue({ type: 'end' }) end end - # Track in-flight tasks from `system` task lifecycle frames. - # - # `task_started` marks a task in flight; `task_notification` or a - # `task_updated` patch with a terminal status clears it. Terminal - # completion can arrive as either frame (not every terminal task emits a - # notification), so both are handled; Set deletion keeps the pair - # idempotent. - # - # This is a mitigation, not a complete answer to Python #1088. An empty - # set means "nothing we know of is running", which is not the same as - # "the run is over": a task that settles *before* the turn's result frame - # leaves the set empty at that result, so stdin closes even though the - # completion may still wake the parent for a continuation turn. What this - # does fix is the common ordering, where the task outlives the turn that - # spawned it. - # - # Only delegated agent work is tracked (DEFERRING_TASK_TYPES). A - # background *shell* is also reported through these frames, but it may - # never reach a terminal status, and the CLI in stream-json mode only - # exits on stdin EOF — tracking one would withhold the close forever. - # - # `background_tasks_changed` is deliberately not consumed, in either - # direction: its payload is the live *background* set, while a subagent - # is registered in the foreground and only flips to backgrounded later - # without a second task_started, so narrowing against the snapshot would - # drop an agent that goes on to outlive its turn, and widening from it - # could admit an id no later frame ever clears (observer agents suppress - # both their start and terminal frames). - def track_task_lifecycle(message) - task_id = message[:task_id] - return if task_id.nil? || task_id.to_s.empty? - - case message[:subtype] - when 'task_started' - @inflight_tasks.add(task_id) if DEFERRING_TASK_TYPES.include?(message[:task_type]) - when 'task_notification' - @inflight_tasks.delete(task_id) - when 'task_updated' - patch = message[:patch] - status = patch.is_a?(Hash) ? patch[:status] : nil - @inflight_tasks.delete(task_id) if TERMINAL_TASK_STATUSES.include?(status) - end - end - - # A result ends a turn, not necessarily the run: a background agent that - # finished just before it still wakes the session for another turn, whose - # hook, permission and SDK MCP requests need stdin (Python #1190/#1279). A - # CLI that reports session state stays "running" while such a turn is - # owed, so wait for "idle" (some hosts send it just before the result). - # Without state events the result is all there is to go on. - def on_result - @result_received = true - @turn_in_progress = false - if @session_state.nil? || @session_state == 'idle' || !bidirectional_needs? - maybe_end_run - elsif @session_state != 'requires_action' - # While the SDK is still answering a request the ceiling waits for - # the "running" that follows. - arm_run_end_ceiling - end - end - - # Track the CLI's session_state_changed state (Python #1279). - def on_session_state(state) - @session_state = state - if state == 'idle' - maybe_end_run if @result_received - return - end - # Work the CLI took up after the run ended (a finished background task - # woke it) reopens the run until the next "idle". - reopen_run - if state == 'requires_action' - # The host is answering a request; stdin must outlast it. - clear_run_end_ceiling - else - rearm_run_end_ceiling_between_turns - end - end - - # End the run unless a tracked background task is still in flight: such - # a task may still need hook/SDK-MCP control responses over stdin - # (Python #1088), and its completion wakes the parent for a follow-up - # turn whose result (or "idle") ends the run then. A CLI that reports - # session state never reports "idle" with an agent still live, so this - # matters for CLIs that report "idle" at every turn end or not at all. - def maybe_end_run - end_run if @inflight_tasks.empty? - end - - # The run is over: wake the stdin-closing waiter. Idempotent. - # - # The run is marked ended BEFORE the ceiling is cleared: stopping the - # sleeper task yields to whatever else is ready, and a #stream_input task - # that writes its next message in that gap must find the run already - # ended, so that #reopen_run gives the message a run of its own. Cleared - # first, the message joined the run that was about to end, and stdin - # closed before the message's own run had produced a frame. - def end_run - @run_end.end! - clear_run_end_ceiling - end - - # Reopen an ended run for work that started after it ended. A waiter the - # ended run already woke still closes stdin; this makes a later wait - # (stream_input's, once its prompts are all written) wait for the new - # work too. Once stdin is closed, or the reader is gone, the run stays - # ended. - def reopen_run - @run_end = RunEnd.new if @run_end.ended? && !@run_final - end - - # End the run anyway once the ceiling passes with no new turn. The CLI's - # own background-wait ceiling only counts once stdin is closed, so without - # this, work that never finishes would hold "running", and stdin, open - # forever. It counts only the wait between turns: restarted at each result - # and whenever the CLI reports "running" again, cleared by main-thread - # turn activity and by "requires_action", never armed mid-turn. - # - # The sleeper is a child of the read task, NOT a #spawn_task entry: it is re-armed at every - # result and every "running", and @child_tasks never prunes. The read - # task's stop cascades to it, and every exit path clears it (#end_run in - # the read loop's ensure, #wait_for_result_and_end_input's ensure), so a - # pending sleeper can never keep the enclosing reactor alive. - def arm_run_end_ceiling - clear_run_end_ceiling - # A frame read while close is under way (the result branch flushes the - # mirror first) must not leave a sleeper behind either. - return if @run_end_ceiling_ms <= 0 || @run_end.ended? || @run_final || @closed || - @turn_in_progress || !bidirectional_needs? - - generation = @run_end_ceiling_generation - seconds = [@run_end_ceiling_ms, MAX_RUN_END_CEILING_MS].min / 1000.0 - @run_end_ceiling_task = @task.async do - sleep seconds - end_run_at_ceiling(generation) - end - end - - # Restart the ceiling if the run is between turns, past a result, with - # the CLI still reporting work ("running"). - def rearm_run_end_ceiling_between_turns - return unless @result_received - return if @session_state.nil? || NON_RUNNING_SESSION_STATES.include?(@session_state) - - arm_run_end_ceiling - end - - def end_run_at_ceiling(generation) - # Cleared or re-armed while this sleeper was already waking up. - return unless generation == @run_end_ceiling_generation - - # Detach before ending the run: #end_run clears the ceiling, and - # clearing it must not stop the task running this very method. - @run_end_ceiling_task = nil - # A tracked background agent still running may still need stdin for its - # hook, permission and SDK MCP requests (Python #1088), so it is not cut - # off; the ceiling starts over once it settles (#read_messages). - return unless @inflight_tasks.empty? - - # A control request the SDK is still answering (a slow hook or SDK MCP - # tool) must be able to write its reply, so the clock starts over rather - # than closing stdin under it. Ruby-only guard: Python relies on the CLI - # reporting requires_action for every such request. - return arm_run_end_ceiling unless @inflight_control_request_tasks.empty? - - end_run - end - - def clear_run_end_ceiling - @run_end_ceiling_generation += 1 - task = @run_end_ceiling_task - @run_end_ceiling_task = nil - task&.stop - end - # Whether the CLI may still send control requests that need a reply. # # SDK MCP servers, hooks and the can_use_tool permission callback are all @@ -777,6 +520,17 @@ def bidirectional_needs? !@sdk_mcp_servers.empty? || !@hooks.empty? || !!@can_use_tool end + # The run-end ceiling's timer (RunLifecycle's sleeper): runs +on_wake+ once + # +seconds+ have passed. A child of the read task, NOT a #spawn_task entry: + # it is re-armed at every result and every "running", and @child_tasks + # never prunes. The read task's stop cascades to it. + def sleep_on_read_task(seconds, &on_wake) + @task.async do + sleep seconds + on_wake.call + end + end + # Flush the transcript-mirror batcher, swallowing errors — a mirror failure # must never propagate into the read loop or its teardown. def flush_transcript_mirror @@ -1744,7 +1498,7 @@ def rewind_files(user_message_uuid) # wait BETWEEN turns is bounded: if the CLI still reports "running" # CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS (10 minutes by default, 0 for no # limit) after a result with no new turn, the run ends anyway - # (#arm_run_end_ceiling). A turn under way, a request the SDK is still + # (RunLifecycle#arm). A turn under way, a request the SDK is still # answering and a tracked background agent still in flight (Python #1088) # stop that clock. The run is guaranteed to end: as above, or in # read_messages' ensure when the process exits early. @@ -1755,10 +1509,9 @@ def rewind_files(user_message_uuid) # control requests from that later turn can find stdin closed. # Single-message and String prompts are fully covered. def wait_for_result_and_end_input - @run_end.wait if bidirectional_needs? + @run_lifecycle.wait ensure - @run_final = true - clear_run_end_ceiling + @run_lifecycle.stdin_closing @transport.end_input end @@ -1781,9 +1534,7 @@ def stream_input(stream) line = Query.serialize_user_message(message, @verbatim_prompts) # This message owes a run of its own, result included: an earlier # one having ended does not end it. - reopen_run - @result_received = false - clear_run_end_ceiling + @run_lifecycle.message_will_write writeln(line) wrote_message = true end @@ -1819,7 +1570,7 @@ def stream_input(stream) if wrote_message wait_for_result_and_end_input else - @run_final = true + @run_lifecycle.empty_input @transport.end_input end end diff --git a/lib/claude_agent_sdk/query/run_lifecycle.rb b/lib/claude_agent_sdk/query/run_lifecycle.rb new file mode 100644 index 00000000..679b5d42 --- /dev/null +++ b/lib/claude_agent_sdk/query/run_lifecycle.rb @@ -0,0 +1,373 @@ +# frozen_string_literal: true + +require 'async/condition' + +module ClaudeAgentSDK + class Query + # When stdin may close: one run's end, decided from the CLI's frames and one + # timer (the ceiling). Reactor-only, like every caller: no mutex, and never + # reached from a FiberBoundary thread or from Query#close_now. An event may + # hand execution to another fiber mid-call, at RunEnd#end!'s signal, the + # sleeper's async, a handle's #stop and #wait. So the order is the contract: + # #end_run marks the run ended BEFORE it signals or stops the sleeper (a + # message written in that gap gets a run of its own); a ceiling that wakes + # checks its generation, detaches, then acts, and re-arms instead of ending + # while the SDK still answers a control request. The sleeper is a child of + # the read task and every exit path clears it (#reader_gone, #stdin_closing). + # + # @api private + class RunLifecycle + # Task types whose completion runs a follow-up turn, and which therefore + # may still need the control channel after the turn's result frame. + # + # Mirrors the set the CLI itself holds a result back for, which is + # narrower than its notion of "delegated agent work". The types left out + # are left out on purpose: + # - background shells and monitors run indefinitely by design, so + # deferring the close on one withholds it forever rather than briefly; + # - teammates are long-lived too — their status stays running for their + # whole lifetime, so they never settle the ledger; + # - remote agents can be long-running monitors the CLI likewise refuses + # to wait on. + # Anything added here must be a type that reliably reaches a terminal + # status, or it will hang the query (see #track_task_lifecycle). + DEFERRING_TASK_TYPES = %w[local_agent local_workflow].freeze + + # Frame types that mark a main-thread turn under way (when they carry no + # parent_tool_use_id), and the states that do not re-arm the ceiling. + TURN_FRAME_TYPES = %w[assistant stream_event].freeze + NON_RUNNING_SESSION_STATES = %w[idle requires_action].freeze + + # One run's end, set at most once (the Python SDK's per-run anyio.Event). + # A waiter holds the object it started waiting on, so a run that ends and + # is then reopened (#reopen_run swaps in a fresh RunEnd) still releases the + # waiters the ended run woke, while later waits wait for the new run. + # Reactor-only, like every caller. + class RunEnd + def initialize + @ended = false + @condition = Async::Condition.new + end + + def ended? + @ended + end + + def end! + return if @ended + + @ended = true + @condition.signal + end + + def wait + @condition.wait until @ended + end + end + + # The CLI's latest session_state_changed state, or nil while it sends + # none (a CLI too old to honor CLAUDE_CODE_SDK_READS_SESSION_STATE). A + # CLI that reports state stays "running" while a background agent is + # live or its completion is still to be handled, and reports "idle" once + # no further turn is owed. + attr_reader :session_state + + # +ceiling_ms+ bounds the wait between turns (0 = never armed; see #arm). + # +sleeper+ is called as `sleeper.call(seconds) { ... }`: it runs the block + # once +seconds+ have passed and returns a handle that responds to #stop + # (Query#sleep_on_read_task in production). The three predicates read + # Query's facts when they are called and must not suspend. + def initialize(ceiling_ms:, sleeper:, bidirectional_needs:, closed:, control_requests_in_flight:) + @ceiling_ms = ceiling_ms + @sleeper = sleeper + @bidirectional_needs = bidirectional_needs + @closed = closed + @control_requests_in_flight = control_requests_in_flight + # Ends when the run is over, so the stdin-closing waiter can wake (Python + # #1088, #1190/#1279). Work the CLI takes up after the run ended swaps in + # a fresh one (#reopen_run). + @run_end = RunEnd.new + @result_received = false + @session_state = nil + # Ends the run if no new turn starts within the ceiling after a result + # (#arm). The generation tells a sleeper that woke after it was cleared + # or re-armed to stand down; it only ever grows. + @ceiling_handle = nil + @ceiling_generation = 0 + # A main-thread turn is under way (its assistant/stream_event frames + # have started and its result has not arrived): the ceiling counts only + # the wait between turns, so it is not armed meanwhile. + @turn_in_progress = false + # Set once stdin is closed or the reader is gone: the run then stays + # ended, since nothing can wait on a reopened one. + @final = false + # Task IDs of started-but-not-finished deferring tasks. A result frame + # only ends one turn, not the run: a background task keeps running past + # it and still needs stdin for hook/SDK-MCP control responses (Python + # #1088/#1103), so a result that arrives while this set is non-empty + # must not close stdin. + @inflight_tasks = Set.new + end + + # One frame the read loop did not route elsewhere (anything but control + # and transcript_mirror frames), frames marked sdk_host_only included; + # for a result, after the mirror flush. + def frame(message) + type = message[:type] + if type == 'system' + on_system(message) + elsif type == 'result' + on_result + elsif TURN_FRAME_TYPES.include?(type) && message[:parent_tool_use_id].nil? + on_main_thread_turn + end + end + + # A user message was serialized and is about to be written. It owes a run + # of its own, result included: an earlier one having ended does not end it. + def message_will_write + reopen_run + @result_received = false + clear + end + + # Wait for the end of the run that is current now, when the CLI may still + # send control requests that need a reply; return at once otherwise. + def wait + @run_end.wait if @bidirectional_needs.call + end + + # Stdin is about to close after a wait: no run can be reopened any more. + def stdin_closing + @final = true + clear + end + + # Stdin is about to close with nothing written: final, but an armed + # sleeper is left for the read task's stop and the run is not ended + # (the #1226 trade-off; see Query#stream_input). + def empty_input + @final = true + end + + # The read loop is over: wake the stdin-closing waiter so it does not + # stall on early exit; with the reader gone the run stays ended. Also + # stops a pending sleeper, which would otherwise keep the reactor alive. + def reader_gone + @final = true + end_run + end + + def ended? + @run_end.ended? + end + + def final? + @final + end + + def ceiling_armed? + !@ceiling_handle.nil? + end + + def inflight?(task_id) + @inflight_tasks.include?(task_id) + end + + private + + # Track task lifecycle frames so results can tell "one turn ended" apart + # from "the run is done" (Python #1088/#1103), then the session state. + def on_system(message) + had_tasks_in_flight = !@inflight_tasks.empty? + track_task_lifecycle(message) + # The ceiling left the last tracked agent alone; the wait between + # turns starts over now that it settled. + rearm_between_turns if had_tasks_in_flight && @inflight_tasks.empty? + on_session_state(message[:state]) if message[:subtype] == 'session_state_changed' + end + + # A main-thread turn is under way, so the ceiling stops (it counts only + # the wait between turns, as the CLI's does) and the run reopens even if + # the ceiling ended it while no state changed. + def on_main_thread_turn + @turn_in_progress = true + reopen_run + clear + end + + # Track in-flight tasks from `system` task lifecycle frames. + # + # `task_started` marks a task in flight; `task_notification` or a + # `task_updated` patch with a terminal status clears it. Terminal + # completion can arrive as either frame (not every terminal task emits a + # notification), so both are handled; Set deletion keeps the pair + # idempotent. + # + # This is a mitigation, not a complete answer to Python #1088. An empty + # set means "nothing we know of is running", which is not the same as + # "the run is over": a task that settles *before* the turn's result frame + # leaves the set empty at that result, so stdin closes even though the + # completion may still wake the parent for a continuation turn. What this + # does fix is the common ordering, where the task outlives the turn that + # spawned it. + # + # Only delegated agent work is tracked (DEFERRING_TASK_TYPES). A + # background *shell* is also reported through these frames, but it may + # never reach a terminal status, and the CLI in stream-json mode only + # exits on stdin EOF — tracking one would withhold the close forever. + # + # `background_tasks_changed` is deliberately not consumed, in either + # direction: its payload is the live *background* set, while a subagent + # is registered in the foreground and only flips to backgrounded later + # without a second task_started, so narrowing against the snapshot would + # drop an agent that goes on to outlive its turn, and widening from it + # could admit an id no later frame ever clears (observer agents suppress + # both their start and terminal frames). + def track_task_lifecycle(message) + task_id = message[:task_id] + return if task_id.nil? || task_id.to_s.empty? + + case message[:subtype] + when 'task_started' + @inflight_tasks.add(task_id) if DEFERRING_TASK_TYPES.include?(message[:task_type]) + when 'task_notification' + @inflight_tasks.delete(task_id) + when 'task_updated' + patch = message[:patch] + status = patch.is_a?(Hash) ? patch[:status] : nil + @inflight_tasks.delete(task_id) if TERMINAL_TASK_STATUSES.include?(status) + end + end + + # A result ends a turn, not necessarily the run: a background agent that + # finished just before it still wakes the session for another turn, whose + # hook, permission and SDK MCP requests need stdin (Python #1190/#1279). A + # CLI that reports session state stays "running" while such a turn is + # owed, so wait for "idle" (some hosts send it just before the result). + # Without state events the result is all there is to go on. + def on_result + @result_received = true + @turn_in_progress = false + if @session_state.nil? || @session_state == 'idle' || !@bidirectional_needs.call + maybe_end_run + elsif @session_state != 'requires_action' + # While the SDK is still answering a request the ceiling waits for + # the "running" that follows. + arm + end + end + + # Track the CLI's session_state_changed state (Python #1279). + def on_session_state(state) + @session_state = state + if state == 'idle' + maybe_end_run if @result_received + return + end + # Work the CLI took up after the run ended (a finished background task + # woke it) reopens the run until the next "idle". + reopen_run + if state == 'requires_action' + # The host is answering a request; stdin must outlast it. + clear + else + rearm_between_turns + end + end + + # End the run unless a tracked background task is still in flight: such + # a task may still need hook/SDK-MCP control responses over stdin + # (Python #1088), and its completion wakes the parent for a follow-up + # turn whose result (or "idle") ends the run then. A CLI that reports + # session state never reports "idle" with an agent still live, so this + # matters for CLIs that report "idle" at every turn end or not at all. + def maybe_end_run + end_run if @inflight_tasks.empty? + end + + # The run is over: wake the stdin-closing waiter. Idempotent. + # + # The run is marked ended BEFORE the ceiling is cleared: stopping the + # sleeper yields to whatever else is ready, and a Query#stream_input task + # that writes its next message in that gap must find the run already + # ended, so that #reopen_run gives the message a run of its own. Cleared + # first, the message joined the run that was about to end, and stdin + # closed before the message's own run had produced a frame. + def end_run + @run_end.end! + clear + end + + # Reopen an ended run for work that started after it ended. A waiter the + # ended run already woke still closes stdin; this makes a later wait + # (Query#stream_input's, once its prompts are all written) wait for the + # new work too. Once stdin is closed, or the reader is gone, the run + # stays ended. + def reopen_run + @run_end = RunEnd.new if @run_end.ended? && !@final + end + + # End the run anyway once the ceiling passes with no new turn. The CLI's + # own background-wait ceiling only counts once stdin is closed, so without + # this, work that never finishes would hold "running", and stdin, open + # forever. It counts only the wait between turns: restarted at each result + # and whenever the CLI reports "running" again, cleared by main-thread + # turn activity and by "requires_action", never armed mid-turn. + # + # Every exit path clears the sleeper (#end_run from #reader_gone, + # #stdin_closing), so a pending one can never keep the enclosing reactor + # alive; the production sleeper is a child of the read task besides + # (Query#sleep_on_read_task). + def arm + clear + # A frame read while close is under way (the read loop flushes the + # mirror before a result gets here) must not leave a sleeper behind + # either. + return if @ceiling_ms <= 0 || @run_end.ended? || @final || @closed.call || + @turn_in_progress || !@bidirectional_needs.call + + generation = @ceiling_generation + seconds = [@ceiling_ms, MAX_RUN_END_CEILING_MS].min / 1000.0 + @ceiling_handle = @sleeper.call(seconds) { ceiling_passed(generation) } + end + + # Restart the ceiling if the run is between turns, past a result, with + # the CLI still reporting work ("running"). + def rearm_between_turns + return unless @result_received + return if @session_state.nil? || NON_RUNNING_SESSION_STATES.include?(@session_state) + + arm + end + + def ceiling_passed(generation) + # Cleared or re-armed while this sleeper was already waking up. + return unless generation == @ceiling_generation + + # Detach before ending the run: #end_run clears the ceiling, and + # clearing it must not stop the task running this very method. + @ceiling_handle = nil + # A tracked background agent still running may still need stdin for its + # hook, permission and SDK MCP requests (Python #1088), so it is not cut + # off; the ceiling starts over once it settles (#on_system). + return unless @inflight_tasks.empty? + + # A control request the SDK is still answering (a slow hook or SDK MCP + # tool) must be able to write its reply, so the clock starts over rather + # than closing stdin under it. Ruby-only guard: Python relies on the CLI + # reporting requires_action for every such request. + return arm if @control_requests_in_flight.call + + end_run + end + + def clear + @ceiling_generation += 1 + handle = @ceiling_handle + @ceiling_handle = nil + handle&.stop + end + end + end +end diff --git a/spec/support/fake_sleeper.rb b/spec/support/fake_sleeper.rb new file mode 100644 index 00000000..3258c081 --- /dev/null +++ b/spec/support/fake_sleeper.rb @@ -0,0 +1,100 @@ +# frozen_string_literal: true + +# The run-end ceiling's timer, without a clock. Query::RunLifecycle takes a +# sleeper — `sleeper.call(seconds) { ... }`, returning a handle that responds +# to #stop — and this one never wakes by itself: it keeps every arm, and the +# example decides which one wakes and when. +# +# sleeper = FakeSleeper.new +# query = ClaudeAgentSDK::Query.new(transport: transport, is_streaming_mode: true, sleeper: sleeper) +# ... +# sleeper.fire # the sleep of the arm that is pending is over +# sleeper.fire_stale # an arm that was stopped or replaced wakes anyway: it was already waking up +# +# What the lifecycle believes is read from the lifecycle (#ceiling_armed?); +# what it did to its sleeper is read here (#arms, #stops, #pending_arms). +# A lifecycle that keeps track of its sleeper has at most one arm pending, +# the one it holds: an earlier arm it replaced without stopping stays in +# #pending_arms, where an example can see it, and makes #fire refuse. +class FakeSleeper + # One call of the sleeper: the handle the lifecycle holds. + class Arm + attr_reader :seconds + + def initialize(sleeper, seconds, on_wake) + @sleeper = sleeper + @seconds = seconds + @on_wake = on_wake + @stopped = false + @woken = false + end + + def stop + @stopped = true + @sleeper.stopped + end + + # Neither stopped nor woken: the sleep is still running. + def pending? + !@stopped && !@woken + end + + def wake + @woken = true + @on_wake.call + end + end + + # Every arm so far, oldest first, and how many times a handle was stopped. + attr_reader :arms, :stops + + # Called inside every Arm#stop, for an example that needs to see the state a + # stop finds (a real stop hands the reactor to whatever else is ready). + attr_accessor :on_stop + + def initialize + @arms = [] + @stops = 0 + @on_stop = nil + end + + def call(seconds, &on_wake) + arm = Arm.new(self, seconds, on_wake) + @arms << arm + arm + end + + def stopped + @stops += 1 + @on_stop&.call + end + + # Every arm that is still sleeping, oldest first: one at most, unless a + # sleeper was replaced without being stopped. + def pending_arms + @arms.select(&:pending?) + end + + # Some arm is still sleeping. + def armed? + !pending_arms.empty? + end + + # The pending arm's sleep is over. + def fire + pending = pending_arms + raise "FakeSleeper#fire: #{pending.length} arms are pending, not one" unless pending.length == 1 + + pending.first.wake + end + + # The latest arm that is no longer pending (stopped, or already woken) + # wakes: the sleeper it stood for was past its sleep when it was stopped or + # replaced, and still runs its block. + def fire_stale + arm = @arms.reject(&:pending?).last + raise 'FakeSleeper#fire_stale: every arm is still pending' unless arm + + arm.wake + end +end diff --git a/spec/unit/query/run_lifecycle_spec.rb b/spec/unit/query/run_lifecycle_spec.rb new file mode 100644 index 00000000..a03032cb --- /dev/null +++ b/spec/unit/query/run_lifecycle_spec.rb @@ -0,0 +1,613 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'async' + +# When query() may close stdin, as a table: each example is a sequence of +# events through Query::RunLifecycle's interface and what it must leave +# behind. No Query, no transport and no clock take part: the lifecycle gets a +# FakeSleeper, which the example wakes by hand, and three facts it would read +# from its Query. Only the #wait rows need a reactor. +# +# The groups are named after the behaviours the lifecycle has to keep: +# +# B1 with a control channel in use, the run ends at the CLI's "idle" after +# a result (Python #1279); "idle" before any result ends nothing +# B2 a CLI that sends no session state ends the run at the first result +# with no tracked task in flight (#1088); only local_agent and +# local_workflow tasks are tracked +# B3 neither a result nor "idle" ends the run while a tracked task is in +# flight +# B4 the ceiling: armed at a result, and at each "running" after one, while +# the CLI still reports work; never before the first result, mid-turn, +# once the run ended, is final or the Query closed, or with no control +# channel in use; 0 means no ceiling, and the sleep is capped +# B5 a main-thread turn stops the clock and reopens the run, a subagent's +# frames do not; "requires_action" stops it and the "running" that +# follows restarts it; a ceiling that passes with a tracked agent in +# flight stands down, and the clock starts over once the agent settles +# B6 a ceiling that passes while the SDK is still answering a control +# request re-arms instead of ending the run +# B7 each message written owes a run of its own +# B9 an ended run is reopened as a fresh one, which the waiters of the +# ended run do not wait for; a final run is never reopened, and being +# final does not end it +# B10 the run is marked ended before the sleeper is stopped +# B11 every exit path clears the sleeper +# B14 empty input is final only: nothing is stopped, nothing is ended +# Q5 a wake from an earlier arm never touches a later one +# +# The wiring of a Query to its lifecycle (which frames reach it and in what +# order with the mirror flush and the stream, the real sleeper, the real +# predicates) is covered through Query in query_run_end_spec.rb and +# query_end_run_order_spec.rb. +RSpec.describe ClaudeAgentSDK::Query::RunLifecycle do + let(:sleeper) { FakeSleeper.new } + # What the lifecycle reads from its Query, each time it asks. + let(:facts) { { bidirectional_needs: true, closed: false, control_requests_in_flight: false } } + let(:lifecycle) do |example| + described_class.new(ceiling_ms: example.metadata.fetch(:ceiling_ms, 600_000), sleeper: sleeper, + bidirectional_needs: -> { facts[:bidirectional_needs] }, + closed: -> { facts[:closed] }, + control_requests_in_flight: -> { facts[:control_requests_in_flight] }) + end + + # Once an event is over, the lifecycle holds a sleeper exactly when one is + # still sleeping, and no other sleeps: it never loses track of one (an arm + # replaced without being stopped stays pending), and never keeps one that + # woke or was stopped. + after { expect(sleeper.pending_arms.length).to eq(lifecycle.ceiling_armed? ? 1 : 0) } + + def state(value) + { type: 'system', subtype: 'session_state_changed', state: value, session_id: 's', sdk_host_only: true } + end + + def result + sample_result_message + end + + def turn_frame(type = 'assistant', parent_tool_use_id: nil) + { type: type, message: { role: 'assistant', content: [] }, parent_tool_use_id: parent_tool_use_id } + end + + def task_started(id, task_type = 'local_agent') + { type: 'system', subtype: 'task_started', task_id: id, task_type: task_type }.compact + end + + def task_done(id, status = 'completed') + { type: 'system', subtype: 'task_notification', task_id: id, status: status } + end + + def task_updated(id, patch) + { type: 'system', subtype: 'task_updated', task_id: id, patch: patch } + end + + def feed(*frames) + frames.each { |frame| lifecycle.frame(frame) } + end + + # Whether the run had ended after each of +frames+. + def ended_after_each(*frames) + frames.map do |frame| + lifecycle.frame(frame) + lifecycle.ended? + end + end + + def armed? + lifecycle.ceiling_armed? + end + + describe 'B1: the run ends at idle after a result' do + it 'does not end at the result while the CLI reports running, and ends at the idle that follows' do + expect(ended_after_each(state('running'), result, state('idle'))).to eq([false, false, true]) + end + + it 'does not end at an idle that comes before any result, and ends at the result' do + expect(ended_after_each(state('running'), state('idle'), result)).to eq([false, false, true]) + end + + it 'does the same when idle is the first state the CLI reports' do + expect(ended_after_each(state('idle'), result)).to eq([false, true]) + end + + # Messages written before any result are merged into fewer turns: results + # follow each other with no idle until all queued input is served. + it 'does not end at a second result with no idle in between' do + expect(ended_after_each(state('running'), result, result, state('idle'))).to eq([false, false, false, true]) + end + + it 'reports the latest session state, and none from a CLI that sends none' do + expect(lifecycle.session_state).to be_nil + feed(result) + expect(lifecycle.session_state).to be_nil + + feed(state('running'), state('requires_action')) + expect(lifecycle.session_state).to eq('requires_action') + end + end + + describe 'B2: with no session state, the run ends at the first result with no tracked task in flight' do + it 'ends at the first result' do + expect(ended_after_each(result)).to eq([true]) + end + + it 'ends at the first result when nothing needs the control channel, whatever the state says' do + facts[:bidirectional_needs] = false + + expect(ended_after_each(state('running'), result)).to eq([false, true]) + end + + it 'tracks task_started only for deferring task types' do + feed(task_started('a', 'local_agent'), task_started('w', 'local_workflow'), + task_started('s', 'local_shell'), task_started('n', nil)) + + expect(%w[a w s n].select { |id| lifecycle.inflight?(id) }).to eq(%w[a w]) + end + + it 'ignores frames with a missing or empty task_id' do + feed({ type: 'system', subtype: 'task_started', task_type: 'local_agent' }, task_started('')) + + expect([nil, ''].select { |id| lifecycle.inflight?(id) }).to be_empty + expect(ended_after_each(result)).to eq([true]) # nothing in the ledger holds the run open + end + + it 'clears on task_notification regardless of status' do + feed(task_started('a'), task_done('a', 'failed')) + + expect(lifecycle.inflight?('a')).to be(false) + expect(ended_after_each(result)).to eq([true]) + end + + it 'clears on terminal task_updated statuses only' do + feed(task_started('a'), task_updated('a', { status: 'running' })) + expect(lifecycle.inflight?('a')).to be(true) + + feed(task_updated('a', { status: 'killed' })) + expect(lifecycle.inflight?('a')).to be(false) + expect(ended_after_each(result)).to eq([true]) + end + + it 'tolerates a non-Hash or absent patch on task_updated' do + feed(task_started('a'), task_updated('a', 'completed'), { type: 'system', subtype: 'task_updated', task_id: 'a' }) + + expect(lifecycle.inflight?('a')).to be(true) + end + + it 'ignores background_tasks_changed in both directions' do + feed(task_started('a'), { type: 'system', subtype: 'background_tasks_changed', tasks: [] }) + expect(lifecycle.inflight?('a')).to be(true) + + feed({ type: 'system', subtype: 'background_tasks_changed', + tasks: [{ task_id: 'b', task_type: 'local_agent', description: 'other' }] }) + expect(%w[a b].select { |id| lifecycle.inflight?(id) }).to eq(%w[a]) + end + + it 'is a no-op for terminal frames about unknown task ids' do + feed(task_done('ghost'), task_updated('ghost', { status: 'failed' })) + + expect(lifecycle.inflight?('ghost')).to be(false) + expect(ended_after_each(result)).to eq([true]) + end + end + + describe 'B3: a tracked task in flight holds the run open' do + { 'a task_notification' => { subtype: 'task_notification', status: 'completed' }, + 'a terminal task_updated' => { subtype: 'task_updated', patch: { status: 'killed' } } }.each do |drain, settled| + it "does not end at a result with a task in flight, and ends at the next one once #{drain} settled it" do + frames = [task_started('bg'), result, { type: 'system', task_id: 'bg', **settled }, result] + + expect(ended_after_each(*frames)).to eq([false, false, false, true]) + end + end + + it 'does not end at idle either: only the idle of the follow-up turn ends the run (B2/B3)' do + frames = [task_started('bg'), result, state('idle'), task_done('bg'), state('running'), turn_frame, result, state('idle')] + + expect(ended_after_each(*frames)).to eq([false, false, false, false, false, false, false, true]) + end + + it 'does not hold the run open for a task that settled before the result when the CLI reports no state' do + expect(ended_after_each(task_started('bg'), task_done('bg'), result)).to eq([false, false, true]) + end + end + + describe 'B4: the ceiling is armed only for the wait between turns' do + capped_seconds = { 300 => 0.3, 10**12 => 2_147_483.647 } + + [nil, 'running', 'idle', 'requires_action'].each do |session_state| + [0, 300, 10**12].each do |ms| + [true, false].each do |needs| + arms = session_state == 'running' && ms.positive? && needs + ends = !needs || session_state.nil? || session_state == 'idle' + title = "a result under state #{session_state.inspect}, ceiling #{ms}, control channel #{needs ? 'in use' : 'unused'}: " \ + "#{arms ? "arms for #{capped_seconds.fetch(ms)}s" : 'arms nothing'}, #{ends ? 'ends the run' : 'leaves the run open'}" + + it title, ceiling_ms: ms do + facts[:bidirectional_needs] = needs + feed(state(session_state)) if session_state + feed(result) + + expect([armed?, lifecycle.ended?]).to eq([arms, ends]) + expect(sleeper.arms.map(&:seconds)).to eq(arms ? [capped_seconds.fetch(ms)] : []) + end + end + end + end + + it 'caps the sleep at MAX_RUN_END_CEILING_MS', ceiling_ms: 10**12 do + feed(state('running'), result) + + expect(sleeper.arms.last.seconds).to eq(ClaudeAgentSDK::Query::MAX_RUN_END_CEILING_MS / 1000.0) + end + + it 'is not armed by running before the first result' do + feed(state('running')) + + expect(armed?).to be(false) + end + + it 'starts over at each running after a result, and at each result' do + feed(state('running'), result) + expect([sleeper.arms.length, sleeper.stops]).to eq([1, 0]) + + feed(state('running')) + expect([sleeper.arms.length, sleeper.stops]).to eq([2, 1]) + + feed(result) + expect([sleeper.arms.length, sleeper.stops]).to eq([3, 2]) + expect(armed?).to be(true) + end + + it 'is not armed mid-turn' do + feed(state('running'), result, turn_frame, task_started('bg'), task_done('bg'), state('running')) + + expect(armed?).to be(false) + expect(lifecycle.ended?).to be(false) + end + + it 'is not armed once the run has ended' do + feed(state('running'), result) + sleeper.fire + expect(lifecycle.ended?).to be(true) + + feed(result) # another result of the same run: no state change reopened it + expect(armed?).to be(false) + end + + it 'is not armed once the run is final, ended or not' do + feed(state('running'), result) + lifecycle.stdin_closing # final, and the CLI never reported idle: the run has not ended + feed(state('running')) + expect([armed?, lifecycle.ended?]).to eq([false, false]) + + feed(turn_frame, result) + expect(armed?).to be(false) + end + + it 'is not armed while the Query is closed (B4 closed)' do + feed(state('running')) + facts[:closed] = true + feed(result) + + expect(armed?).to be(false) + expect(lifecycle.ended?).to be(false) + end + end + + describe 'B5: what stops the clock and what starts it over' do + %w[assistant stream_event].each do |type| + it "is stopped by a main-thread #{type} and re-armed by the turn's result" do + feed(state('running'), result, turn_frame(type)) + expect(armed?).to be(false) + + sleeper.fire_stale # the stopped sleeper was already waking up + expect([armed?, lifecycle.ended?]).to eq([false, false]) + + feed(result) + expect(armed?).to be(true) + sleeper.fire + expect(lifecycle.ended?).to be(true) + end + + it "is not stopped by a subagent's #{type}" do + feed(state('running'), result, turn_frame(type, parent_tool_use_id: 'toolu_1')) + + expect(armed?).to be(true) + expect([sleeper.arms.length, sleeper.stops]).to eq([1, 0]) # the same sleeper, not a new one + sleeper.fire + expect(lifecycle.ended?).to be(true) + end + end + + it 'reopens a run the ceiling ended when a main-thread turn starts with no state change' do + feed(state('running'), result) + sleeper.fire + expect(lifecycle.ended?).to be(true) + + feed(turn_frame) + expect([lifecycle.ended?, armed?]).to eq([false, false]) + expect(ended_after_each(result, state('idle'))).to eq([false, true]) + end + + it 'does not reopen an ended run for a subagent frame' do + feed(state('running'), result, state('idle'), turn_frame(parent_tool_use_id: 'toolu_1')) + + expect(lifecycle.ended?).to be(true) + end + + it 'is stopped by requires_action and re-armed by the running that follows' do + feed(state('running'), result, state('requires_action')) + expect(armed?).to be(false) + + sleeper.fire_stale + expect([armed?, lifecycle.ended?]).to eq([false, false]) + + feed(state('running')) + expect(armed?).to be(true) + sleeper.fire + expect(lifecycle.ended?).to be(true) + end + + it 'is not armed by a result that arrives while the SDK is answering a request' do + feed(state('requires_action'), result) + expect([armed?, lifecycle.ended?]).to eq([false, false]) + + feed(state('running')) + expect(armed?).to be(true) + sleeper.fire + expect(lifecycle.ended?).to be(true) + end + + %w[running requires_action].each do |wake| + it "reopens an ended run for work the CLI takes up after idle (#{wake})" do + feed(state('running'), result, state('idle')) + expect(lifecycle.ended?).to be(true) + + feed(state(wake)) + expect(lifecycle.ended?).to be(false) + expect(armed?).to be(wake == 'running') # past a result, between turns + expect(ended_after_each(state('running'), turn_frame, result, state('idle'))).to eq([false, false, false, true]) + end + end + + it 'leaves a tracked agent alone when the ceiling passes, and starts over once the agent settles' do + feed(state('running'), task_started('bg'), result) + expect(armed?).to be(true) + + sleeper.fire # with bg still running + expect([armed?, lifecycle.ended?]).to eq([false, false]) + + feed(task_done('bg')) + expect(armed?).to be(true) + sleeper.fire + expect(lifecycle.ended?).to be(true) + end + + it 'does not start over when a task settles while another is still in flight' do + feed(state('running'), task_started('one'), task_started('two'), result) + sleeper.fire + feed(task_done('one')) + expect(armed?).to be(false) + + feed(task_done('two')) + expect(armed?).to be(true) + end + end + + describe 'B6: a ceiling that passes under a control request the SDK is still answering' do + it 're-arms instead of ending the run, and ends it at the next wake once the request is answered' do + feed(state('running'), result) + facts[:control_requests_in_flight] = true + sleeper.fire + expect([armed?, lifecycle.ended?]).to eq([true, false]) # the clock starts over + expect(sleeper.arms.length).to eq(2) + + facts[:control_requests_in_flight] = false + sleeper.fire + expect([armed?, lifecycle.ended?]).to eq([false, true]) + end + + it 'leaves the request alone when a tracked agent is in flight too: the agent decides' do + feed(state('running'), task_started('bg'), result) + facts[:control_requests_in_flight] = true + sleeper.fire + + expect([armed?, lifecycle.ended?]).to eq([false, false]) + end + end + + describe 'B7: a message that is about to be written owes a run of its own' do + it 'reopens an ended run and arms nothing' do + feed(result) + expect(lifecycle.ended?).to be(true) + + lifecycle.message_will_write + expect([lifecycle.ended?, armed?]).to eq([false, false]) + end + + it 'leaves a run that has not ended as it is, and stops its clock' do + feed(state('running'), result) + lifecycle.message_will_write + + expect([lifecycle.ended?, armed?, sleeper.stops]).to eq([false, false, 1]) + end + + it "owes a result of its own: a late idle of the previous message's run does not end it" do + feed(state('running'), result, state('idle')) + lifecycle.message_will_write + + expect(ended_after_each(state('idle'), state('running'), result, state('idle'))).to eq([false, false, false, true]) + end + + it 'does not arm at a running that comes before its own result' do + feed(state('running'), result, state('idle')) + lifecycle.message_will_write + feed(state('running')) + + expect(armed?).to be(false) + end + + it 'releases the waiter the ended run woke, and makes a later wait wait for the new run' do + Async do |task| + woken = task.async { lifecycle.wait } + feed(result) # the run the first waiter holds ends + lifecycle.message_will_write # and a fresh one takes its place + task.with_timeout(10) { woken.wait } + + later = task.async { lifecycle.wait } + 3.times { task.yield } + expect(later.finished?).to be(false) + + feed(result) + task.with_timeout(10) { later.wait } + expect(lifecycle.ended?).to be(true) + ensure + [woken, later].each { |waiter| waiter&.stop } + end.wait + end + end + + describe 'B9: final' do + it 'keeps an ended run ended, whatever comes next' do + feed(result) + lifecycle.stdin_closing + expect([lifecycle.final?, lifecycle.ended?]).to eq([true, true]) + + lifecycle.message_will_write + feed(state('running'), turn_frame) + expect([lifecycle.final?, lifecycle.ended?]).to eq([true, true]) + end + + it 'does not end a run that has not ended' do + expect(lifecycle.final?).to be(false) + feed(state('running'), result) + lifecycle.stdin_closing + + expect([lifecycle.final?, lifecycle.ended?]).to eq([true, false]) + expect(ended_after_each(state('idle'))).to eq([true]) # the CLI's idle still ends it + end + end + + describe 'B10: the run is ended before its sleeper is stopped' do + # Stopping a real sleeper hands the reactor to whatever else is ready; a + # message written then must find the run already ended (B7). + def ended_when_stopped + seen = [] + sleeper.on_stop = -> { seen << lifecycle.ended? } + yield + seen + ensure + sleeper.on_stop = nil + end + + it 'at the idle that ends the run' do + feed(state('running'), result) + + expect(ended_when_stopped { feed(state('idle')) }).to eq([true]) + end + + it 'when the reader is gone' do + feed(state('running'), result) + + expect(ended_when_stopped { lifecycle.reader_gone }).to eq([true]) + end + end + + describe 'B11: every exit path clears the sleeper' do + before { feed(state('running'), result) } + + it 'reader_gone stops it, ends the run and makes it final' do + lifecycle.reader_gone + + expect([armed?, sleeper.stops, lifecycle.ended?, lifecycle.final?]).to eq([false, 1, true, true]) + end + + it 'stdin_closing stops it and makes the run final' do + lifecycle.stdin_closing + + expect([armed?, sleeper.stops, lifecycle.ended?, lifecycle.final?]).to eq([false, 1, false, true]) + end + end + + # A characterization of the empty-input shape, not a promise: closing stdin + # with nothing written cannot tell a resume that has work of its own from one + # that has none (anthropics/claude-agent-sdk-python#1226), and a resumed CLI + # may have armed the ceiling by then. The sleeper is left for the read task's + # stop. + describe 'B14: empty input is final only (#1226)' do + it 'leaves an armed sleeper armed and the run unended' do + feed(state('running'), result) + lifecycle.empty_input + + expect([lifecycle.final?, armed?, sleeper.stops, lifecycle.ended?]).to eq([true, true, 0, false]) + end + + it 'is final with nothing armed too' do + lifecycle.empty_input + + expect([lifecycle.final?, armed?, lifecycle.ended?]).to eq([true, false, false]) + end + end + + describe 'Q5: a wake from an earlier arm never touches a later one' do + it 'stands down when its arm was replaced by the re-arm under a control request' do + feed(state('running'), result) # the first arm + facts[:control_requests_in_flight] = true + sleeper.fire # re-armed: the second arm + facts[:control_requests_in_flight] = false + + sleeper.fire_stale # the first arm wakes again + expect([armed?, lifecycle.ended?, sleeper.stops]).to eq([true, false, 0]) # the second arm is intact + + sleeper.fire + expect([armed?, lifecycle.ended?]).to eq([false, true]) + end + + it 'stands down when the run was reopened and armed again since' do + feed(state('running'), result, turn_frame, result) # stopped by the turn, armed again by its result + + sleeper.fire_stale + expect([armed?, lifecycle.ended?]).to eq([true, false]) + end + end + + describe '#wait' do + it 'returns at once when nothing needs the control channel' do + facts[:bidirectional_needs] = false + + Async { |task| task.with_timeout(10) { lifecycle.wait } }.wait + expect(lifecycle.ended?).to be(false) + end + + it 'returns at once when the run has already ended' do + feed(result) + + Async { |task| task.with_timeout(10) { lifecycle.wait } }.wait + end + + it 'waits for the end of the run when the control channel is in use' do + Async do |task| + waiter = task.async { lifecycle.wait } + feed(state('running'), result) + 3.times { task.yield } + expect(waiter.finished?).to be(false) + + feed(state('idle')) + task.with_timeout(10) { waiter.wait } + ensure + waiter&.stop + end.wait + end + + it 'is released when the reader is gone' do + Async do |task| + waiter = task.async { lifecycle.wait } + lifecycle.reader_gone + task.with_timeout(10) { waiter.wait } + ensure + waiter&.stop + end.wait + end + end +end diff --git a/spec/unit/query_end_run_order_spec.rb b/spec/unit/query_end_run_order_spec.rb index 4ba86227..5fbc202f 100644 --- a/spec/unit/query_end_run_order_spec.rb +++ b/spec/unit/query_end_run_order_spec.rb @@ -4,13 +4,14 @@ require 'async' require 'async/queue' -# Query#end_run has one suspension point: stopping the ceiling sleeper hands -# the reactor to whatever else is ready. A stream_input task that is ready at -# that moment writes its next message, and the run that message belongs to is -# decided by what it sees: an ended run is reopened as a fresh one, a run -# that is still open is joined. So the run must already count as ended when -# end_run reaches that suspension point, or the message joins the run that is -# about to end and stdin closes before its own run produced a frame. +# Ending a run (Query::RunLifecycle#end_run) hands the reactor over: stopping +# the ceiling sleeper lets whatever else is ready run. A stream_input task +# that is ready at that moment writes its next message, and the run that +# message belongs to is decided by what it sees: an ended run is reopened as +# a fresh one, a run that is still open is joined. So the run must already +# count as ended when end_run reaches that point, or the message joins the +# run that is about to end and stdin closes before its own run produced a +# frame. # # No clock takes part in the ordering below: two enqueues make the read loop # and the stream task ready, in that order, and the example then only yields @@ -95,7 +96,7 @@ def reactor_settles(task, limit: 10) reactor_settles(task) { transport.frames_handled == 2 } # The precondition of the whole example: the result armed the ceiling # sleeper, so ending the run has a task to stop — the suspension point. - expect(query.instance_variable_get(:@run_end_ceiling_task)).not_to be_nil + expect(query.run_lifecycle.ceiling_armed?).to be(true) transport.frames.enqueue(session_state('idle')) # the read loop becomes ready first ... gate.enqueue(:go) # ... and the stream task second diff --git a/spec/unit/query_run_end_spec.rb b/spec/unit/query_run_end_spec.rb index 95f0e730..14b5d30c 100644 --- a/spec/unit/query_run_end_spec.rb +++ b/spec/unit/query_run_end_spec.rb @@ -8,6 +8,13 @@ # between turns by CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS, and at the first # result with no tracked task in flight from a CLI that sends no state. # +# This is the decision as a Query makes it, through its read loop and its +# stdin-closing tasks over a transport double. The decision itself — every +# state and frame order — is tabled in query/run_lifecycle_spec.rb, against +# Query::RunLifecycle alone; what is kept here is what only a Query can +# show: which frames reach the lifecycle and in what order with the mirror +# flush and the stream, the real sleeper, the real predicates. +# # Nothing here waits on a clock to "let the read loop catch up": frames are # followed by a barrier (#with_query), and the ceiling is fired by hand # (#ceiling_passes) in every example but one. @@ -261,6 +268,31 @@ def finish(task, awaited) end end + # A message is serialized before the run is reopened for it: one that + # verbatim_prompts cannot mark raises there, and a run reopened for a + # message that never went out would wait for a result nobody owes. + it 'does not reopen the run for a message that fails to serialize' do + gate = Async::Queue.new + prompts = Enumerator.new do |y| + y << user_message('one') + gate.dequeue + y << 'not a JSON object' + end + + with_query(verbatim_prompts: true) do |query, feed, ended, task, written| + allow(query).to receive(:warn) # stream_input logs the error and goes on to close stdin + streamer = task.async { query.stream_input(prompts) } + expect(written.pop(timeout: 10)).to include('one') + feed.call(state('running'), result, state('idle')) # the first message's run ends + expect(ended).to be_empty # the stream is still open + + gate.enqueue(:go) + finish(task, streamer) + expect(query.run_lifecycle.ended?).to be(true) + expect(ended).not_to be_empty + end + end + # Frame sequence observed from CLI 2.1.285 for three messages written # before any result: it merges queued messages into fewer turns (two # results here) and reports no idle until all queued input is served. @@ -303,154 +335,72 @@ def finish(task, awaited) end end - describe 'run-end ceiling' do - # The ceiling is a sleeper task that calls end_run_at_ceiling(generation) - # when its sleep is over. These examples leave the default ceiling of ten - # minutes in place, so the real sleeper never wakes, and make that call - # themselves: what is under test is which sleeper is armed when, and what - # one that wakes is allowed to do — not how long the sleep takes. The one - # example that sleeps through a real ceiling is the first. - def ceiling_armed?(query) - !query.instance_variable_get(:@run_end_ceiling_task).nil? - end - - # The generation the armed sleeper holds (it stands down when it wakes - # with a stale one). - def ceiling_generation(query) - query.instance_variable_get(:@run_end_ceiling_generation) - end - - # What the sleeper armed at +generation+ does when its sleep is over; - # +feed+ then lets what that woke run. - def ceiling_passes(query, generation, feed) - query.send(:end_run_at_ceiling, generation) - feed.call - end - - it 'ends the run when no idle arrives within the ceiling after a result' do - with_query(run_end_ceiling_ms: 300) do |query, feed, ended, task| - waiter = task.async { query.wait_for_result_and_end_input } - feed.call(state('running'), result) - - # No idle follows: only the ceiling, a real one here, ends this run. - finish(task, waiter) - expect(ended).not_to be_empty - end - end - - it 'does not end the run at a result while the CLI still reports running' do - with_query do |query, feed, ended, task| - task.async { query.wait_for_result_and_end_input } - feed.call(state('running'), result) - - expect(ended).to be_empty - expect(ceiling_armed?(query)).to be(true) # the clock for the wait between turns - end - end - - it 'is stopped by a main-thread turn and re-armed by its result' do - with_query do |query, feed, ended, task| - waiter = task.async { query.wait_for_result_and_end_input } - feed.call(state('running'), result) - stopped = ceiling_generation(query) - feed.call(main_assistant) - expect(ceiling_armed?(query)).to be(false) - - ceiling_passes(query, stopped, feed) # the stopped sleeper was already waking up - expect(ended).to be_empty - - feed.call(result) - expect(ceiling_armed?(query)).to be(true) - ceiling_passes(query, ceiling_generation(query), feed) - finish(task, waiter) - expect(ended).not_to be_empty - end - end - - it 'is not stopped by subagent messages' do - with_query do |query, feed, ended, task| - waiter = task.async { query.wait_for_result_and_end_input } - feed.call(state('running'), result) - armed = ceiling_generation(query) - feed.call(main_assistant.merge(parent_tool_use_id: 'toolu_1')) - expect(ceiling_armed?(query)).to be(true) - expect(ceiling_generation(query)).to eq(armed) # the same sleeper, not a new one - - ceiling_passes(query, armed, feed) - finish(task, waiter) - expect(ended).not_to be_empty + describe 'the mirror flush at a result' do + # A result wakes whoever waits for the run's end and whoever reads the + # stream, and both may go on to read the session store: the mirror is + # flushed first. The flush suspends the read loop (a lock, a thread + # join), so the order shows while it is parked. + it 'ends the run and hands the result to the stream only once the mirror is flushed' do + queue = Async::Queue.new + transport, ended = queue_fed_transport(queue) + query = build_query(transport) + gate = Async::Queue.new + flushes = 0 + batcher = instance_double(ClaudeAgentSDK::TranscriptMirrorBatcher, enqueue: nil, close: nil) + allow(batcher).to receive(:flush) do + flushes += 1 + gate.dequeue if flushes == 1 # the flush for the result; the one at the end of the read loop passes end - end + query.set_transcript_mirror_batcher(batcher) + seen = [] - it 'is stopped by requires_action and re-armed by the running that follows' do - with_query do |query, feed, ended, task| + Async do |task| + query.start + task.async { query.receive_messages { |message| seen << message[:type] } } waiter = task.async { query.wait_for_result_and_end_input } - feed.call(state('running'), result) - stopped = ceiling_generation(query) - feed.call(state('requires_action')) - expect(ceiling_armed?(query)).to be(false) + queue.enqueue(result) # from a CLI that sends no state: this result ends the run + task.with_timeout(10) { task.yield until flushes == 1 } + 5.times { task.yield } # whatever the result could wake before the flush is done has run - ceiling_passes(query, stopped, feed) - expect(ended).to be_empty + expect([seen, query.run_lifecycle.ended?, ended]).to eq([[], false, []]) - feed.call(state('running')) - expect(ceiling_armed?(query)).to be(true) - ceiling_passes(query, ceiling_generation(query), feed) + gate.enqueue(:flushed) finish(task, waiter) - expect(ended).not_to be_empty - end - end - - # "running" before any result is a turn starting, not the wait between - # turns that the ceiling bounds. - it 'is not armed by running before the first result' do - with_query do |query, feed, ended, task| - task.async { query.wait_for_result_and_end_input } - feed.call(state('running')) - - expect(ceiling_armed?(query)).to be(false) - expect(ended).to be_empty - end + task.with_timeout(10) { task.yield until seen.any? } + expect([seen, query.run_lifecycle.ended?, ended]).to eq([%w[result], true, [true]]) + ensure + query.close + release_parked_tasks(task) + end.wait end + end - it 'is not armed mid-turn' do - with_query do |query, feed, ended, task| - task.async { query.wait_for_result_and_end_input } - feed.call(state('running'), result, main_assistant, - task_started('bg-1'), task_done('bg-1'), state('running')) + describe 'run-end ceiling' do + # The ceiling is a sleeper that ends the run when its sleep is over. Which + # sleeper is armed when, and what one that wakes may do, is tabled in + # query/run_lifecycle_spec.rb. The two examples here that need one to wake + # hand the Query a FakeSleeper (`sleeper:`), which never wakes by itself, + # and wake it by hand; the others keep the real sleeper, a child of the + # read task, and only the first sleeps through a real ceiling. + let(:sleeper) { FakeSleeper.new } - expect(ceiling_armed?(query)).to be(false) - expect(ended).to be_empty - end + def ceiling_armed?(query) + query.run_lifecycle.ceiling_armed? end - it 'is not armed by a result that arrives while the SDK is answering a request' do - with_query do |query, feed, ended, task| - waiter = task.async { query.wait_for_result_and_end_input } - feed.call(state('requires_action'), result) - expect(ceiling_armed?(query)).to be(false) - expect(ended).to be_empty - - feed.call(state('running')) - expect(ceiling_armed?(query)).to be(true) - ceiling_passes(query, ceiling_generation(query), feed) - finish(task, waiter) - expect(ended).not_to be_empty - end + # The sleep of the armed sleeper is over; +feed+ then lets what that woke + # run. + def ceiling_passes(feed) + sleeper.fire + feed.call end - it 'leaves a tracked agent alone and starts over once it settles' do - with_query do |query, feed, ended, task| + it 'ends the run when no idle arrives within the ceiling after a result' do + with_query(run_end_ceiling_ms: 300) do |query, feed, ended, task| waiter = task.async { query.wait_for_result_and_end_input } - feed.call(state('running'), task_started('bg-1'), result) - - ceiling_passes(query, ceiling_generation(query), feed) # with bg-1 still running - expect(ended).to be_empty - expect(ceiling_armed?(query)).to be(false) + feed.call(state('running'), result) - feed.call(task_done('bg-1')) - expect(ceiling_armed?(query)).to be(true) - ceiling_passes(query, ceiling_generation(query), feed) + # No idle follows: only the ceiling, a real one here, ends this run. finish(task, waiter) expect(ended).not_to be_empty end @@ -469,13 +419,13 @@ def ceiling_passes(query, generation, feed) input: { hook_event_name: 'PreToolUse', tool_name: 'Bash', tool_input: {}, session_id: 's', cwd: '/tmp' } } } - with_query do |query, feed, ended, task, written| + with_query(sleeper: sleeper) do |query, feed, ended, task, written| query.instance_variable_set(:@hook_callbacks, { 'hook_0' => parked_hook }) waiter = task.async { query.wait_for_result_and_end_input } feed.call(state('running'), result, hook_request) expect(entered.pop(timeout: 10)).to be(true) - ceiling_passes(query, ceiling_generation(query), feed) # while the hook is still running + ceiling_passes(feed) # while the hook is still running expect(ended).to be_empty expect(ceiling_armed?(query)).to be(true) # the clock starts over instead @@ -484,7 +434,7 @@ def ceiling_passes(query, generation, feed) expect(ended).to be_empty # on a stdin that was still open feed.call - ceiling_passes(query, ceiling_generation(query), feed) + ceiling_passes(feed) finish(task, waiter) expect(ended).not_to be_empty ensure @@ -492,34 +442,6 @@ def ceiling_passes(query, generation, feed) end end - it 'arms no ceiling when the ceiling is 0, and waits for idle' do - with_query(run_end_ceiling_ms: 0) do |query, feed, ended, task| - waiter = task.async { query.wait_for_result_and_end_input } - feed.call(state('running'), result) - expect(ceiling_armed?(query)).to be(false) - expect(ended).to be_empty - - feed.call(state('idle')) - finish(task, waiter) - expect(ended).not_to be_empty - end - end - - # 10**12 ms is past what a timer can take; the ceiling is capped - # (MAX_RUN_END_CEILING_MS), so arming it must neither raise nor fire. - it 'waits for idle with a ceiling of 10**12' do - with_query(run_end_ceiling_ms: 10**12) do |query, feed, ended, task| - waiter = task.async { query.wait_for_result_and_end_input } - feed.call(state('running'), result) - expect(ceiling_armed?(query)).to be(true) - expect(ended).to be_empty - - feed.call(state('idle')) - finish(task, waiter) - expect(ended).not_to be_empty - end - end - it 'reopens a stream the ceiling ended for a main-thread turn' do gate = Async::Queue.new prompts = Enumerator.new do |y| @@ -527,10 +449,10 @@ def ceiling_passes(query, generation, feed) gate.dequeue end - with_query do |query, feed, ended, task| + with_query(sleeper: sleeper) do |query, feed, ended, task| streamer = task.async { query.stream_input(prompts) } feed.call(state('running'), result) - ceiling_passes(query, ceiling_generation(query), feed) # ends the run while the stream is still open + ceiling_passes(feed) # ends the run while the stream is still open feed.call(main_assistant) gate.enqueue(:go) feed.call # the stream is exhausted: the streamer waits for the reopened run @@ -554,6 +476,61 @@ def ceiling_passes(query, generation, feed) end end + # A result the read loop already holds when the Query is closed must not + # leave a sleeper behind. Query#close marks the Query closed, closes the + # mirror batcher and only then stops the reader, so the reader's own flush + # for that result can return in between, with the run neither ended nor + # final: all that keeps the lifecycle from arming is that it asks the + # Query whether it is closed each time it would. + # + # Staged with two gates and no clock. The read loop parks in the flush for + # the result; the batcher's close, which Query#close calls once the Query + # is marked closed, releases that flush and parks in turn. The reader then + # finishes the result while the close has not yet stopped it. + { + 'a task of the reactor' => ->(query, task) { task.async { query.close } }, + 'another thread' => ->(query, _task) { Thread.new { query.close } } + }.each do |origin, start_close| + it "arms no ceiling for a result that resumes after a close from #{origin} has begun" do + queue = Async::Queue.new + transport, = queue_fed_transport(queue) + query = build_query(transport, sleeper: sleeper) + flush_gate = Async::Queue.new + close_gate = Async::Queue.new + flushes = 0 + batcher = instance_double(ClaudeAgentSDK::TranscriptMirrorBatcher, enqueue: nil) + allow(batcher).to receive(:flush) do + flushes += 1 + flush_gate.dequeue if flushes == 1 # the flush for the result; the one at the end of the read loop passes + end + allow(batcher).to receive(:close) do + flush_gate.enqueue(:closed) # the Query is marked closed and its reader still runs + close_gate.dequeue + end + query.set_transcript_mirror_batcher(batcher) + + Async do |task| + query.start + handled = Thread::Queue.new # a barrier: answered once the read loop is done with the result + [state('running'), result, handled].each { |frame| queue.enqueue(frame) } + task.with_timeout(10) { task.yield until flushes == 1 } + + closing = start_close.call(query, task) + expect(handled.pop(timeout: 10)).to eq(:reached) + lifecycle = query.run_lifecycle + expect([lifecycle.ended?, lifecycle.final?, lifecycle.ceiling_armed?, sleeper.arms.length]).to eq([false, false, false, 0]) + + close_gate.enqueue(:stop_the_reader) + task.with_timeout(10) { task.yield while closing.alive? } + expect([lifecycle.ended?, lifecycle.final?, sleeper.arms.length]).to eq([true, true, 0]) + ensure + close_gate.enqueue(:stop_the_reader) # never leave a close parked + query.close + release_parked_tasks(task) + end.wait + end + end + # A pending ceiling sleeper is a child task; if an exit path forgot to # clear it, the enclosing reactor would stay alive for the full ceiling. it 'leaves no sleeper behind when the CLI exits mid-wait' do diff --git a/spec/unit/query_spec.rb b/spec/unit/query_spec.rb index 0a620fd3..c57b49e7 100644 --- a/spec/unit/query_spec.rb +++ b/spec/unit/query_spec.rb @@ -268,7 +268,7 @@ def hooks_config # background_tasks_changed is typed for consumers (BackgroundTasksChangedMessage) # but must stay invisible to the stdin-close bookkeeping, in both - # directions — see the comment above Query#track_task_lifecycle. + # directions — see the comment above Query::RunLifecycle#track_task_lifecycle. it 'does not narrow the in-flight set from an empty background_tasks_changed snapshot' do queue = Async::Queue.new transport, ended = queue_fed_transport(queue) @@ -283,7 +283,7 @@ def hooks_config queue.enqueue({ type: 'system', subtype: 'background_tasks_changed', tasks: [] }) queue.enqueue(sample_result_message) task.sleep 0.05 - expect(query.instance_variable_get(:@inflight_tasks).to_a).to eq(%w[fg-1]) + expect(query.run_lifecycle.inflight?('fg-1')).to be(true) expect(ended).to be_empty # fg-1 still in flight despite the empty snapshot queue.enqueue({ type: 'system', subtype: 'task_notification', task_id: 'fg-1', status: 'completed' }) @@ -308,7 +308,7 @@ def hooks_config tasks: [{ task_id: 'bg-9', task_type: 'local_agent', description: 'observer' }] }) queue.enqueue(sample_result_message) task.with_timeout(2.0) { waiter.wait } - expect(query.instance_variable_get(:@inflight_tasks)).to be_empty + expect(query.run_lifecycle.inflight?('bg-9')).to be(false) expect(ended).not_to be_empty # first result still closes stdin ensure query.close @@ -397,70 +397,6 @@ def hooks_config end end - describe '#track_task_lifecycle' do - let(:query) { described_class.new(transport: mock_transport, is_streaming_mode: true) } - - def inflight - query.instance_variable_get(:@inflight_tasks) - end - - def track(message) - query.send(:track_task_lifecycle, message) - end - - it 'adds task_started only for deferring task types' do - track({ type: 'system', subtype: 'task_started', task_id: 'a', task_type: 'local_agent' }) - track({ type: 'system', subtype: 'task_started', task_id: 'w', task_type: 'local_workflow' }) - track({ type: 'system', subtype: 'task_started', task_id: 's', task_type: 'local_shell' }) - track({ type: 'system', subtype: 'task_started', task_id: 'n' }) - expect(inflight.to_a.sort).to eq(%w[a w]) - end - - it 'ignores frames with a missing or empty task_id' do - track({ type: 'system', subtype: 'task_started', task_type: 'local_agent' }) - track({ type: 'system', subtype: 'task_started', task_id: '', task_type: 'local_agent' }) - expect(inflight).to be_empty - end - - it 'clears on task_notification regardless of status' do - track({ type: 'system', subtype: 'task_started', task_id: 'a', task_type: 'local_agent' }) - track({ type: 'system', subtype: 'task_notification', task_id: 'a', status: 'failed' }) - expect(inflight).to be_empty - end - - it 'clears on terminal task_updated statuses only' do - track({ type: 'system', subtype: 'task_started', task_id: 'a', task_type: 'local_agent' }) - track({ type: 'system', subtype: 'task_updated', task_id: 'a', patch: { status: 'running' } }) - expect(inflight.to_a).to eq(%w[a]) - - track({ type: 'system', subtype: 'task_updated', task_id: 'a', patch: { status: 'killed' } }) - expect(inflight).to be_empty - end - - it 'tolerates a non-Hash or absent patch on task_updated' do - track({ type: 'system', subtype: 'task_started', task_id: 'a', task_type: 'local_agent' }) - track({ type: 'system', subtype: 'task_updated', task_id: 'a', patch: 'completed' }) - track({ type: 'system', subtype: 'task_updated', task_id: 'a' }) - expect(inflight.to_a).to eq(%w[a]) - end - - it 'ignores background_tasks_changed in both directions' do - track({ type: 'system', subtype: 'task_started', task_id: 'a', task_type: 'local_agent' }) - track({ type: 'system', subtype: 'background_tasks_changed', tasks: [] }) - expect(inflight.to_a).to eq(%w[a]) - - track({ type: 'system', subtype: 'background_tasks_changed', - tasks: [{ task_id: 'b', task_type: 'local_agent', description: 'other' }] }) - expect(inflight.to_a).to eq(%w[a]) - end - - it 'is a no-op for terminal frames about unknown task ids' do - track({ type: 'system', subtype: 'task_notification', task_id: 'ghost', status: 'completed' }) - track({ type: 'system', subtype: 'task_updated', task_id: 'ghost', patch: { status: 'failed' } }) - expect(inflight).to be_empty - end - end - describe '#reconnect_mcp_server' do it 'sends mcp_reconnect control request with camelCase serverName' do transport = instance_double(ClaudeAgentSDK::Transport, write: nil)