From 391476e96d09e28350a51a3e2a72e390f718b6b7 Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Mon, 13 Jul 2026 13:51:34 +0000 Subject: [PATCH 01/13] Add "Sample messages read" section to control-channel design doc MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Designs (does not implement) a per-sample conversation read — HTTP GET /evals//sample/messages and CLI `inspect ctl sample messages` — so agents can live-watch a sample's .messages, including samples still in the buffer and not yet written to the .eval log. Co-authored-by: Ransom Richardson <1209015+ransomr@users.noreply.github.com> Co-Authored-By: Claude Fable 5 --- design/control-channel.md | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/design/control-channel.md b/design/control-channel.md index 354c394475..52fe6e37a0 100644 --- a/design/control-channel.md +++ b/design/control-channel.md @@ -79,6 +79,8 @@ inspect ctl sample errors [TASK] # triage: samples that errored or we inspect ctl sample events TASK SID [EPOCH] # one sample's transcript events (cursored pull) # --cursor CURSOR / --tail N / --type / --full / --since-time / --until # (-f / --follow push is phase 4) +inspect ctl sample messages TASK SID [EPOCH] # one sample's current conversation, snapshot + # (planned — see "Sample messages read") inspect ctl process list # running Inspect processes (pid / keep-alive / tasks) inspect ctl process keep [PID] # keep a running process alive after its eval finishes inspect ctl process release [PID] # release a lingering keep-alive process @@ -136,6 +138,7 @@ inspect ctl │ ├── show TASK SID [EPOCH] # was: sample │ ├── errors [TASK] # was: errors (no TASK = across all tasks) │ ├── events TASK SID [EPOCH] # was: events (phase-4 --follow lands here unchanged) +│ ├── messages TASK SID [EPOCH] # planned: current-conversation snapshot (see "Sample messages read") │ ├── cancel TASK SID [EPOCH] # planned: cancel-sample │ └── requeue TASK SID [EPOCH] # planned: requeue ├── config [TASK] [...] # was: limits + buffer; view/retune launch config, scope per knob @@ -206,6 +209,7 @@ Known accepted edge in the busy handling: the sole-server rule counts *discovere | `ctl cancel-sample ` | planned | `ctl sample cancel TASK SID [EPOCH]` | compound verb dissolves; `EPOCH` required when the task runs >1 epoch | | `ctl requeue ` | planned | `ctl sample requeue TASK SID [EPOCH]` | `EPOCH` required when the task runs >1 epoch | | `ctl set-limit --time N` | planned | fold into `ctl config TASK --time-limit/--token-limit/--message-limit` | all retunable launch flags in one surface, same spellings as `inspect eval`; task-scoped knobs, so `TASK` follows the mutation selector rule | +| — | planned | `ctl sample messages TASK SID [EPOCH]` | new read: the sample's current conversation, summary-projected — see "Sample messages read" | | — | new | `ctl process list` | pids, keep-alive status, hosted task_ids; shipped with the reorg — the group's other verbs take a `PID` selector, so pids must be enumerable in-group | | — | later | `ctl eval-set list / show / cancel` | group slot ready-made | @@ -272,6 +276,7 @@ The conceptual surface, independent of wire protocol. Each operation becomes eit - List running evals (id, task, status, started_at, sample counts). - Eval status detail (config, current limits, sample queue depth, in-flight samples, completed counts, model usage so far). - List samples within an eval (status, started_at, current model/tool, token usage). +- Read one sample's current conversation — the live `TaskState.messages`, summary-projected (planned; see "Sample messages read"). **Read (eval-set-level)** - List eval-sets. @@ -349,6 +354,7 @@ The URL scheme has one rule — **three scopes, three roots**: process-scoped op | Samples changed since (recency delta) | `GET /evals//samples?active_since=` | 2 ✅ | | Sample transcript events (pull) | `GET /evals//sample/events?sample_id=&epoch=&since=` (JSON) | 2 ✅ | | Sample transcript events (push) | the pull URL with `Accept: text/event-stream` (SSE) | 4 | +| Sample conversation messages (snapshot) | `GET /evals//sample/messages?sample_id=&epoch=` | planned | | Eval-wide transcript fan-in (push only) | `GET /evals//samples/events` (SSE) | 4 | | Flush buffered samples to the log | `POST /tasks//log-flush` | 3 ✅ | | Read / modify retunable config (concurrency limits, buffer params) | `GET`+`PATCH /config` (process) and `/tasks//config` (task) | 3 ✅ (max-samples / max-sandboxes / max-subprocesses / max-connections / log-buffer / log-shared) | @@ -726,6 +732,34 @@ A wall-clock window is a snapshot query (no exactly-once requirement), which is Unlocks watchdog agents (cursored polling). Live-render TUIs follow once phase-4 push lands. +### Sample messages read (planned) + +> **Status: planned** (design only — nothing in this section is implemented). Motivated by issue #70: agents live-watching a sample want the *conversation* (`.messages`), not just the transcript firehose — in particular for samples that exist only in the process buffer, not yet written to the `.eval` log. + +**The gap.** A human watching a live sample opens `inspect view` and reads the conversation. An agent has no equivalent: the closest surface is `sample events`, and reconstructing "what does the conversation look like right now" from the transcript is genuinely hard — the messages ride inside `ModelEvent.input` / `output` (truncated in the default projection, enormous under `--full`), solver / agent code can assign `state.messages` without emitting any model event, and compaction rewrites the conversation while the event stream records only a `compaction` marker. Meanwhile the thing the agent wants already exists, materialized, in the running sample's `TaskState.messages` — serve it directly. The buffered case is what makes this control-channel work rather than `inspect log` work: for a running sample — or a completed one still sitting in the log buffer awaiting a flush — the conversation exists only inside the eval process, and the control channel is the only surface that can see it. + +**Surface.** + +- **HTTP:** `GET /evals//sample/messages?sample_id=&epoch=` — attempt-scoped beside `sample` and `sample/events`, with `sample_id` a query param per the reserved-characters rule. +- **CLI:** `inspect ctl sample messages TASK SID [EPOCH]` — a new verb in the `sample` noun group. Standard selector conventions apply: `TASK` is required (a `SAMPLE_ID` follows it), and `EPOCH` defaults to 1 on reads with the resolved `{sample_id, epoch}` echoed in the envelope. + +**Snapshot, not stream — deliberately no cursor.** Transcript events earned a cursor because they're append-only; messages are not. Compaction replaces a prefix of the list with a summary, and solver / agent code can reorder, edit, or wholesale reassign `state.messages` between reads. An index cursor over a rewritable sequence would promise the exactly-once resume it can't deliver (the phase-2 cursor design's whole job is "monotonic, gap-free" — this sequence is neither). So `messages` is a **snapshot read**: each call returns the current conversation (or a tail of it), enveloped with `as_of`, the resolved `{sample_id, epoch}`, the sample's `status`, and the total message `count`. The watch loop composes with the existing reads: poll `sample list --active-since` for "what changed", then drill into `messages` on the sample of interest; the `count` moving (or not) between polls is the cheap staleness signal. Incremental *event*-grain watching stays `sample events`' job. If a delta shape is ever wanted here, the right token is a state version invalidated by any rewrite — never an index — but that's deferred until a consumer actually needs it; for the same reason there is no phase-4 SSE `--follow` on messages (follow the events stream, or poll). + +**Summary-shaped by default.** A conversation is typically the largest object in a sample — a full multi-turn agentic `ChatMessage` list with tool outputs can exceed a watching agent's entire context. Mirroring the events projection: + +- **Default: compact projection per message** — index, message id, role, truncated text (non-text content items summarized as `[image]` / `[audio]` / ...), tool-call function names + truncated arguments on assistant messages, truncated output on tool messages. +- **`--tail N`** — the last N messages; the unseeded default is a recent tail (e.g. the last 20), per the "never an empty or overwhelming first page" rule from the events read. `--all` for the whole list. +- **`--full`** — raw `ChatMessage` JSON instead of the projection (combines with `--tail` to keep it bounded). +- **`--json`** throughout, per the agent output contract. + +**Data source: the same running-vs-terminal split as the sibling reads.** Running samples serve straight from `active_samples` → the live `TaskState.messages` — in-memory, no log involved, which is exactly why buffered-only samples work. Terminal samples serve the completed sample's messages from the recorder's buffer via `EvalState.live`, falling back to the on-disk log — the `samples` / `sample` / `events` pattern, unchanged. In the drill-down family each verb then answers one question: `sample show` — "how is it doing", `sample events` — "what happened, in order", `sample messages` — "what does the model see". + +**Scope: the sample's main conversation.** `TaskState.messages` is the sample's top-level thread. Nested agents whose conversations don't share that thread (e.g. an agent invoked as a tool, whose inner exchange never lands in `state.messages`) remain visible through `sample events` (span-tagged); a span / agent selector on `messages` is a plausible follow-on if live subagent-watching becomes a real ask — not part of the first slice. + +**Why not fold into `sample show`?** `show` is the summary; the conversation is the detail. Folding it in would break the summary-then-detail shape constraint the whole read surface is built on (and make `show` unusable as the cheap poll target). Separate verb, same relationship as `show` vs `events`. + +**Sequencing.** Read-only — no new directive machinery, no security-hardening dependency — so it extends the phase-2 read surface and can land independently of the remaining phase-3 directives. One shared-infrastructure note: the compact message projection should share its truncation helpers with the events projection (which already truncates `ModelEvent` message content) so the two renderings of the same objects don't drift. + ### Phase 3 — modification (direct) methods The first endpoints that **mutate the run**, each idempotent and supporting `?dry_run=true` / `--dry-run` from day one. Shipped so far: the log-flush and buffer-params directives and the concurrency-config directive (`max_samples` / `max_sandboxes` / `max_subprocesses` / `max_connections`) — all surfaced through `ctl task log-flush` and `ctl config`; adding a task to a running eval, then cancel / drain / requeue and the per-sample time/token/message limits follow. The Security model's "future hardening" (SO_PEERCRED UID check, self-targeting guard) lands with this phase, since it introduces the first state-mutating writes. From 256c1641dd9cb590760b7a12fbb68006442502d6 Mon Sep 17 00:00:00 2001 From: Ransom Richardson Date: Mon, 13 Jul 2026 12:11:07 -0400 Subject: [PATCH 02/13] Address review comments on the sample-messages design section - Name the new running-side plumbing: a live TaskState handle on ActiveSample, kept current by set_sample_state (the ContextVar is unreachable from the control server's task) - Correct the events-projection claims: default projection omits ModelEvent.input entirely; truncation covers completions and tool arguments/results - Reconcile --all with the selector-conventions rejection note - Align the wrapped comment in the CLI sketch Co-Authored-By: Claude Fable 5 --- design/control-channel.md | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/design/control-channel.md b/design/control-channel.md index 52fe6e37a0..a4a071a063 100644 --- a/design/control-channel.md +++ b/design/control-channel.md @@ -79,8 +79,9 @@ inspect ctl sample errors [TASK] # triage: samples that errored or we inspect ctl sample events TASK SID [EPOCH] # one sample's transcript events (cursored pull) # --cursor CURSOR / --tail N / --type / --full / --since-time / --until # (-f / --follow push is phase 4) + inspect ctl sample messages TASK SID [EPOCH] # one sample's current conversation, snapshot - # (planned — see "Sample messages read") + # (planned — see "Sample messages read") inspect ctl process list # running Inspect processes (pid / keep-alive / tasks) inspect ctl process keep [PID] # keep a running process alive after its eval finishes inspect ctl process release [PID] # release a lingering keep-alive process @@ -736,7 +737,7 @@ Unlocks watchdog agents (cursored polling). Live-render TUIs follow once phase-4 > **Status: planned** (design only — nothing in this section is implemented). Motivated by issue #70: agents live-watching a sample want the *conversation* (`.messages`), not just the transcript firehose — in particular for samples that exist only in the process buffer, not yet written to the `.eval` log. -**The gap.** A human watching a live sample opens `inspect view` and reads the conversation. An agent has no equivalent: the closest surface is `sample events`, and reconstructing "what does the conversation look like right now" from the transcript is genuinely hard — the messages ride inside `ModelEvent.input` / `output` (truncated in the default projection, enormous under `--full`), solver / agent code can assign `state.messages` without emitting any model event, and compaction rewrites the conversation while the event stream records only a `compaction` marker. Meanwhile the thing the agent wants already exists, materialized, in the running sample's `TaskState.messages` — serve it directly. The buffered case is what makes this control-channel work rather than `inspect log` work: for a running sample — or a completed one still sitting in the log buffer awaiting a flush — the conversation exists only inside the eval process, and the control channel is the only surface that can see it. +**The gap.** A human watching a live sample opens `inspect view` and reads the conversation. An agent has no equivalent: the closest surface is `sample events`, and reconstructing "what does the conversation look like right now" from the transcript is genuinely hard — the messages ride inside `ModelEvent.input` / `output` (the default projection omits the `input` messages entirely — only a truncated completion survives — and `--full` is enormous), solver / agent code can assign `state.messages` without emitting any model event, and compaction rewrites the conversation while the event stream records only a `compaction` marker. Meanwhile the thing the agent wants already exists, materialized, in the running sample's `TaskState.messages` — serve it directly. The buffered case is what makes this control-channel work rather than `inspect log` work: for a running sample — or a completed one still sitting in the log buffer awaiting a flush — the conversation exists only inside the eval process, and the control channel is the only surface that can see it. **Surface.** @@ -748,17 +749,17 @@ Unlocks watchdog agents (cursored polling). Live-render TUIs follow once phase-4 **Summary-shaped by default.** A conversation is typically the largest object in a sample — a full multi-turn agentic `ChatMessage` list with tool outputs can exceed a watching agent's entire context. Mirroring the events projection: - **Default: compact projection per message** — index, message id, role, truncated text (non-text content items summarized as `[image]` / `[audio]` / ...), tool-call function names + truncated arguments on assistant messages, truncated output on tool messages. -- **`--tail N`** — the last N messages; the unseeded default is a recent tail (e.g. the last 20), per the "never an empty or overwhelming first page" rule from the events read. `--all` for the whole list. +- **`--tail N`** — the last N messages; the unseeded default is a recent tail (e.g. the last 20), per the "never an empty or overwhelming first page" rule from the events read. `--all` for the whole list. (This is not the selector-widening `--all` rejected in "Selector conventions" — the sample is already pinned by its positional selectors, so the flag can only widen over *messages*; there is no scope ambiguity to smuggle into a flag.) - **`--full`** — raw `ChatMessage` JSON instead of the projection (combines with `--tail` to keep it bounded). - **`--json`** throughout, per the agent output contract. -**Data source: the same running-vs-terminal split as the sibling reads.** Running samples serve straight from `active_samples` → the live `TaskState.messages` — in-memory, no log involved, which is exactly why buffered-only samples work. Terminal samples serve the completed sample's messages from the recorder's buffer via `EvalState.live`, falling back to the on-disk log — the `samples` / `sample` / `events` pattern, unchanged. In the drill-down family each verb then answers one question: `sample show` — "how is it doing", `sample events` — "what happened, in order", `sample messages` — "what does the model see". +**Data source: the same running-vs-terminal split as the sibling reads — with one new plumbing piece on the running side.** Running samples serve the live `TaskState.messages` — in-memory, no log involved, which is exactly why buffered-only samples work. But unlike the sibling reads, whose running-side data sources already hang off `ActiveSample` (`transcript` for `events`, the counter fields for `show` / `list`), nothing the control server can reach holds the live `TaskState` today: it lives in the `_sample_state` ContextVar (`solver/_task_state.py`), set inside the sample's own async context — invisible from the server's request-handler task. The slice therefore adds a live-state handle on `ActiveSample`. Capturing it once at sample start isn't enough: solvers can *replace* the `TaskState` object outright via `set_sample_state` (which is why `_eval/task/run.py` re-reads `sample_state() or state` after each solver step), so `set_sample_state` must also refresh the `ActiveSample` field — the same running-sample update flow `set_active_sample_total_messages` already rides. Two consistency notes fall out: between those replacement points the conversation is mutated *in place* (the turn loop appends to the same list; even `state.messages = ...` assignment lands on the same `TaskState` object), so the handle reads live at message grain, not solver grain; and since the control server shares the eval's event loop, a snapshot copied in the request handler can never observe a half-applied append or rewrite. Terminal samples need nothing new: the completed sample's messages come from the recorder's buffer via `EvalState.live` (`read_sample`), falling back to the on-disk log — the `samples` / `sample` / `events` pattern, unchanged. In the drill-down family each verb then answers one question: `sample show` — "how is it doing", `sample events` — "what happened, in order", `sample messages` — "what does the model see". **Scope: the sample's main conversation.** `TaskState.messages` is the sample's top-level thread. Nested agents whose conversations don't share that thread (e.g. an agent invoked as a tool, whose inner exchange never lands in `state.messages`) remain visible through `sample events` (span-tagged); a span / agent selector on `messages` is a plausible follow-on if live subagent-watching becomes a real ask — not part of the first slice. **Why not fold into `sample show`?** `show` is the summary; the conversation is the detail. Folding it in would break the summary-then-detail shape constraint the whole read surface is built on (and make `show` unusable as the cheap poll target). Separate verb, same relationship as `show` vs `events`. -**Sequencing.** Read-only — no new directive machinery, no security-hardening dependency — so it extends the phase-2 read surface and can land independently of the remaining phase-3 directives. One shared-infrastructure note: the compact message projection should share its truncation helpers with the events projection (which already truncates `ModelEvent` message content) so the two renderings of the same objects don't drift. +**Sequencing.** Read-only — no new directive machinery, no security-hardening dependency — so it extends the phase-2 read surface and can land independently of the remaining phase-3 directives. One shared-infrastructure note: the compact message projection should share its truncation helpers with the events projection (which already truncates model / tool event content — completions, tool arguments / results) so the two renderings of the same objects don't drift. ### Phase 3 — modification (direct) methods From 381ef0f8ecdd32b3a4cb47981ee6c6cca405bf7c Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Mon, 13 Jul 2026 16:36:37 +0000 Subject: [PATCH 03/13] feat(control): implement inspect ctl sample messages Add the sample-conversation snapshot read designed in this PR: a new GET /evals//sample/messages endpoint and `inspect ctl sample messages TASK SID [EPOCH]` CLI verb. Running samples serve their live TaskState.messages (via a new ActiveSample.live_state handle refreshed by set_sample_state, so buffered-but-unlogged samples work); terminal samples fall back through the recorder buffer to the on-disk log. Summary-projected by default with --tail/--all/--full, deliberately snapshot (no cursor). Co-authored-by: Ransom Richardson <1209015+ransomr@users.noreply.github.com> --- CHANGELOG.md | 1 + design/control-channel.md | 14 +- src/inspect_ai/_cli/ctl.py | 221 ++++++++++++++++++- src/inspect_ai/_control/messages.py | 211 ++++++++++++++++++ src/inspect_ai/_control/server.py | 30 ++- src/inspect_ai/log/_samples.py | 30 +++ src/inspect_ai/solver/_task_state.py | 5 + tests/_control/test_ctl.py | 101 +++++++++ tests/_control/test_eval_set_integration.py | 77 +++++++ tests/_control/test_messages.py | 230 ++++++++++++++++++++ tests/_control/test_server.py | 51 +++++ 11 files changed, 961 insertions(+), 10 deletions(-) create mode 100644 src/inspect_ai/_control/messages.py create mode 100644 tests/_control/test_messages.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 6a2db1696e..9e0d0f32cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,7 @@ - Eval: `EvalSample` and `EvalSampleSummary` now record `turn_count` and the sample's token limit (`token_limit`, `token_limit_type`, and metered `token_limit_usage`). - Analysis: `samples_df` gains default `turn_count` and `token_limit_usage` columns, and `evals_df` configuration columns gain `token_limit_type`. - Control Channel: `inspect ctl sample list` now shows per-sample turn count and, when a token limit is configured, its computed usage and configured ceiling. +- Control Channel: New `inspect ctl sample messages TASK SID [EPOCH]` reads a running (or buffered-but-unlogged) sample's current conversation as a snapshot, with `--tail`/`--all`/`--full`. - Bugfix: Crash recovery now reconstructs `sample.messages` from the first model role's conversation when a solver runs multiple agents concurrently, instead of returning whichever agent's model call happened to fire last. (#4414) ## 0.3.246 (10 July 2026) diff --git a/design/control-channel.md b/design/control-channel.md index f44ec75ba6..5e7da8c429 100644 --- a/design/control-channel.md +++ b/design/control-channel.md @@ -81,7 +81,7 @@ inspect ctl sample events TASK SID [EPOCH] # one sample's transcript events (cu # (-f / --follow push is phase 4) inspect ctl sample messages TASK SID [EPOCH] # one sample's current conversation, snapshot - # (planned — see "Sample messages read") + # --tail N / --all / --full (see "Sample messages read") inspect ctl process list # running Inspect processes (pid / keep-alive / tasks) inspect ctl process keep [PID] # keep a running process alive after its eval finishes inspect ctl process release [PID] # release a lingering keep-alive process @@ -139,7 +139,7 @@ inspect ctl │ ├── show TASK SID [EPOCH] # was: sample │ ├── errors [TASK] # was: errors (no TASK = across all tasks) │ ├── events TASK SID [EPOCH] # was: events (phase-4 --follow lands here unchanged) -│ ├── messages TASK SID [EPOCH] # planned: current-conversation snapshot (see "Sample messages read") +│ ├── messages TASK SID [EPOCH] # shipped: current-conversation snapshot (see "Sample messages read") │ ├── cancel TASK SID [EPOCH] # shipped (EPOCH required when the task runs >1 epoch) │ └── requeue TASK SID [EPOCH] # planned: requeue ├── config [TASK] [...] # was: limits + buffer; view/retune launch config, scope per knob @@ -210,7 +210,7 @@ Known accepted edge in the busy handling: the sole-server rule counts *discovere | `ctl cancel-sample ` | **shipped** (as noun form only) | `ctl sample cancel TASK SID [EPOCH]` | compound verb dissolves; `EPOCH` required when the task runs >1 epoch | | `ctl requeue ` | planned | `ctl sample requeue TASK SID [EPOCH]` | `EPOCH` required when the task runs >1 epoch | | `ctl set-limit --time N` | planned | fold into `ctl config TASK --time-limit/--token-limit/--message-limit` | all retunable launch flags in one surface, same spellings as `inspect eval`; task-scoped knobs, so `TASK` follows the mutation selector rule | -| — | planned | `ctl sample messages TASK SID [EPOCH]` | new read: the sample's current conversation, summary-projected — see "Sample messages read" | +| — | **shipped** | `ctl sample messages TASK SID [EPOCH]` | new read: the sample's current conversation, summary-projected — see "Sample messages read" | | — | new | `ctl process list` | pids, keep-alive status, hosted task_ids; shipped with the reorg — the group's other verbs take a `PID` selector, so pids must be enumerable in-group | | — | later | `ctl eval-set list / show / cancel` | group slot ready-made | @@ -277,7 +277,7 @@ The conceptual surface, independent of wire protocol. Each operation becomes eit - List running evals (id, task, status, started_at, sample counts). - Eval status detail (config, current limits, sample queue depth, in-flight samples, completed counts, model usage so far). - List samples within an eval (status, started_at, current model/tool, token usage). -- Read one sample's current conversation — the live `TaskState.messages`, summary-projected (planned; see "Sample messages read"). +- Read one sample's current conversation — the live `TaskState.messages`, summary-projected (see "Sample messages read"). **Read (eval-set-level)** - List eval-sets. @@ -355,7 +355,7 @@ The URL scheme has one rule — **three scopes, three roots**: process-scoped op | Samples changed since (recency delta) | `GET /evals//samples?active_since=` | 2 ✅ | | Sample transcript events (pull) | `GET /evals//sample/events?sample_id=&epoch=&since=` (JSON) | 2 ✅ | | Sample transcript events (push) | the pull URL with `Accept: text/event-stream` (SSE) | 4 | -| Sample conversation messages (snapshot) | `GET /evals//sample/messages?sample_id=&epoch=` | planned | +| Sample conversation messages (snapshot) | `GET /evals//sample/messages?sample_id=&epoch=` | 2 ✅ | | Eval-wide transcript fan-in (push only) | `GET /evals//samples/events` (SSE) | 4 | | Flush buffered samples to the log | `POST /tasks//log-flush` | 3 ✅ | | Read / modify retunable config (concurrency limits, buffer params) | `GET`+`PATCH /config` (process) and `/tasks//config` (task) | 3 ✅ (max-samples / max-sandboxes / max-subprocesses / max-connections / log-buffer / log-shared) | @@ -738,9 +738,9 @@ A wall-clock window is a snapshot query (no exactly-once requirement), which is Unlocks watchdog agents (cursored polling). Live-render TUIs follow once phase-4 push lands. -### Sample messages read (planned) +### Sample messages read (shipped) -> **Status: planned** (design only — nothing in this section is implemented). Motivated by issue #70: agents live-watching a sample want the *conversation* (`.messages`), not just the transcript firehose — in particular for samples that exist only in the process buffer, not yet written to the `.eval` log. +> **Status: shipped** (extends the phase-2 read surface). Motivated by issue #70: agents live-watching a sample want the *conversation* (`.messages`), not just the transcript firehose — in particular for samples that exist only in the process buffer, not yet written to the `.eval` log. **The gap.** A human watching a live sample opens `inspect view` and reads the conversation. An agent has no equivalent: the closest surface is `sample events`, and reconstructing "what does the conversation look like right now" from the transcript is genuinely hard — the messages ride inside `ModelEvent.input` / `output` (the default projection omits the `input` messages entirely — only a truncated completion survives — and `--full` is enormous), solver / agent code can assign `state.messages` without emitting any model event, and compaction rewrites the conversation while the event stream records only a `compaction` marker. Meanwhile the thing the agent wants already exists, materialized, in the running sample's `TaskState.messages` — serve it directly. The buffered case is what makes this control-channel work rather than `inspect log` work: for a running sample — or a completed one still sitting in the log buffer awaiting a flush — the conversation exists only inside the eval process, and the control channel is the only surface that can see it. diff --git a/src/inspect_ai/_cli/ctl.py b/src/inspect_ai/_cli/ctl.py index 00e2de33ac..35c30aa9aa 100644 --- a/src/inspect_ai/_cli/ctl.py +++ b/src/inspect_ai/_cli/ctl.py @@ -12,7 +12,7 @@ ``drain`` are planned. - ``sample`` — one sample (``TASK SAMPLE_ID [EPOCH]``) or a task's samples: ``list`` (implied by the bare noun), ``show``, ``errors``, ``events``, - ``cancel``; ``requeue`` is planned. + ``messages``, ``cancel``; ``requeue`` is planned. - ``config`` — a top-level *command* (not a group): view / retune launch configuration mid-flight (concurrency limits, log buffering). Scope is a property of each knob (task vs process), labeled in the output. @@ -57,6 +57,12 @@ # output contract in design/control-channel.md). _DEFAULT_EVENTS_TAIL = 20 +# Messages shown on an unseeded `sample messages` read (no --tail / --all): a +# recent tail rather than the whole conversation, which for a long agentic run +# can exceed a watching agent's context. Same "never empty, never a flood" +# rationale as the events tail. +_DEFAULT_MESSAGES_TAIL = 20 + # One source of truth for each retunable config knob's scope. The `ctl config` # option help tags, the composed JSON view's per-knob "scope" labels, and the # human rendering's [task]/[process] labels all derive from this table, so a @@ -571,6 +577,69 @@ def sample_events_command( ) +@sample_group.command("messages") +@click.argument("task") +@click.argument("sample_id") +@click.argument("epoch", required=False, type=int, default=1) +@click.option( + "--tail", + type=int, + default=None, + help=( + "Only the last N messages (default: the recent tail — " + f"{_DEFAULT_MESSAGES_TAIL}). Use --all for the whole conversation." + ), +) +@click.option( + "--all", + "show_all", + is_flag=True, + default=False, + help="Show the whole conversation instead of a recent tail.", +) +@click.option( + "--full", + is_flag=True, + default=False, + help="Return raw ChatMessage JSON instead of the compact summary.", +) +@click.option( + "--json", + "as_json", + is_flag=True, + default=False, + help="Output as JSON (the `{as_of, status, count, messages}` envelope).", +) +def sample_messages_command( + task: str, + sample_id: str, + epoch: int, + tail: int | None, + show_all: bool, + full: bool, + as_json: bool, +) -> None: + """Read one sample's current conversation (a snapshot). + + Returns the sample's `TaskState.messages` as they look right now — a + snapshot, not a stream: the message list is rewritable (compaction, + solver edits), so there is no resume cursor. The default is a recent tail; + pass `--all` for the whole conversation or `--tail N` for a specific + window, and `--full` for raw `ChatMessage` JSON. For incremental, + event-grain watching use `inspect ctl sample events`. EPOCH defaults to 1 + (the response echoes the resolved epoch). + """ + _run_sample_messages( + task, + sample_id, + epoch, + tail=tail, + show_all=show_all, + full=full, + as_json=as_json, + ) + + @sample_group.command("cancel") @click.argument("task") @click.argument("sample_id") @@ -1554,6 +1623,78 @@ def _run_sample_events( _print_events(page) +@_envelope_failures +def _run_sample_messages( + task: str, + sample_id: str, + epoch: int, + *, + tail: int | None, + show_all: bool, + full: bool, + as_json: bool, +) -> None: + # `--all` and `--tail` are mutually exclusive ways to size the window; + # an explicit --tail with --all is contradictory, so reject it rather + # than silently letting one win. + if show_all and tail is not None: + with _structured_failures(as_json): + _fail( + "invalid_request", + "--all and --tail are mutually exclusive (--all shows every " + "message; --tail sizes a recent window).", + ) + # The unseeded default is a recent tail — never an overwhelming first + # page. --all disables it; an explicit --tail overrides it. + if not show_all and tail is None: + tail = _DEFAULT_MESSAGES_TAIL + + fetched = _fetch_sample_summaries() + summaries = fetched.summaries + if not summaries: + if as_json: + # Uniform --json shape even on the empty page (task_id is + # unresolvable with no running evals). + empty_page: dict[str, Any] = { + "task_id": None, + "sample_id": sample_id, + "epoch": epoch, + "status": None, + "count": 0, + "messages": [], + } + click.echo(json_lib.dumps(empty_page, indent=2)) + return + _echo_no_running_evals() + return + + target = _resolve_target_eval(summaries, task, busy_pids=fetched.busy_pids) + page = _fetch_sample_messages( + target["socket_path"], + target["eval_id"], + sample_id, + epoch, + tail=tail, + full=full, + ) + # Echo the resolved identifiers so a defaulted epoch is visible and the + # row round-trips into other commands' selectors. + page = { + "task_id": target.get("task_id"), + "sample_id": sample_id, + "epoch": epoch, + **page, + } + + if as_json: + click.echo(json_lib.dumps(page, indent=2)) + return + + click.echo(_task_header(target)) + click.echo() + _print_messages(page, full=full) + + def _looks_like_timestamp(value: str) -> bool: """Whether a would-be cursor value reads as a unix timestamp. @@ -2880,6 +3021,38 @@ def _fetch_sample_events( ) +def _fetch_sample_messages( + socket_path: str, + eval_id: str, + sample_id: str, + epoch: int, + *, + tail: int | None, + full: bool, +) -> dict[str, Any]: + """Query one control server for a snapshot of a sample's conversation. + + The authoritative read behind ``sample messages``: like the sample detail + and events reads, it rides the full narrated busy-retry policy rather than + failing on a momentary event-loop stall. + """ + # sample_id (and all params) go in the query string so reserved-char ids + # address correctly; drop unset options so server defaults apply. + params: dict[str, Any] = {"sample_id": sample_id, "epoch": epoch, "full": full} + if tail is not None: + params["tail"] = tail + return _request_json( + socket_path, + f"/evals/{eval_id}/sample/messages", + params=params, + what=f"messages for sample {sample_id}", + not_found=( + f"Sample '{sample_id}' (epoch {epoch}) not found — it may " + "not have started or not yet been written to the log." + ), + ) + + def _request_json( socket_path: str, path: str, @@ -3291,6 +3464,52 @@ def _print_events(page: dict[str, Any]) -> None: click.echo(f"next: {nxt} (resume with --cursor)") +def _print_messages(page: dict[str, Any], *, full: bool) -> None: + """Render a conversation snapshot (per-message rows) plus a count footer.""" + messages = page.get("messages") or [] + count = int(page.get("count") or 0) + status = page.get("status") + + if full: + # Raw mode is for machine consumption; the human rendering is the + # compact projection, so just pretty-print the raw messages. + click.echo(json_lib.dumps(messages, indent=2)) + elif not messages: + click.echo("(no messages)") + else: + rows: list[tuple[str, ...]] = [] + for m in messages: + rows.append( + ( + str(m.get("index", "")), + str(m.get("role", "") or ""), + _message_summary(m), + ) + ) + _render_table(("#", "role", "content"), rows) + + shown = len(messages) + footer = f"{shown} of {count} message" + ("" if count == 1 else "s") + if shown < count: + footer += " (use --all for the whole conversation)" + if status: + footer += f" · {status}" + click.echo() + click.echo(footer) + + +def _message_summary(m: dict[str, Any]) -> str: + """One-line summary for a message row (best-effort over compact fields).""" + parts = [str(m.get("content") or "")] + for call in m.get("tool_calls") or []: + parts.append( + f"→ {call.get('function') or '?'}({_truncate(str(call.get('arguments') or ''), 30)})" + ) + if m.get("error"): + parts.append(f"error: {m['error']}") + return _truncate(" ".join(p for p in parts if p), 100) + + def _event_summary(e: dict[str, Any]) -> str: """One-line summary for an event row (best-effort over compact fields).""" t = e.get("event") diff --git a/src/inspect_ai/_control/messages.py b/src/inspect_ai/_control/messages.py new file mode 100644 index 0000000000..1a4f9a81d8 --- /dev/null +++ b/src/inspect_ai/_control/messages.py @@ -0,0 +1,211 @@ +"""One sample's current conversation for the control channel. + +Backs ``GET /evals//sample/messages`` (and ``inspect ctl sample +messages``): a **snapshot** of one sample's ``TaskState.messages`` — the +top-level conversation as it looks right now — read from the live +``TaskState`` while the sample is running, and once terminal from the +recorder's sample (buffer, then on-disk log). + +Unlike ``sample events``, this is deliberately *not* cursored. The message +list is rewritable — compaction replaces a prefix with a summary, and solver / +agent code can edit or wholesale-reassign ``state.messages`` — so an index +cursor over it could not deliver the exactly-once resume the event cursor +promises. Each call returns the whole conversation (or a recent tail), +enveloped with ``as_of`` / the sample ``status`` / the total ``count``; a +watcher polls (the moving ``count`` is the cheap staleness signal) or follows +``sample events`` for incremental, event-grain watching. + +See ``design/control-channel.md`` ("Sample messages read") for the full +rationale. +""" + +from __future__ import annotations + +import time +from typing import TYPE_CHECKING, Any, NamedTuple + +# The compact message projection shares its truncation helpers with the events +# projection so the two renderings of the same underlying objects can't drift. +from inspect_ai._control.events import _to_text, _truncate + +if TYPE_CHECKING: + from inspect_ai.model._chat_message import ChatMessage + + +class MessagesSource(NamedTuple): + """One resolvable source of a sample's conversation. + + Produced by :func:`_running_source` (live ``TaskState``) and + :func:`_logged_source` (recorder / on-disk log); consumed by + :func:`sample_messages`. + """ + + messages: list["ChatMessage"] + """The sample's current conversation (a snapshot copy).""" + + status: str + """``running`` / ``completed`` / ``error``.""" + + +async def sample_messages( + eval_id: str, + sample_id: str, + epoch: int, + *, + tail: int | None = None, + full: bool = False, +) -> dict[str, Any] | None: + """A snapshot of one sample's current conversation. + + Returns an ``{as_of, status, count, messages}`` envelope (see the module + docstring), or ``None`` when the eval/sample isn't found in this process — + the endpoint turns that into a 404. + + Args: + eval_id: The eval's id. + sample_id: The sample's id (string; matched against running + logged). + epoch: The sample epoch. + tail: Only the last ``tail`` messages (``None`` = the whole list). + full: Raw serialized ``ChatMessage`` objects instead of the compact + projection. + """ + # `as_of` is stamped before the read so a client comparing successive + # `count`s can't miss a change that lands mid-read. + as_of = time.time() + + source = _running_source(eval_id, sample_id, epoch) + if source is None: + source = await _logged_source(eval_id, sample_id, epoch) + if source is None: + return None + + messages, status = source + count = len(messages) + + # `tail` selects a recent window; the projection still reports each + # message's absolute index so a tailed view lines up with the full one. + start = max(0, count - tail) if tail is not None else 0 + projected = [ + _project(message, index, full) + for index, message in enumerate(messages[start:], start=start) + ] + + return { + "as_of": as_of, + "status": status, + "count": count, + "messages": projected, + } + + +# --- sources --------------------------------------------------------------- + + +def _running_source(eval_id: str, sample_id: str, epoch: int) -> MessagesSource | None: + """The live source for a sample, or ``None`` if it isn't running here. + + Reads the sample's live ``TaskState.messages`` off ``ActiveSample + .live_state`` — an in-memory snapshot, no log involved, which is why a + sample that exists only in the process buffer still has a conversation to + serve. The control server shares the eval's event loop, so copying the + list here can never observe a half-applied append or rewrite. + """ + from inspect_ai._control.state import find_active_sample + + s = find_active_sample(eval_id, sample_id, epoch) + if s is None or s.live_state is None: + return None + return MessagesSource( + messages=list(s.live_state.messages), + status="completed" if s.completed is not None else "running", + ) + + +async def _logged_source( + eval_id: str, sample_id: str, epoch: int +) -> MessagesSource | None: + """The terminal source for a sample (recorder buffer, then on-disk log). + + ``None`` when the eval/sample isn't available here. Reads the full + ``EvalSample`` via :func:`inspect_ai._control.state._full_sample` — the + same gap-free recorder-then-log source the error-detail and events reads + use — and resolves its content attachments so image/large-text refs render + as real content rather than ``attachment://`` placeholders. + """ + from inspect_ai._control.state import _full_sample + + sample = await _full_sample(eval_id, sample_id, epoch) + if sample is None: + return None + + from inspect_ai.log._condense import resolve_sample_attachments + + sample = resolve_sample_attachments(sample, "full") + return MessagesSource( + messages=list(sample.messages or []), + status="error" if sample.error is not None else "completed", + ) + + +# --- projection ------------------------------------------------------------ + + +def _project(message: "ChatMessage", index: int, full: bool) -> dict[str, Any]: + """Raw serialized message (``full``) or a compact, context-cheap summary. + + The compact form carries the message's index / id / role plus a truncated + text rendering (non-text content items summarized as ``[image]`` / + ``[audio]`` / …); assistant messages add their tool-call function names and + truncated arguments, tool messages their truncated output and any error. + """ + if full: + out = message.model_dump(mode="json") + out["index"] = index + return out + + projected: dict[str, Any] = { + "index": index, + "id": message.id, + "role": message.role, + "content": _content_summary(message), + } + if message.role == "assistant": + tool_calls = getattr(message, "tool_calls", None) + if tool_calls: + projected["tool_calls"] = [ + { + "id": call.id, + "function": call.function, + "arguments": _truncate(_to_text(call.arguments)), + } + for call in tool_calls + ] + elif message.role == "tool": + projected["function"] = getattr(message, "function", None) + tool_error = getattr(message, "error", None) + projected["error"] = ( + getattr(tool_error, "message", None) if tool_error else None + ) + return projected + + +def _content_summary(message: "ChatMessage") -> str: + """Truncated text of a message, with non-text content items summarized. + + A string content renders (truncated) directly; a list of content items + renders each text/reasoning item's text and each non-text item as a + ``[kind]`` placeholder, so an image-bearing message reads as + ``… [image] …`` rather than a wall of base64. + """ + content = message.content + if isinstance(content, str): + return _truncate(content) + + parts: list[str] = [] + for item in content: + item_type = getattr(item, "type", None) + if item_type in ("text", "reasoning"): + parts.append(getattr(item, item_type, "") or "") + else: + parts.append(f"[{item_type or 'content'}]") + return _truncate(" ".join(p for p in parts if p)) diff --git a/src/inspect_ai/_control/server.py b/src/inspect_ai/_control/server.py index 5b1f493de2..54bc2ee234 100644 --- a/src/inspect_ai/_control/server.py +++ b/src/inspect_ai/_control/server.py @@ -16,8 +16,9 @@ Current scope is the phase 1-2 read surface — ``GET /tasks`` (per-task summaries), ``GET /evals/{id}/samples`` (sample listing, with an ``active_since`` recency delta), ``GET /evals/{id}/sample`` (error -detail), and ``GET /evals/{id}/sample/events`` (cursored transcript -pull) — plus ``POST /release`` / ``POST /keep`` for keep-alive control +detail), ``GET /evals/{id}/sample/events`` (cursored transcript +pull), and ``GET /evals/{id}/sample/messages`` (conversation snapshot) — +plus ``POST /release`` / ``POST /keep`` for keep-alive control and the first phase-3 directives: the config/log-flush mutations and ``POST /tasks/{id}/cancel`` / ``POST /evals/{id}/sample/cancel``. The remaining directives (drain / requeue / add-task) and SSE push land @@ -42,6 +43,7 @@ from inspect_ai._control.discovery import default_socket_path, discovery_dir from inspect_ai._control.events import sample_events from inspect_ai._control.limits import process_limits, task_limits +from inspect_ai._control.messages import sample_messages from inspect_ai._control.state import ( current_eval_summaries, current_sample_summaries, @@ -360,6 +362,30 @@ async def get_sample_events( ) return page + # Per-sample conversation snapshot (`TaskState.messages`). Like the + # other per-sample routes, `sample_id` is a query param (ids may carry + # URL-reserved characters). Deliberately not cursored — the message + # list is rewritable (compaction / solver edits), so each call returns + # the current conversation (or a `tail`), enveloped with `as_of` / + # `status` / `count`. `full` returns raw ChatMessage JSON. + @app.get("/evals/{eval_id}/sample/messages") + async def get_sample_messages( + eval_id: str, + sample_id: str, + epoch: int = 1, + tail: int | None = None, + full: bool = False, + ) -> Any: + page = await sample_messages( + eval_id, sample_id, epoch, tail=tail, full=full + ) + if page is None: + return JSONResponse( + status_code=404, + content={"error": f"sample {sample_id} (epoch {epoch}) not found"}, + ) + return page + # Flush the task's buffered completed samples to the (possibly remote, # eg. S3) log now, so they're readable without waiting for the flush # buffer to fill. Keyed by task_id (resolved to the latest attempt), diff --git a/src/inspect_ai/log/_samples.py b/src/inspect_ai/log/_samples.py index 0ae5a51cbf..1b70761dfc 100644 --- a/src/inspect_ai/log/_samples.py +++ b/src/inspect_ai/log/_samples.py @@ -20,6 +20,7 @@ from inspect_ai.hooks._hooks import SampleEvent from inspect_ai.log._log import EvalRetryError from inspect_ai.model._model_call import ModelCall, ModelCallFilter + from inspect_ai.solver._task_state import TaskState import anyio from anyio.abc import TaskGroup @@ -92,6 +93,15 @@ def __init__( self.total_messages = 0 self.total_tokens = 0 self.total_turns = 0 + # The sample's live `TaskState`, so the control channel can serve the + # current conversation (`TaskState.messages`) without reaching into the + # context-bound `_sample_state` ContextVar (unreachable from the control + # server's request-handler task). Captured when the sample starts and + # refreshed by `set_sample_state` (via `set_active_sample_state`) so it + # survives a solver replacing the `TaskState` object outright — mirrors + # the `sample_state() or state` re-read the task runner does. `None` + # only in the brief window before the first state is set. + self.live_state: "TaskState | None" = None self.token_limit_usage: int | None = None self.total_cost: float | None = None self.fallback_models: list[str] = [] @@ -316,6 +326,12 @@ async def active_sample( _active_samples.append(active) _sample_active.set(active) + # Capture the state the runner set before entering this context (via + # `set_sample_state`, which precedes `active_sample`); subsequent solver + # reassignments refresh it through `set_active_sample_state`. + from inspect_ai.solver._task_state import sample_state + + active.live_state = sample_state() # Open the ACP session for this sample's lifetime. The session is the # ACP-specific transport layer (pub/sub, approver registry, transcript # snapshot, etc.); it produces into whatever agent_channel() is bound to @@ -432,6 +448,20 @@ def set_active_sample_total_messages(total_messages: int) -> None: active.total_messages = total_messages +def set_active_sample_state(state: "TaskState") -> None: + """Refresh the active sample's live `TaskState` handle. + + Called from `set_sample_state` so the control channel's view of the + conversation survives a solver replacing the `TaskState` object outright + (the `sample_state() or state` re-read in the task runner). In-place + mutation between replacements needs no hook — the handle already points at + the object whose `messages` list is being appended to. + """ + active = sample_active() + if active: + active.live_state = state + + def set_active_sample_fallback_models(fallback_models: list[str]) -> None: active = sample_active() if active: diff --git a/src/inspect_ai/solver/_task_state.py b/src/inspect_ai/solver/_task_state.py index da26f33db3..9088bde7d9 100644 --- a/src/inspect_ai/solver/_task_state.py +++ b/src/inspect_ai/solver/_task_state.py @@ -473,6 +473,11 @@ def sample_state() -> TaskState | None: def set_sample_state(state: TaskState) -> None: _sample_state.set(state) + # Keep the control channel's live-conversation handle current when a solver + # replaces the TaskState object (see `ActiveSample.live_state`). + from inspect_ai.log._samples import set_active_sample_state + + set_active_sample_state(state) _sample_state: ContextVar[TaskState] = ContextVar("sample_state") diff --git a/tests/_control/test_ctl.py b/tests/_control/test_ctl.py index eb558f72ae..c1ed881a95 100644 --- a/tests/_control/test_ctl.py +++ b/tests/_control/test_ctl.py @@ -1784,6 +1784,107 @@ def test_events_json_no_servers_echoes_identifiers( assert payload["events"] == [] and payload["next"] is None and payload["done"] +def test_print_messages_table_and_footer( + capsys: pytest.CaptureFixture[str], +) -> None: + from inspect_ai._cli.ctl import _print_messages + + page = { + "status": "running", + "count": 5, + "messages": [ + {"index": 3, "role": "user", "content": "what is the weather?"}, + { + "index": 4, + "role": "assistant", + "content": "let me check", + "tool_calls": [{"function": "search", "arguments": "weather"}], + }, + ], + } + _print_messages(page, full=False) + out = capsys.readouterr().out + assert "role" in out.splitlines()[0] # table header + assert "what is the weather?" in out and "search" in out + # footer: shown-of-total, the --all hint (a tail was shown), and status + assert "2 of 5 messages" in out + assert "--all" in out + assert "running" in out + + +def test_print_messages_empty(capsys: pytest.CaptureFixture[str]) -> None: + from inspect_ai._cli.ctl import _print_messages + + _print_messages({"status": "completed", "count": 0, "messages": []}, full=False) + out = capsys.readouterr().out + assert "(no messages)" in out + # nothing withheld, so no --all hint + assert "--all" not in out + + +def test_messages_unseeded_defaults_to_recent_tail( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The first messages page is a recent tail, not the whole conversation.""" + captured: dict[str, Any] = {} + + def fake_messages( + socket_path: Any, eval_id: str, sample_id: str, epoch: int, **kwargs: Any + ) -> dict[str, Any]: + captured.clear() + captured.update(kwargs) + return {"as_of": 1.0, "status": "running", "count": 0, "messages": []} + + _patch_surface(monkeypatch, [_full_summary("aaa111", "t1")]) + monkeypatch.setattr("inspect_ai._cli.ctl._fetch_sample_messages", fake_messages) + runner = cli_runner() + + result = runner.invoke( + ctl_command, ["sample", "messages", "aaa111", "s1", "--json"] + ) + assert result.exit_code == 0, result.output + assert captured["tail"] == 20 + + # --all disables the default tail (whole conversation) + runner.invoke(ctl_command, ["sample", "messages", "aaa111", "s1", "--all"]) + assert captured["tail"] is None + + # an explicit --tail overrides the default + runner.invoke(ctl_command, ["sample", "messages", "aaa111", "s1", "--tail", "3"]) + assert captured["tail"] == 3 + + # the resolved identifiers are echoed on the page + result = runner.invoke( + ctl_command, ["sample", "messages", "aaa111", "s1", "--json"] + ) + payload = json.loads(result.stdout) + assert payload["task_id"] == "aaa111" + assert (payload["sample_id"], payload["epoch"]) == ("s1", 1) + + +def test_messages_all_and_tail_are_mutually_exclusive() -> None: + result = cli_runner().invoke( + ctl_command, ["sample", "messages", "t", "s1", "--all", "--tail", "5"] + ) + assert result.exit_code != 0 + assert "mutually exclusive" in result.stderr + + +def test_messages_json_no_servers_echoes_identifiers( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The no-running-evals empty page keeps the identifier echo shape.""" + monkeypatch.setattr("inspect_ai._cli.ctl.list_discovered_servers", lambda: []) + result = cli_runner().invoke( + ctl_command, ["sample", "messages", "aaa111", "s1", "--json"] + ) + assert result.exit_code == 0, result.output + payload = json.loads(result.stdout) + assert payload["task_id"] is None + assert (payload["sample_id"], payload["epoch"]) == ("s1", 1) + assert payload["messages"] == [] and payload["count"] == 0 + + def test_group_option_before_verb_forwards(monkeypatch: pytest.MonkeyPatch) -> None: """A mirrored option given at the group level reaches the explicit verb.""" _patch_surface(monkeypatch, [_full_summary("aaa111", "t1")]) diff --git a/tests/_control/test_eval_set_integration.py b/tests/_control/test_eval_set_integration.py index ff28260c45..0a37b0cf93 100644 --- a/tests/_control/test_eval_set_integration.py +++ b/tests/_control/test_eval_set_integration.py @@ -38,6 +38,7 @@ from inspect_ai._control.discovery import list_discovered_servers from inspect_ai._control.eval_state import get_eval_states from inspect_ai._control.events import decode_cursor, sample_events +from inspect_ai._control.messages import sample_messages from inspect_ai._control.state import ( current_eval_summaries, current_sample_summaries, @@ -2019,6 +2020,82 @@ async def capture() -> dict: assert any(r["sample_id"] == 1 for r in res["active_all"]) +# --- messages / GET /evals//sample/messages ---------------------------- + + +def test_ctl_messages_snapshots_running_sample_conversation( + short_data_dir: Path, +) -> None: + """A running sample's live ``TaskState.messages`` are readable as a snapshot. + + A solver runs ``generate()`` (appending the assistant reply to the + conversation) then parks in flight; the messages read must surface the live + conversation from ``ActiveSample.live_state`` — no log involved — with the + running ``status`` and the total ``count``, ``--tail`` must window from the + end, and the compact projection must carry role + content per message. + """ + + @solver + def gen_then_park() -> Solver: + async def solve(state: TaskState, generate: Generate) -> TaskState: + state = await generate(state) + await park_now() # hold the sample in flight after a model call + return state + + return solve + + @task + def task_one() -> Task: + return Task( + dataset=[Sample(id=1, input="hello", target="ok")], + solver=[gen_then_park()], + name="task_one", + ) + + log_dir = str(short_data_dir / "logs") + Path(log_dir).mkdir() + + async def ready() -> bool: + evals = await current_eval_summaries(0.0) + return bool(evals) and evals[0]["samples"]["in_flight"] == 1 + + async def capture() -> dict: + eid = (await current_eval_summaries(0.0))[0]["eval_id"] + return { + "page": await sample_messages(eid, "1", 1), + "tail": await sample_messages(eid, "1", 1, tail=1), + } + + with probe(ready, capture) as p: + eval_set( + tasks=[task_one()], + log_dir=log_dir, + model="mockllm/model", + retry_attempts=0, + ) + + res = p.result + assert res is not None, "sample never reached in-flight after generate()" + + page = res["page"] + assert page is not None + assert set(page) >= {"as_of", "status", "count", "messages"} + # served live from TaskState.messages while the sample is parked in flight + assert page["status"] == "running" + # user prompt + assistant reply from generate() + assert page["count"] >= 2 + roles = [m["role"] for m in page["messages"]] + assert "user" in roles and "assistant" in roles + assert all("content" in m and "index" in m for m in page["messages"]) + + # --tail windows from the end: one message, count unchanged, absolute index + tail = res["tail"] + assert tail is not None + assert len(tail["messages"]) == 1 + assert tail["count"] == page["count"] + assert tail["messages"][0]["index"] == page["count"] - 1 + + # Cross-sample channel for the transition test below: the subject sample (id 1) # stashes the cursor it was handed while running; the observer (id 2) resumes it # once the subject is terminal. Module-level because the two run as separate diff --git a/tests/_control/test_messages.py b/tests/_control/test_messages.py new file mode 100644 index 0000000000..fd12e17bfa --- /dev/null +++ b/tests/_control/test_messages.py @@ -0,0 +1,230 @@ +"""Unit tests for the control-channel per-sample messages helpers. + +The per-message projection is a pure function over `ChatMessage`s, exercised +directly. The end-to-end `sample_messages` (live `TaskState` vs terminal +recorder/log source, tail windowing) is exercised by monkeypatching the two +sources the way `test_events.py` does. +""" + +from types import SimpleNamespace +from typing import Any + +import pytest +from test_helpers.live_eval_data import FakeLiveEvalData + +from inspect_ai._control.messages import _content_summary, _project, sample_messages +from inspect_ai._util.content import ContentImage, ContentText +from inspect_ai.model._chat_message import ( + ChatMessageAssistant, + ChatMessageSystem, + ChatMessageTool, + ChatMessageUser, +) +from inspect_ai.tool._tool_call import ToolCall, ToolCallError + +# --- projection ----------------------------------------------------------- + + +def test_project_compact_user_message() -> None: + out = _project(ChatMessageUser(id="m1", content="hello there"), 0, full=False) + assert out["index"] == 0 + assert out["id"] == "m1" + assert out["role"] == "user" + assert out["content"] == "hello there" + # a plain user message has no tool-call / tool fields + assert "tool_calls" not in out + + +def test_project_compact_assistant_with_tool_calls() -> None: + message = ChatMessageAssistant( + id="a1", + content="calling a tool", + tool_calls=[ + ToolCall(id="c1", function="search", arguments={"query": "weather"}) + ], + ) + out = _project(message, 3, full=False) + assert out["role"] == "assistant" + assert out["content"] == "calling a tool" + [call] = out["tool_calls"] + assert call["function"] == "search" + assert "weather" in call["arguments"] + + +def test_project_compact_tool_message_with_error() -> None: + message = ChatMessageTool( + id="t1", + content="stack trace…", + function="search", + error=ToolCallError(type="unknown", message="boom"), + ) + out = _project(message, 4, full=False) + assert out["role"] == "tool" + assert out["function"] == "search" + assert out["error"] == "boom" + + +def test_project_full_is_raw_dump() -> None: + out = _project(ChatMessageUser(id="m1", content="hi"), 2, full=True) + # raw form keeps the full model dump, plus the injected index + assert out["index"] == 2 + assert out["role"] == "user" + assert out["content"] == "hi" + + +def test_content_summary_summarizes_non_text_items() -> None: + message = ChatMessageUser( + content=[ + ContentText(text="look at this"), + ContentImage(image="data:image/png;base64,AAAA"), + ] + ) + summary = _content_summary(message) + assert "look at this" in summary + # the image is summarized, not dumped as base64 + assert "[image]" in summary + assert "base64" not in summary + + +def test_content_summary_truncates_long_text() -> None: + message = ChatMessageUser(content="x" * 1000) + summary = _content_summary(message) + assert len(summary) < 1000 + assert summary.endswith("…") + + +# --- running source (live TaskState) -------------------------------------- + + +def _fake_running_sample(messages: list[Any], *, completed: bool = False) -> Any: + """A minimal stand-in for an in-flight ``ActiveSample`` carrying a state.""" + return SimpleNamespace( + eval_id="e1", + epoch=1, + sample=SimpleNamespace(id=1), + live_state=SimpleNamespace(messages=messages), + completed=object() if completed else None, + ) + + +async def test_running_sample_serves_live_messages( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import inspect_ai.log._samples as samples_mod + + messages = [ + ChatMessageSystem(content="be helpful"), + ChatMessageUser(content="hi"), + ChatMessageAssistant(content="hello"), + ] + monkeypatch.setattr( + samples_mod, "active_samples", lambda: [_fake_running_sample(messages)] + ) + + page = await sample_messages("e1", "1", 1) + assert page is not None + assert page["status"] == "running" + assert page["count"] == 3 + assert [m["role"] for m in page["messages"]] == ["system", "user", "assistant"] + # indices are absolute + assert [m["index"] for m in page["messages"]] == [0, 1, 2] + + +async def test_running_sample_tail_windows_from_the_end( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import inspect_ai.log._samples as samples_mod + + messages = [ChatMessageUser(content=f"m{i}") for i in range(10)] + monkeypatch.setattr( + samples_mod, "active_samples", lambda: [_fake_running_sample(messages)] + ) + + page = await sample_messages("e1", "1", 1, tail=3) + assert page is not None + # count is the full conversation length; only the tail is projected, with + # its absolute indices preserved + assert page["count"] == 10 + assert [m["index"] for m in page["messages"]] == [7, 8, 9] + assert [m["content"] for m in page["messages"]] == ["m7", "m8", "m9"] + + +async def test_missing_sample_returns_none(monkeypatch: pytest.MonkeyPatch) -> None: + import inspect_ai._control.state as state_mod + import inspect_ai.log._samples as samples_mod + + monkeypatch.setattr(samples_mod, "active_samples", lambda: []) + + async def no_sample(*args: Any, **kwargs: Any) -> Any: + return None + + monkeypatch.setattr(state_mod, "_full_sample", no_sample) + + assert await sample_messages("e1", "nope", 1) is None + + +# --- terminal source (recorder / log) ------------------------------------- + + +async def test_terminal_sample_serves_logged_messages( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import inspect_ai.log._samples as samples_mod + from inspect_ai._control.eval_state import clear_all_eval_states, register_eval + from inspect_ai.log._log import EvalSample + + monkeypatch.setattr(samples_mod, "active_samples", lambda: []) + + sample = EvalSample( + id="s1", + epoch=1, + input="question", + target="answer", + messages=[ + ChatMessageUser(content="question"), + ChatMessageAssistant(content="answer"), + ], + ) + + async def read_sample(id: Any, epoch: int, *, exclude_fields: Any = None) -> Any: + return sample if str(id) == "s1" and epoch == 1 else None + + try: + register_eval("e1", 1, live=FakeLiveEvalData(sample=read_sample)) + page = await sample_messages("e1", "s1", 1) + assert page is not None + assert page["status"] == "completed" + assert page["count"] == 2 + assert [m["role"] for m in page["messages"]] == ["user", "assistant"] + finally: + clear_all_eval_states() + + +async def test_terminal_errored_sample_reports_error_status( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import inspect_ai.log._samples as samples_mod + from inspect_ai._control.eval_state import clear_all_eval_states, register_eval + from inspect_ai.log._log import EvalError, EvalSample + + monkeypatch.setattr(samples_mod, "active_samples", lambda: []) + + sample = EvalSample( + id="s1", + epoch=1, + input="q", + target="t", + messages=[ChatMessageUser(content="q")], + error=EvalError(message="boom", traceback="", traceback_ansi=""), + ) + + async def read_sample(id: Any, epoch: int, *, exclude_fields: Any = None) -> Any: + return sample + + try: + register_eval("e1", 1, live=FakeLiveEvalData(sample=read_sample)) + page = await sample_messages("e1", "s1", 1) + assert page is not None + assert page["status"] == "error" + finally: + clear_all_eval_states() diff --git a/tests/_control/test_server.py b/tests/_control/test_server.py index b601656658..ac3bc7bf96 100644 --- a/tests/_control/test_server.py +++ b/tests/_control/test_server.py @@ -415,6 +415,57 @@ async def _fake( assert missing.status_code == 404 +async def test_sample_messages_endpoint_round_trips_and_404( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The sample-messages route passes `tail`/`full`, round-trips a reserved id, 404s. + + `GET /evals//sample/messages`: `sample_id` as a query param (so + reserved chars address), `tail`/`full` forwarded, and a missing sample → + 404. The helper logic is unit-tested in test_messages.py; this pins the + route wiring. + """ + from inspect_ai._control import server as server_mod + + seen: dict[str, object] = {} + + async def _fake( + eval_id: str, + sample_id: str, + epoch: int, + *, + tail: object, + full: object, + ) -> dict[str, object] | None: + seen["sample_id"] = sample_id + seen["tail"] = tail + seen["full"] = full + if sample_id == "missing": + return None + return {"as_of": 1.0, "status": "running", "count": 0, "messages": []} + + monkeypatch.setattr(server_mod, "sample_messages", _fake) + + app = server_mod.ControlServer(run_id="test")._build_app() + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient( + transport=transport, base_url="http://localhost" + ) as client: + ok = await client.get( + "/evals/e1/sample/messages", + params={"sample_id": "case/001", "tail": 5, "full": "true"}, + ) + assert ok.status_code == 200, ok.text + assert seen["sample_id"] == "case/001" # reserved-char id round-trips + assert seen["tail"] == 5 + assert seen["full"] is True + + missing = await client.get( + "/evals/e1/sample/messages", params={"sample_id": "missing"} + ) + assert missing.status_code == 404 + + async def test_404_body_shape_distinguishes_missing_route( monkeypatch: pytest.MonkeyPatch, ) -> None: From 1fe63b8a42904f16d89f3c2a2fbc11b796083a64 Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Mon, 13 Jul 2026 16:48:15 +0000 Subject: [PATCH 04/13] fix: import EvalError/EvalSample from public inspect_ai.log module mypy failed because inspect_ai.log._log does not explicitly export EvalError (it re-imports it from inspect_ai._util.error). Import both names from the public inspect_ai.log package instead. Co-Authored-By: Claude Fable 5 --- tests/_control/test_messages.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/_control/test_messages.py b/tests/_control/test_messages.py index fd12e17bfa..dfb2656079 100644 --- a/tests/_control/test_messages.py +++ b/tests/_control/test_messages.py @@ -205,7 +205,7 @@ async def test_terminal_errored_sample_reports_error_status( ) -> None: import inspect_ai.log._samples as samples_mod from inspect_ai._control.eval_state import clear_all_eval_states, register_eval - from inspect_ai.log._log import EvalError, EvalSample + from inspect_ai.log import EvalError, EvalSample monkeypatch.setattr(samples_mod, "active_samples", lambda: []) From 5ebd5c2dc7896771f71dd887aab40197afa04eef Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Mon, 13 Jul 2026 17:09:33 +0000 Subject: [PATCH 05/13] fix(control): refresh live_state at Chain/Plan step boundaries; validate --tail; uniform empty-page envelope MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review round 1 on the sample-messages read: - Chain/Plan now call set_sample_state after each solver step, so ActiveSample.live_state (and sample_state()) survive a solver replacing the TaskState object outright (e.g. returning a fork() result or a deepcopy). The @solver decorator's own solver-boundary refresh is dead code (instance-level __call__ patching is bypassed by type-level dunder lookup) — the design doc now records this. - inspect ctl sample messages --tail is validated (>= 1); a negative ?tail= on the raw endpoint clamps to an empty window instead of overshooting the slice. - The no-running-evals --json empty page now carries as_of (None) so the envelope shape stays uniform. - Updated the live_state comments in log/_samples.py and the design doc's "Data source" paragraph to describe the actual refresh mechanism. Co-Authored-By: Claude Fable 5 --- design/control-channel.md | 2 +- src/inspect_ai/_cli/ctl.py | 6 ++- src/inspect_ai/_control/messages.py | 7 ++- src/inspect_ai/log/_samples.py | 19 ++++---- src/inspect_ai/solver/_chain.py | 6 ++- src/inspect_ai/solver/_plan.py | 9 +++- tests/_control/test_ctl.py | 13 ++++++ tests/_control/test_messages.py | 18 ++++++++ tests/solver/test_chain.py | 67 +++++++++++++++++++++++++++++ 9 files changed, 132 insertions(+), 15 deletions(-) diff --git a/design/control-channel.md b/design/control-channel.md index 5e7da8c429..6e8a9cc1fe 100644 --- a/design/control-channel.md +++ b/design/control-channel.md @@ -758,7 +758,7 @@ Unlocks watchdog agents (cursored polling). Live-render TUIs follow once phase-4 - **`--full`** — raw `ChatMessage` JSON instead of the projection (combines with `--tail` to keep it bounded). - **`--json`** throughout, per the agent output contract. -**Data source: the same running-vs-terminal split as the sibling reads — with one new plumbing piece on the running side.** Running samples serve the live `TaskState.messages` — in-memory, no log involved, which is exactly why buffered-only samples work. But unlike the sibling reads, whose running-side data sources already hang off `ActiveSample` (`transcript` for `events`, the counter fields for `show` / `list`), nothing the control server can reach holds the live `TaskState` today: it lives in the `_sample_state` ContextVar (`solver/_task_state.py`), set inside the sample's own async context — invisible from the server's request-handler task. The slice therefore adds a live-state handle on `ActiveSample`. Capturing it once at sample start isn't enough: solvers can *replace* the `TaskState` object outright via `set_sample_state` (which is why `_eval/task/run.py` re-reads `sample_state() or state` after each solver step), so `set_sample_state` must also refresh the `ActiveSample` field — the same running-sample update flow `set_active_sample_total_messages` already rides. Two consistency notes fall out: between those replacement points the conversation is mutated *in place* (the turn loop appends to the same list; even `state.messages = ...` assignment lands on the same `TaskState` object), so the handle reads live at message grain, not solver grain; and since the control server shares the eval's event loop, a snapshot copied in the request handler can never observe a half-applied append or rewrite. Terminal samples need nothing new: the completed sample's messages come from the recorder's buffer via `EvalState.live` (`read_sample`), falling back to the on-disk log — the `samples` / `sample` / `events` pattern, unchanged. In the drill-down family each verb then answers one question: `sample show` — "how is it doing", `sample events` — "what happened, in order", `sample messages` — "what does the model see". +**Data source: the same running-vs-terminal split as the sibling reads — with one new plumbing piece on the running side.** Running samples serve the live `TaskState.messages` — in-memory, no log involved, which is exactly why buffered-only samples work. But unlike the sibling reads, whose running-side data sources already hang off `ActiveSample` (`transcript` for `events`, the counter fields for `show` / `list`), nothing the control server can reach holds the live `TaskState` today: it lives in the `_sample_state` ContextVar (`solver/_task_state.py`), set inside the sample's own async context — invisible from the server's request-handler task. The slice therefore adds a live-state handle on `ActiveSample`. Capturing it once at sample start isn't enough: a solver can *replace* the `TaskState` object outright by returning a new one (e.g. a `fork()` result or a deepcopy), so the handle must be refreshed wherever state is threaded step-to-step — the `Chain` / `Plan` step loops call `set_sample_state` after each solver step, and `set_sample_state` refreshes the `ActiveSample` field, the same running-sample update flow `set_active_sample_total_messages` already rides. (Neither of the mechanisms one might expect to carry this refresh actually fires per step: the `@solver` decorator's solver-boundary `set_sample_state` is dead code — it patches `__call__` as an *instance* attribute, which Python's type-level dunder lookup bypasses — and the `sample_state() or state` re-reads in `_eval/task/run.py` are error/interrupt-recovery paths, not per-step.) Two consistency notes fall out: between those replacement points the conversation is mutated *in place* (the turn loop appends to the same list; even `state.messages = ...` assignment lands on the same `TaskState` object), so the handle reads live at message grain, not solver grain; and since the control server shares the eval's event loop, a snapshot copied in the request handler can never observe a half-applied append or rewrite. Terminal samples need nothing new: the completed sample's messages come from the recorder's buffer via `EvalState.live` (`read_sample`), falling back to the on-disk log — the `samples` / `sample` / `events` pattern, unchanged. In the drill-down family each verb then answers one question: `sample show` — "how is it doing", `sample events` — "what happened, in order", `sample messages` — "what does the model see". **Scope: the sample's main conversation.** `TaskState.messages` is the sample's top-level thread. Nested agents whose conversations don't share that thread (e.g. an agent invoked as a tool, whose inner exchange never lands in `state.messages`) remain visible through `sample events` (span-tagged); a span / agent selector on `messages` is a plausible follow-on if live subagent-watching becomes a real ask — not part of the first slice. diff --git a/src/inspect_ai/_cli/ctl.py b/src/inspect_ai/_cli/ctl.py index 35c30aa9aa..bde42e867c 100644 --- a/src/inspect_ai/_cli/ctl.py +++ b/src/inspect_ai/_cli/ctl.py @@ -583,7 +583,7 @@ def sample_events_command( @click.argument("epoch", required=False, type=int, default=1) @click.option( "--tail", - type=int, + type=click.IntRange(min=1), default=None, help=( "Only the last N messages (default: the recent tail — " @@ -1654,11 +1654,13 @@ def _run_sample_messages( if not summaries: if as_json: # Uniform --json shape even on the empty page (task_id is - # unresolvable with no running evals). + # unresolvable with no running evals; as_of is None because no + # server stamped a read time). empty_page: dict[str, Any] = { "task_id": None, "sample_id": sample_id, "epoch": epoch, + "as_of": None, "status": None, "count": 0, "messages": [], diff --git a/src/inspect_ai/_control/messages.py b/src/inspect_ai/_control/messages.py index 1a4f9a81d8..404eaee821 100644 --- a/src/inspect_ai/_control/messages.py +++ b/src/inspect_ai/_control/messages.py @@ -65,7 +65,8 @@ async def sample_messages( eval_id: The eval's id. sample_id: The sample's id (string; matched against running + logged). epoch: The sample epoch. - tail: Only the last ``tail`` messages (``None`` = the whole list). + tail: Only the last ``tail`` messages (``None`` = the whole list; + negative values clamp to 0, an empty window). full: Raw serialized ``ChatMessage`` objects instead of the compact projection. """ @@ -84,7 +85,9 @@ async def sample_messages( # `tail` selects a recent window; the projection still reports each # message's absolute index so a tailed view lines up with the full one. - start = max(0, count - tail) if tail is not None else 0 + # A negative `tail` (raw HTTP callers — the CLI validates) clamps to an + # empty window rather than overshooting the slice bounds. + start = max(0, count - max(0, tail)) if tail is not None else 0 projected = [ _project(message, index, full) for index, message in enumerate(messages[start:], start=start) diff --git a/src/inspect_ai/log/_samples.py b/src/inspect_ai/log/_samples.py index 1b70761dfc..751a9a6cd2 100644 --- a/src/inspect_ai/log/_samples.py +++ b/src/inspect_ai/log/_samples.py @@ -97,10 +97,11 @@ def __init__( # current conversation (`TaskState.messages`) without reaching into the # context-bound `_sample_state` ContextVar (unreachable from the control # server's request-handler task). Captured when the sample starts and - # refreshed by `set_sample_state` (via `set_active_sample_state`) so it - # survives a solver replacing the `TaskState` object outright — mirrors - # the `sample_state() or state` re-read the task runner does. `None` - # only in the brief window before the first state is set. + # refreshed by `set_sample_state` (via `set_active_sample_state`) — + # called at sample start, after each `Chain` / `Plan` solver step, and + # pre-scoring — so it survives a solver replacing the `TaskState` + # object outright (e.g. returning a `fork()` result or a deepcopy). + # `None` only in the brief window before the first state is set. self.live_state: "TaskState | None" = None self.token_limit_usage: int | None = None self.total_cost: float | None = None @@ -328,7 +329,8 @@ async def active_sample( _sample_active.set(active) # Capture the state the runner set before entering this context (via # `set_sample_state`, which precedes `active_sample`); subsequent solver - # reassignments refresh it through `set_active_sample_state`. + # reassignments refresh it through the `set_sample_state` calls at + # `Chain` / `Plan` step boundaries. from inspect_ai.solver._task_state import sample_state active.live_state = sample_state() @@ -453,9 +455,10 @@ def set_active_sample_state(state: "TaskState") -> None: Called from `set_sample_state` so the control channel's view of the conversation survives a solver replacing the `TaskState` object outright - (the `sample_state() or state` re-read in the task runner). In-place - mutation between replacements needs no hook — the handle already points at - the object whose `messages` list is being appended to. + (`Chain` / `Plan` call `set_sample_state` after each solver step for + exactly this reason). In-place mutation between replacements needs no + hook — the handle already points at the object whose `messages` list is + being appended to. """ active = sample_active() if active: diff --git a/src/inspect_ai/solver/_chain.py b/src/inspect_ai/solver/_chain.py index e67abf53f9..21cbbcedd3 100644 --- a/src/inspect_ai/solver/_chain.py +++ b/src/inspect_ai/solver/_chain.py @@ -6,7 +6,7 @@ from inspect_ai.agent._as_solver import as_solver from ._solver import Generate, Solver, solver -from ._task_state import TaskState +from ._task_state import TaskState, set_sample_state @solver @@ -85,6 +85,10 @@ async def __call__( async with solver_transcript(slv, state) as st: state = await slv(state, generate) st.complete(state) + # a solver may return a *new* TaskState (e.g. a fork() result or a + # deepcopy) — refresh the context handle so `sample_state()` and + # the control channel's live view track the threaded state + set_sample_state(state) if state.completed: break diff --git a/src/inspect_ai/solver/_plan.py b/src/inspect_ai/solver/_plan.py index f00080cf32..1d287ec905 100644 --- a/src/inspect_ai/solver/_plan.py +++ b/src/inspect_ai/solver/_plan.py @@ -13,7 +13,7 @@ ) from ._solver import Generate, Solver -from ._task_state import TaskState +from ._task_state import TaskState, set_sample_state logger = getLogger(__name__) @@ -105,6 +105,12 @@ async def __call__( state = await solver(state, generate) st.complete(state) + # a solver may return a *new* TaskState (e.g. a fork() result + # or a deepcopy) — refresh the context handle so + # `sample_state()` and the control channel's live view track + # the threaded state + set_sample_state(state) + # check for completed if state.completed: # exit loop @@ -115,6 +121,7 @@ async def __call__( async with solver_transcript(self.finish, state) as st: state = await self.finish(state, generate) st.complete(state) + set_sample_state(state) finally: # always do cleanup if we have one diff --git a/tests/_control/test_ctl.py b/tests/_control/test_ctl.py index c1ed881a95..5551dae09b 100644 --- a/tests/_control/test_ctl.py +++ b/tests/_control/test_ctl.py @@ -1883,6 +1883,19 @@ def test_messages_json_no_servers_echoes_identifiers( assert payload["task_id"] is None assert (payload["sample_id"], payload["epoch"]) == ("s1", 1) assert payload["messages"] == [] and payload["count"] == 0 + # the envelope shape stays uniform: as_of is present (None — no server + # stamped a read time) + assert "as_of" in payload and payload["as_of"] is None + + +def test_messages_rejects_non_positive_tail() -> None: + """--tail must be >= 1; a negative/zero window is a usage error.""" + for value in ("-3", "0"): + result = cli_runner().invoke( + ctl_command, ["sample", "messages", "t", "s1", "--tail", value] + ) + assert result.exit_code != 0 + assert "--tail" in result.stderr def test_group_option_before_verb_forwards(monkeypatch: pytest.MonkeyPatch) -> None: diff --git a/tests/_control/test_messages.py b/tests/_control/test_messages.py index dfb2656079..bead597140 100644 --- a/tests/_control/test_messages.py +++ b/tests/_control/test_messages.py @@ -149,6 +149,24 @@ async def test_running_sample_tail_windows_from_the_end( assert [m["content"] for m in page["messages"]] == ["m7", "m8", "m9"] +async def test_negative_tail_clamps_to_empty_window( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A negative tail (raw HTTP callers) is an empty window, not a crash.""" + import inspect_ai.log._samples as samples_mod + + messages = [ChatMessageUser(content=f"m{i}") for i in range(5)] + monkeypatch.setattr( + samples_mod, "active_samples", lambda: [_fake_running_sample(messages)] + ) + + page = await sample_messages("e1", "1", 1, tail=-3) + assert page is not None + # count still reports the full conversation; the window is just empty + assert page["count"] == 5 + assert page["messages"] == [] + + async def test_missing_sample_returns_none(monkeypatch: pytest.MonkeyPatch) -> None: import inspect_ai._control.state as state_mod import inspect_ai.log._samples as samples_mod diff --git a/tests/solver/test_chain.py b/tests/solver/test_chain.py index 0acc0b0711..af56b557fd 100644 --- a/tests/solver/test_chain.py +++ b/tests/solver/test_chain.py @@ -1,11 +1,20 @@ from copy import deepcopy +from types import SimpleNamespace +from typing import cast +import pytest +from test_helpers.utils import simple_task_state + +from inspect_ai.model import ChatMessageUser from inspect_ai.solver import ( Generate, + Plan, + Solver, TaskState, chain, solver, ) +from inspect_ai.solver._task_state import sample_state, set_sample_state @solver @@ -25,3 +34,61 @@ def test_solver_chain(): assert len(chain2) == 6 assert len(chain(chain2, deepcopy(chain2))) == 12 + + +@solver +def replacer(): + """A solver that returns a *new* TaskState (the fork()/deepcopy pattern).""" + + async def solve(state: TaskState, _generate: Generate): + return deepcopy(state) + + return solve + + +@solver +def appender(): + async def solve(state: TaskState, _generate: Generate): + state.messages.append(ChatMessageUser(content="appended")) + return state + + return solve + + +async def _run_with_active_sample( + composed: Solver, monkeypatch: pytest.MonkeyPatch +) -> tuple[TaskState, TaskState, SimpleNamespace]: + """Run `composed` with a fake ActiveSample tracking `live_state`.""" + import inspect_ai.log._samples as samples_mod + + state = simple_task_state() + set_sample_state(state) + active = SimpleNamespace(live_state=state) + monkeypatch.setattr(samples_mod, "sample_active", lambda: active) + result = await composed(state, cast(Generate, None)) + return state, result, active + + +async def test_chain_refreshes_sample_state_on_replacement( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A step returning a new TaskState must not strand sample_state()/live_state.""" + state, result, active = await _run_with_active_sample( + chain(replacer(), appender()), monkeypatch + ) + assert result is not state + assert sample_state() is result + assert active.live_state is result + assert any(m.text == "appended" for m in active.live_state.messages) + + +async def test_plan_refreshes_sample_state_on_replacement( + monkeypatch: pytest.MonkeyPatch, +) -> None: + state, result, active = await _run_with_active_sample( + Plan([replacer(), appender()], internal=True), monkeypatch + ) + assert result is not state + assert sample_state() is result + assert active.live_state is result + assert any(m.text == "appended" for m in active.live_state.messages) From 51cc71b9c6c9ba5fca69073764f53249b1b8d438 Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Mon, 13 Jul 2026 17:36:46 +0000 Subject: [PATCH 06/13] fix(control): guard live_state refresh against fork-branch capture; split messages 404 by origin - Chain/Plan step-boundary set_sample_state now passes the superseded state, making the ActiveSample.live_state refresh a compare-and-swap: a Chain/Plan running inside a fork() subtask (whose copied context reaches the shared ActiveSample) threads a deepcopy lineage the handle never pointed at, so its refreshes can't hijack the sample's main conversation. Regression tests for both Chain and Plan branches; design doc updated. - inspect ctl sample messages now passes not_found_missing_route so a router 404 (older inspect without the endpoint) names version skew instead of misreporting "sample not found". - Drop redundant nested _structured_failures in _run_sample_messages (already wrapped by @_envelope_failures). Co-Authored-By: Claude Fable 5 --- design/control-channel.md | 2 +- src/inspect_ai/_cli/ctl.py | 18 ++++++++++----- src/inspect_ai/log/_samples.py | 17 ++++++++++++-- src/inspect_ai/solver/_chain.py | 5 ++++- src/inspect_ai/solver/_plan.py | 10 ++++++--- src/inspect_ai/solver/_task_state.py | 20 +++++++++++++---- tests/_control/test_ctl.py | 12 ++++++++++ tests/solver/test_chain.py | 33 ++++++++++++++++++++++++++++ 8 files changed, 100 insertions(+), 17 deletions(-) diff --git a/design/control-channel.md b/design/control-channel.md index 6e8a9cc1fe..c801190560 100644 --- a/design/control-channel.md +++ b/design/control-channel.md @@ -758,7 +758,7 @@ Unlocks watchdog agents (cursored polling). Live-render TUIs follow once phase-4 - **`--full`** — raw `ChatMessage` JSON instead of the projection (combines with `--tail` to keep it bounded). - **`--json`** throughout, per the agent output contract. -**Data source: the same running-vs-terminal split as the sibling reads — with one new plumbing piece on the running side.** Running samples serve the live `TaskState.messages` — in-memory, no log involved, which is exactly why buffered-only samples work. But unlike the sibling reads, whose running-side data sources already hang off `ActiveSample` (`transcript` for `events`, the counter fields for `show` / `list`), nothing the control server can reach holds the live `TaskState` today: it lives in the `_sample_state` ContextVar (`solver/_task_state.py`), set inside the sample's own async context — invisible from the server's request-handler task. The slice therefore adds a live-state handle on `ActiveSample`. Capturing it once at sample start isn't enough: a solver can *replace* the `TaskState` object outright by returning a new one (e.g. a `fork()` result or a deepcopy), so the handle must be refreshed wherever state is threaded step-to-step — the `Chain` / `Plan` step loops call `set_sample_state` after each solver step, and `set_sample_state` refreshes the `ActiveSample` field, the same running-sample update flow `set_active_sample_total_messages` already rides. (Neither of the mechanisms one might expect to carry this refresh actually fires per step: the `@solver` decorator's solver-boundary `set_sample_state` is dead code — it patches `__call__` as an *instance* attribute, which Python's type-level dunder lookup bypasses — and the `sample_state() or state` re-reads in `_eval/task/run.py` are error/interrupt-recovery paths, not per-step.) Two consistency notes fall out: between those replacement points the conversation is mutated *in place* (the turn loop appends to the same list; even `state.messages = ...` assignment lands on the same `TaskState` object), so the handle reads live at message grain, not solver grain; and since the control server shares the eval's event loop, a snapshot copied in the request handler can never observe a half-applied append or rewrite. Terminal samples need nothing new: the completed sample's messages come from the recorder's buffer via `EvalState.live` (`read_sample`), falling back to the on-disk log — the `samples` / `sample` / `events` pattern, unchanged. In the drill-down family each verb then answers one question: `sample show` — "how is it doing", `sample events` — "what happened, in order", `sample messages` — "what does the model see". +**Data source: the same running-vs-terminal split as the sibling reads — with one new plumbing piece on the running side.** Running samples serve the live `TaskState.messages` — in-memory, no log involved, which is exactly why buffered-only samples work. But unlike the sibling reads, whose running-side data sources already hang off `ActiveSample` (`transcript` for `events`, the counter fields for `show` / `list`), nothing the control server can reach holds the live `TaskState` today: it lives in the `_sample_state` ContextVar (`solver/_task_state.py`), set inside the sample's own async context — invisible from the server's request-handler task. The slice therefore adds a live-state handle on `ActiveSample`. Capturing it once at sample start isn't enough: a solver can *replace* the `TaskState` object outright by returning a new one (e.g. a `fork()` result or a deepcopy), so the handle must be refreshed wherever state is threaded step-to-step — the `Chain` / `Plan` step loops call `set_sample_state` after each solver step, and `set_sample_state` refreshes the `ActiveSample` field, the same running-sample update flow `set_active_sample_total_messages` already rides. The step-boundary refresh is a compare-and-swap (the step passes the state it superseded; the handle only moves if it currently points at that state): the handle is a plain attribute on the shared `ActiveSample`, which a `fork()` subtask's copied context still reaches through `sample_active()` — so without the guard a `Chain` / `Plan` running *inside* a fork branch (an advertised pattern; `solver_subtask` special-cases `Chain`) would overwrite the handle with its branch's conversation, non-deterministically (last-finishing branch wins), until the enclosing solver step returned. The guard makes ownership lineage-based: a fork branch threads a deepcopy the handle never pointed at, so its refreshes can't land, while the main thread's always do. (Neither of the mechanisms one might expect to carry this refresh actually fires per step: the `@solver` decorator's solver-boundary `set_sample_state` is dead code — it patches `__call__` as an *instance* attribute, which Python's type-level dunder lookup bypasses — and the `sample_state() or state` re-reads in `_eval/task/run.py` are error/interrupt-recovery paths, not per-step.) Two consistency notes fall out: between those replacement points the conversation is mutated *in place* (the turn loop appends to the same list; even `state.messages = ...` assignment lands on the same `TaskState` object), so the handle reads live at message grain, not solver grain; and since the control server shares the eval's event loop, a snapshot copied in the request handler can never observe a half-applied append or rewrite. Terminal samples need nothing new: the completed sample's messages come from the recorder's buffer via `EvalState.live` (`read_sample`), falling back to the on-disk log — the `samples` / `sample` / `events` pattern, unchanged. In the drill-down family each verb then answers one question: `sample show` — "how is it doing", `sample events` — "what happened, in order", `sample messages` — "what does the model see". **Scope: the sample's main conversation.** `TaskState.messages` is the sample's top-level thread. Nested agents whose conversations don't share that thread (e.g. an agent invoked as a tool, whose inner exchange never lands in `state.messages`) remain visible through `sample events` (span-tagged); a span / agent selector on `messages` is a plausible follow-on if live subagent-watching becomes a real ask — not part of the first slice. diff --git a/src/inspect_ai/_cli/ctl.py b/src/inspect_ai/_cli/ctl.py index bde42e867c..ebb72916cf 100644 --- a/src/inspect_ai/_cli/ctl.py +++ b/src/inspect_ai/_cli/ctl.py @@ -1638,12 +1638,11 @@ def _run_sample_messages( # an explicit --tail with --all is contradictory, so reject it rather # than silently letting one win. if show_all and tail is not None: - with _structured_failures(as_json): - _fail( - "invalid_request", - "--all and --tail are mutually exclusive (--all shows every " - "message; --tail sizes a recent window).", - ) + _fail( + "invalid_request", + "--all and --tail are mutually exclusive (--all shows every " + "message; --tail sizes a recent window).", + ) # The unseeded default is a recent tail — never an overwhelming first # page. --all disables it; an explicit --tail overrides it. if not show_all and tail is None: @@ -3023,6 +3022,12 @@ def _fetch_sample_events( ) +_MESSAGES_ROUTE_MISSING = ( + "This process is running an older inspect without the sample " + "messages endpoint; restart the eval to pick up the current version." +) + + def _fetch_sample_messages( socket_path: str, eval_id: str, @@ -3052,6 +3057,7 @@ def _fetch_sample_messages( f"Sample '{sample_id}' (epoch {epoch}) not found — it may " "not have started or not yet been written to the log." ), + not_found_missing_route=_MESSAGES_ROUTE_MISSING, ) diff --git a/src/inspect_ai/log/_samples.py b/src/inspect_ai/log/_samples.py index 751a9a6cd2..850d04c16e 100644 --- a/src/inspect_ai/log/_samples.py +++ b/src/inspect_ai/log/_samples.py @@ -101,6 +101,9 @@ def __init__( # called at sample start, after each `Chain` / `Plan` solver step, and # pre-scoring — so it survives a solver replacing the `TaskState` # object outright (e.g. returning a `fork()` result or a deepcopy). + # Step-boundary refreshes are compare-and-swap guarded so a `Chain` / + # `Plan` running inside a `fork()` subtask can't capture the handle + # for its branch conversation (see `set_active_sample_state`). # `None` only in the brief window before the first state is set. self.live_state: "TaskState | None" = None self.token_limit_usage: int | None = None @@ -450,7 +453,9 @@ def set_active_sample_total_messages(total_messages: int) -> None: active.total_messages = total_messages -def set_active_sample_state(state: "TaskState") -> None: +def set_active_sample_state( + state: "TaskState", *, replacing: "TaskState | None" = None +) -> None: """Refresh the active sample's live `TaskState` handle. Called from `set_sample_state` so the control channel's view of the @@ -459,9 +464,17 @@ def set_active_sample_state(state: "TaskState") -> None: exactly this reason). In-place mutation between replacements needs no hook — the handle already points at the object whose `messages` list is being appended to. + + When ``replacing`` is given, the refresh is a compare-and-swap: it lands + only if the handle currently points at ``replacing``. The handle lives on + the shared `ActiveSample`, which a `fork()` subtask's copied context still + reaches through `sample_active()` (the ContextVar isolation protecting + `sample_state()` doesn't extend to it) — the guard keeps a branch + `Chain` / `Plan`, threading a deepcopy lineage the handle never pointed + at, from serving its branch conversation as the sample's main thread. """ active = sample_active() - if active: + if active and (replacing is None or active.live_state is replacing): active.live_state = state diff --git a/src/inspect_ai/solver/_chain.py b/src/inspect_ai/solver/_chain.py index 21cbbcedd3..bf8563414d 100644 --- a/src/inspect_ai/solver/_chain.py +++ b/src/inspect_ai/solver/_chain.py @@ -82,13 +82,16 @@ async def __call__( from ._transcript import solver_transcript for slv in self._solvers: + prev_state = state async with solver_transcript(slv, state) as st: state = await slv(state, generate) st.complete(state) # a solver may return a *new* TaskState (e.g. a fork() result or a # deepcopy) — refresh the context handle so `sample_state()` and # the control channel's live view track the threaded state - set_sample_state(state) + # (`replacing` keeps a chain running inside a fork() branch from + # capturing the shared live view — see `set_sample_state`) + set_sample_state(state, replacing=prev_state) if state.completed: break diff --git a/src/inspect_ai/solver/_plan.py b/src/inspect_ai/solver/_plan.py index 1d287ec905..5ece18e4b3 100644 --- a/src/inspect_ai/solver/_plan.py +++ b/src/inspect_ai/solver/_plan.py @@ -101,6 +101,7 @@ async def __call__( # execute steps for index, solver in enumerate(self.steps): # run solver + prev_state = state async with solver_transcript(solver, state) as st: state = await solver(state, generate) st.complete(state) @@ -108,8 +109,10 @@ async def __call__( # a solver may return a *new* TaskState (e.g. a fork() result # or a deepcopy) — refresh the context handle so # `sample_state()` and the control channel's live view track - # the threaded state - set_sample_state(state) + # the threaded state (`replacing` keeps a plan running inside + # a fork() branch from capturing the shared live view — see + # `set_sample_state`) + set_sample_state(state, replacing=prev_state) # check for completed if state.completed: @@ -118,10 +121,11 @@ async def __call__( # execute finish if self.finish: + prev_state = state async with solver_transcript(self.finish, state) as st: state = await self.finish(state, generate) st.complete(state) - set_sample_state(state) + set_sample_state(state, replacing=prev_state) finally: # always do cleanup if we have one diff --git a/src/inspect_ai/solver/_task_state.py b/src/inspect_ai/solver/_task_state.py index 9088bde7d9..95131420ac 100644 --- a/src/inspect_ai/solver/_task_state.py +++ b/src/inspect_ai/solver/_task_state.py @@ -471,13 +471,25 @@ def sample_state() -> TaskState | None: return _sample_state.get(None) -def set_sample_state(state: TaskState) -> None: +def set_sample_state(state: TaskState, *, replacing: TaskState | None = None) -> None: + """Set the context's current `TaskState`. + + Also keeps the control channel's live-conversation handle + (`ActiveSample.live_state`) current when a solver replaces the + `TaskState` object. The ContextVar is context-isolated, but the handle + is a plain attribute on the shared `ActiveSample` — reachable from + `fork()` subtasks, whose branch conversations must not be served as the + sample's main thread. Step-threading callers (the `Chain` / `Plan` step + loops) therefore pass ``replacing`` (the state the step superseded) to + make the handle refresh a compare-and-swap: it only lands if the handle + currently points at ``replacing``, so a lineage that never owned the + handle (a fork branch threads a deepcopy) can't capture it. Without + ``replacing`` the refresh is unconditional (sample start, pre-scoring). + """ _sample_state.set(state) - # Keep the control channel's live-conversation handle current when a solver - # replaces the TaskState object (see `ActiveSample.live_state`). from inspect_ai.log._samples import set_active_sample_state - set_active_sample_state(state) + set_active_sample_state(state, replacing=replacing) _sample_state: ContextVar[TaskState] = ContextVar("sample_state") diff --git a/tests/_control/test_ctl.py b/tests/_control/test_ctl.py index 5551dae09b..1e34c2ea1c 100644 --- a/tests/_control/test_ctl.py +++ b/tests/_control/test_ctl.py @@ -1898,6 +1898,18 @@ def test_messages_rejects_non_positive_tail() -> None: assert "--tail" in result.stderr +def test_messages_missing_route_names_version_skew( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A router 404 (no `error` body) means the server predates the endpoint.""" + _patch_surface(monkeypatch, [_full_summary("aaa111", "t1")]) + _stub_httpx(monkeypatch, [(404, {"detail": "Not Found"})]) + result = cli_runner().invoke(ctl_command, ["sample", "messages", "aaa111", "s1"]) + assert result.exit_code == 1 + assert "older inspect without the sample messages endpoint" in result.stderr + assert "not yet been written" not in result.stderr + + def test_group_option_before_verb_forwards(monkeypatch: pytest.MonkeyPatch) -> None: """A mirrored option given at the group level reaches the explicit verb.""" _patch_surface(monkeypatch, [_full_summary("aaa111", "t1")]) diff --git a/tests/solver/test_chain.py b/tests/solver/test_chain.py index af56b557fd..56835950eb 100644 --- a/tests/solver/test_chain.py +++ b/tests/solver/test_chain.py @@ -92,3 +92,36 @@ async def test_plan_refreshes_sample_state_on_replacement( assert sample_state() is result assert active.live_state is result assert any(m.text == "appended" for m in active.live_state.messages) + + +@pytest.mark.parametrize( + "make_branch", + [ + lambda: chain(replacer(), appender()), + lambda: Plan([replacer(), appender()], internal=True), + ], + ids=["chain", "plan"], +) +async def test_fork_branch_does_not_capture_live_state( + monkeypatch: pytest.MonkeyPatch, make_branch +) -> None: + """A Chain/Plan threading a fork branch's deepcopy lineage must not move the shared live_state handle.""" + import inspect_ai.log._samples as samples_mod + from inspect_ai._util._async import tg_collect + + state = simple_task_state() + set_sample_state(state) + active = SimpleNamespace(live_state=state) + monkeypatch.setattr(samples_mod, "sample_active", lambda: active) + + # what fork()'s subtask does: run the branch on a deepcopy in its own + # task, whose copied context still reaches the shared ActiveSample + branch_state = deepcopy(state) + branch = make_branch() + (branch_result,) = await tg_collect( + [lambda: branch(branch_state, cast(Generate, None))] + ) + assert branch_result is not state + # the ContextVar is context-isolated; the shared handle must hold too + assert sample_state() is state + assert active.live_state is state From f98bb16a67be169e2b7f0bf684c5f9d4ac0adffe Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Mon, 13 Jul 2026 17:55:43 +0000 Subject: [PATCH 07/13] docs(control): document inspect ctl sample messages in control-channel.qmd Co-Authored-By: Claude Fable 5 --- docs/control-channel.qmd | 32 ++++++++++++++++++++++++++++++-- 1 file changed, 30 insertions(+), 2 deletions(-) diff --git a/docs/control-channel.qmd b/docs/control-channel.qmd index f66eaef91f..4adce65722 100644 --- a/docs/control-channel.qmd +++ b/docs/control-channel.qmd @@ -1,6 +1,6 @@ --- title: Control Channel -llms-description: Observe and direct running evals from another process with the inspect ctl commands - live task and sample status, error and retry detail, transcript events, task and sample cancellation, launch-config retuning, log flushing, and keep-alive. +llms-description: Observe and direct running evals from another process with the inspect ctl commands - live task and sample status, error and retry detail, transcript events, conversation snapshots, task and sample cancellation, launch-config retuning, log flushing, and keep-alive. --- ## Overview @@ -18,6 +18,7 @@ Commands are grouped by resource noun (`task`, `sample`, `process`), plus a top- | `inspect ctl sample errors` | List samples that errored or were retried. | | `inspect ctl sample show` | Show one sample's summary and error history. | | `inspect ctl sample events` | Read one sample's transcript events. | +| `inspect ctl sample messages` | Read one sample's current conversation. | | `inspect ctl sample cancel` | Cancel one running sample. | | `inspect ctl config` | View or retune launch configuration mid-flight. | | `inspect ctl process list` | List running Inspect processes. | @@ -26,7 +27,7 @@ Commands are grouped by resource noun (`task`, `sample`, `process`), plus a top- : {tbl-colwidths=[35,65]} -A bare noun implies `list`: `inspect ctl task` ≡ `inspect ctl task list`, and likewise for `sample` and `process`. All commands accept `--json` for structured output, which makes them straightforward to use from scripts and from coding agents like Claude Code. The `list` / `show` / `errors` / `events` commands are read-only. The others direct the run deliberately: `config` retunes launch parameters (never interrupting in-flight work), `task log-flush` forces a log write that would happen anyway, `process keep`/`release` only affect what happens after the eval finishes, and `task cancel` / `sample cancel` interrupt work explicitly (idempotently, with `--dry-run` support). +A bare noun implies `list`: `inspect ctl task` ≡ `inspect ctl task list`, and likewise for `sample` and `process`. All commands accept `--json` for structured output, which makes them straightforward to use from scripts and from coding agents like Claude Code. The `list` / `show` / `errors` / `events` / `messages` commands are read-only. The others direct the run deliberately: `config` retunes launch parameters (never interrupting in-flight work), `task log-flush` forces a log write that would happen anyway, `process keep`/`release` only affect what happens after the eval finishes, and `task cancel` / `sample cancel` interrupt work explicitly (idempotently, with `--dry-run` support). The endpoint is a Unix domain socket under the current user's Inspect data directory. It is not reachable over the network or by other users on the same machine, and it requires no configuration. @@ -155,6 +156,33 @@ Other options: Note that `--cursor` takes the opaque `next` token, never a timestamp — for a wall-clock window use `--since-time`. Events for samples that have already completed are also readable — they are served from the eval's log. +## Conversation Snapshots + +`inspect ctl sample messages` reads one sample's current conversation — its message list as it stands right now: + +```bash +$ inspect ctl sample messages gpqa 17 +# role content +-- --------- -------------------------------------------------- +12 assistant Let me check the data. → bash(ls /data) +13 tool README.md results.csv +14 assistant Based on the results, the compound is... + +3 of 15 messages (use --all for the whole conversation) · running +``` + +Unlike `sample events`, this is a snapshot, not a stream: solver and agent code can rewrite the message list (for example, compaction replaces a prefix with a summary), so there is no resume cursor. Each call returns the conversation as it looks at that moment — by default a recent tail (the last 20 messages), with each row carrying its absolute index so a tailed view lines up with the full one. To watch a sample incrementally, poll and compare the reported total `count` (the cheap staleness signal), or use `inspect ctl sample events` for event-grain resumable reads. + +| Option | Description | +|----|----| +| `--tail N` | Only the last N messages (default 20). Mutually exclusive with `--all`. | +| `--all` | The whole conversation instead of a recent tail. | +| `--full` | Return raw `ChatMessage` JSON instead of compact one-line summaries. | + +: {tbl-colwidths=[35,65]} + +With `--json` the response is an `{as_of, status, count, messages}` envelope, prefixed with the resolved `task_id` / `sample_id` / `epoch` (so a defaulted epoch is visible). As with events, `EPOCH` defaults to 1, and conversations of samples that have already completed are also readable — they are served from the eval's log. + ## Cancellation `inspect ctl sample cancel` cancels one running sample — the typical move when a sample has stalled (high `idle` in `sample list`) or is burning tokens without progress. By default the sample completes and the scorer runs on the work done so far (it is recorded with an `operator` limit, like the in-process TUI's cancel); pass `--error` to mark it errored instead (not permitted for samples configured to fail on errors). The rest of the task is unaffected. From dfa9fcaed8752b9122747ff9976ecf7ab00f4d3d Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Mon, 13 Jul 2026 18:11:08 +0000 Subject: [PATCH 08/13] docs(control): make sample messages example self-consistent (--tail 3) The example's footer (3 of 15 messages) is only producible with --tail 3; the bare command with the default tail of 20 would show all 15. Co-Authored-By: Claude Fable 5 --- docs/control-channel.qmd | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/control-channel.qmd b/docs/control-channel.qmd index 4adce65722..82eacc78d1 100644 --- a/docs/control-channel.qmd +++ b/docs/control-channel.qmd @@ -161,7 +161,7 @@ Note that `--cursor` takes the opaque `next` token, never a timestamp — for a `inspect ctl sample messages` reads one sample's current conversation — its message list as it stands right now: ```bash -$ inspect ctl sample messages gpqa 17 +$ inspect ctl sample messages gpqa 17 --tail 3 # role content -- --------- -------------------------------------------------- 12 assistant Let me check the data. → bash(ls /data) From c4d8ac67cb14ba0da4eaaf139de7f575a7ea7a90 Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Mon, 13 Jul 2026 18:35:59 +0000 Subject: [PATCH 09/13] perf(control): skip event transcript on terminal sample messages read Pass exclude_fields={events, store, output} to _full_sample and resolve attachments in core mode (messages are covered; full additionally walks the excluded events/model-calls), so a terminal 'sample messages' read no longer deserializes and walks the whole transcript. Co-Authored-By: Claude Fable 5 --- src/inspect_ai/_control/messages.py | 15 ++++++++---- tests/_control/test_messages.py | 36 +++++++++++++++++++++++++++++ 2 files changed, 46 insertions(+), 5 deletions(-) diff --git a/src/inspect_ai/_control/messages.py b/src/inspect_ai/_control/messages.py index 404eaee821..1e3c5190f6 100644 --- a/src/inspect_ai/_control/messages.py +++ b/src/inspect_ai/_control/messages.py @@ -129,21 +129,26 @@ async def _logged_source( ) -> MessagesSource | None: """The terminal source for a sample (recorder buffer, then on-disk log). - ``None`` when the eval/sample isn't available here. Reads the full + ``None`` when the eval/sample isn't available here. Reads the ``EvalSample`` via :func:`inspect_ai._control.state._full_sample` — the same gap-free recorder-then-log source the error-detail and events reads - use — and resolves its content attachments so image/large-text refs render - as real content rather than ``attachment://`` placeholders. + use — excluding the heavy fields the response never consumes (``events`` + dwarfs the rest for a long agentic sample), and resolves its content + attachments in ``core`` mode (which covers ``messages``; ``full`` would + additionally walk the excluded events/model-calls) so image/large-text + refs render as real content rather than ``attachment://`` placeholders. """ from inspect_ai._control.state import _full_sample - sample = await _full_sample(eval_id, sample_id, epoch) + sample = await _full_sample( + eval_id, sample_id, epoch, exclude_fields={"events", "store", "output"} + ) if sample is None: return None from inspect_ai.log._condense import resolve_sample_attachments - sample = resolve_sample_attachments(sample, "full") + sample = resolve_sample_attachments(sample, "core") return MessagesSource( messages=list(sample.messages or []), status="error" if sample.error is not None else "completed", diff --git a/tests/_control/test_messages.py b/tests/_control/test_messages.py index bead597140..b21f5f4e0a 100644 --- a/tests/_control/test_messages.py +++ b/tests/_control/test_messages.py @@ -204,7 +204,10 @@ async def test_terminal_sample_serves_logged_messages( ], ) + excluded: list[Any] = [] + async def read_sample(id: Any, epoch: int, *, exclude_fields: Any = None) -> Any: + excluded.append(exclude_fields) return sample if str(id) == "s1" and epoch == 1 else None try: @@ -214,6 +217,39 @@ async def read_sample(id: Any, epoch: int, *, exclude_fields: Any = None) -> Any assert page["status"] == "completed" assert page["count"] == 2 assert [m["role"] for m in page["messages"]] == ["user", "assistant"] + # heavy unconsumed fields are excluded from the read; the fields the + # response is built from are not + assert excluded == [{"events", "store", "output"}] + finally: + clear_all_eval_states() + + +async def test_terminal_sample_resolves_message_attachments( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import inspect_ai.log._samples as samples_mod + from inspect_ai._control.eval_state import clear_all_eval_states, register_eval + from inspect_ai.log._log import EvalSample + + monkeypatch.setattr(samples_mod, "active_samples", lambda: []) + + sample = EvalSample( + id="s1", + epoch=1, + input="question", + target="answer", + messages=[ChatMessageUser(content="attachment://abc123")], + attachments={"abc123": "the real content"}, + ) + + async def read_sample(id: Any, epoch: int, *, exclude_fields: Any = None) -> Any: + return sample + + try: + register_eval("e1", 1, live=FakeLiveEvalData(sample=read_sample)) + page = await sample_messages("e1", "s1", 1) + assert page is not None + assert page["messages"][0]["content"] == "the real content" finally: clear_all_eval_states() From 7690e616209b38c88b0fa11ff7f2715e8f2ed2db Mon Sep 17 00:00:00 2001 From: Ransom Richardson Date: Tue, 14 Jul 2026 14:49:17 -0400 Subject: [PATCH 10/13] fix(control): harden live_state fork-capture guard and pin it with a real-fork test - CAS-guard the @solver wrapper's set_sample_state refreshes (dormant today, but correct if the __call__ patching ever comes alive) - add test_fork_does_not_capture_live_state: real fork()/subtask mechanics, not the deepcopy+spawn simulation; verified it fails when the CAS guard is removed - assert log.status == "success" in eval-based fork tests so in-solver assertion failures can't pass silently as sample errors - document log/_samples.py as the process-wide running-sample registry (module docstring) and tighten the live_state field comment Co-Authored-By: Claude Fable 5 --- src/inspect_ai/log/_samples.py | 43 +++++++++++++----- src/inspect_ai/solver/_solver.py | 11 ++++- tests/solver/test_fork.py | 76 +++++++++++++++++++++++++++++++- 3 files changed, 116 insertions(+), 14 deletions(-) diff --git a/src/inspect_ai/log/_samples.py b/src/inspect_ai/log/_samples.py index 850d04c16e..37c9bb969e 100644 --- a/src/inspect_ai/log/_samples.py +++ b/src/inspect_ai/log/_samples.py @@ -1,3 +1,29 @@ +"""Process-wide registry of running samples (and in-flight model events). + +Despite living in the ``log`` package, this module is not about log +persistence — it's the runtime observability hub for samples currently +executing in this process. Each running sample has an :class:`ActiveSample` +in the module-global registry (:func:`active_samples`): the executing side +(task runner, model layer, limits, solvers) *pushes* live state onto it via +the ``set_active_sample_*`` functions, and cross-task observers (the TUI, +the control channel, ACP) read it. The pushes exist because most sample +state is context-bound (ContextVars, limit trees) and therefore unreachable +from an observer's task — the registry mirrors it outward. Most such +mirrors are copied scalars (token/turn totals, limits); +:attr:`ActiveSample.live_state` is the exception, a *handle* to the +sample's live ``TaskState`` (copying a conversation per append would be +prohibitive), and the one upward-pointing object edge here. + +That role makes this module a cross-layer hub: it's imported from every +layer of the stack, and its own references to higher-layer types +(``TaskState``, ACP transport, hooks) are ``TYPE_CHECKING``-only or +function-local to keep the module import graph acyclic. + +The ``_active_model_event`` half of the file is a sibling concern — tracking +the in-flight ``ModelEvent`` so providers can attach call payloads and retry +counts — cohabiting here rather than part of the sample registry. +""" + import contextlib from contextvars import ContextVar from datetime import datetime, timezone @@ -93,17 +119,12 @@ def __init__( self.total_messages = 0 self.total_tokens = 0 self.total_turns = 0 - # The sample's live `TaskState`, so the control channel can serve the - # current conversation (`TaskState.messages`) without reaching into the - # context-bound `_sample_state` ContextVar (unreachable from the control - # server's request-handler task). Captured when the sample starts and - # refreshed by `set_sample_state` (via `set_active_sample_state`) — - # called at sample start, after each `Chain` / `Plan` solver step, and - # pre-scoring — so it survives a solver replacing the `TaskState` - # object outright (e.g. returning a `fork()` result or a deepcopy). - # Step-boundary refreshes are compare-and-swap guarded so a `Chain` / - # `Plan` running inside a `fork()` subtask can't capture the handle - # for its branch conversation (see `set_active_sample_state`). + # The sample's live `TaskState` — the handle observers read the + # current conversation from (see the module docstring). Refreshed via + # `set_sample_state` (sample start, `Chain`/`Plan` step boundaries, + # pre-scoring) so it survives a solver replacing the state object; + # step-boundary refreshes are compare-and-swap guarded against + # capture by `fork()` branches (see `set_active_sample_state`). # `None` only in the brief window before the first state is set. self.live_state: "TaskState | None" = None self.token_limit_usage: int | None = None diff --git a/src/inspect_ai/solver/_solver.py b/src/inspect_ai/solver/_solver.py index 3981ec33df..de8493af50 100644 --- a/src/inspect_ai/solver/_solver.py +++ b/src/inspect_ai/solver/_solver.py @@ -212,8 +212,13 @@ def solver_wrapper(*args: P.args, **kwargs: P.kwargs) -> Solver: async def call_with_state( state: TaskState, generate: Generate ) -> TaskState: + prev_state = state state = await original_call(state, generate) - set_sample_state(state) + # solvers can run inside a fork() branch, whose copied + # context still reaches the shared ActiveSample — CAS via + # `replacing` so a branch lineage can't capture the live + # view (see `set_sample_state`) + set_sample_state(state, replacing=prev_state) return state registered_solver = solver @@ -227,8 +232,10 @@ async def call_with_state( async def registered_solver( state: TaskState, generate: Generate ) -> TaskState: + prev_state = state state = await solver(state, generate) - set_sample_state(state) + # CAS-guarded for the same fork() reason as above + set_sample_state(state, replacing=prev_state) return state registry_tag( diff --git a/tests/solver/test_fork.py b/tests/solver/test_fork.py index 05d802f672..b95b058a72 100644 --- a/tests/solver/test_fork.py +++ b/tests/solver/test_fork.py @@ -1,5 +1,9 @@ +from copy import deepcopy + from inspect_ai import Task, eval, task from inspect_ai.dataset import Sample +from inspect_ai.log._samples import sample_active +from inspect_ai.model import ChatMessageUser from inspect_ai.solver import ( Generate, TaskState, @@ -7,11 +11,13 @@ solver, ) from inspect_ai.solver._chain import chain +from inspect_ai.solver._task_state import sample_state from inspect_ai.util import store def test_forking_solver(): - eval(forking_solver_task(), model="mockllm/model") + log = eval(forking_solver_task(), model="mockllm/model")[0] + assert log.status == "success" @task @@ -58,3 +64,71 @@ async def solve(state: TaskState, _generate: Generate): return Task( dataset=[Sample(input="Say Hello", target="Hello")], solver=forking_solver() ) + + +def test_fork_does_not_capture_live_state(): + """Real fork() branches must not move the shared `ActiveSample.live_state`. + + Unlike the simulation in `test_chain.py` (deepcopy + task spawn), this + pins the actual `fork()`/`subtask` mechanics: the branch's copied context + still reaches the shared `ActiveSample`, so only the CAS guard in + `set_active_sample_state` keeps a branch `Chain` from serving its branch + conversation as the sample's main thread. + """ + log = eval(fork_live_state_task(), model="mockllm/model")[0] + assert log.status == "success" + + +@task +def fork_live_state_task(): + @solver + def state_replacer(): + """Return a *new* TaskState (the deepcopy pattern the CAS guards).""" + + async def solve(state: TaskState, _generate: Generate): + return deepcopy(state) + + return solve + + @solver + def marker(text: str): + async def solve(state: TaskState, _generate: Generate): + state.messages.append(ChatMessageUser(content=text)) + return state + + return solve + + @solver + def forking_solver(): + async def solve(state: TaskState, _generate: Generate): + active = sample_active() + assert active is not None + # the runner registered this sample's state as the live view + assert active.live_state is state + assert sample_state() is state + + # bare-solver branch + result = await fork(state, marker("branch-a")) + assert result is not state + assert active.live_state is state + assert sample_state() is state + + # Chain branch that replaces the state mid-branch: exercises the + # CAS-guarded step refresh inside the fork subtask (a capture + # would persist in live_state until pre-scoring, so these + # post-fork asserts do detect it) + result = await fork(state, chain(state_replacer(), marker("branch-b"))) + assert result is not state + assert any(m.text == "branch-b" for m in result.messages) + assert active.live_state is state + assert sample_state() is state + # no branch conversation leaked into the main thread + assert not any(m.text in ("branch-a", "branch-b") for m in state.messages) + + return state + + return solve + + return Task( + dataset=[Sample(input="Say Hello", target="Hello")], solver=forking_solver() + ) From 02420d130a059190fff535709ae40f933f1ee004 Mon Sep 17 00:00:00 2001 From: Ransom Richardson Date: Tue, 14 Jul 2026 15:00:43 -0400 Subject: [PATCH 11/13] docs(solver): disambiguate the two fork() roles in step-refresh comments Co-Authored-By: Claude Fable 5 --- src/inspect_ai/solver/_chain.py | 13 ++++++++----- src/inspect_ai/solver/_plan.py | 14 ++++++++------ 2 files changed, 16 insertions(+), 11 deletions(-) diff --git a/src/inspect_ai/solver/_chain.py b/src/inspect_ai/solver/_chain.py index bf8563414d..de5a1a9e9b 100644 --- a/src/inspect_ai/solver/_chain.py +++ b/src/inspect_ai/solver/_chain.py @@ -86,11 +86,14 @@ async def __call__( async with solver_transcript(slv, state) as st: state = await slv(state, generate) st.complete(state) - # a solver may return a *new* TaskState (e.g. a fork() result or a - # deepcopy) — refresh the context handle so `sample_state()` and - # the control channel's live view track the threaded state - # (`replacing` keeps a chain running inside a fork() branch from - # capturing the shared live view — see `set_sample_state`) + # a solver may return a different TaskState object than it was + # passed (eg. a deepcopy it made, or a state it got from fork()) + # — refresh the context handle so `sample_state()` and the + # control channel's live view follow the threaded state. The + # `replacing` CAS covers a separate case: when this chain is + # itself running *inside* a fork() branch, its states are branch + # copies and the refresh must not move the shared live view + # (see `set_sample_state`) set_sample_state(state, replacing=prev_state) if state.completed: break diff --git a/src/inspect_ai/solver/_plan.py b/src/inspect_ai/solver/_plan.py index 5ece18e4b3..5c9f042631 100644 --- a/src/inspect_ai/solver/_plan.py +++ b/src/inspect_ai/solver/_plan.py @@ -106,12 +106,14 @@ async def __call__( state = await solver(state, generate) st.complete(state) - # a solver may return a *new* TaskState (e.g. a fork() result - # or a deepcopy) — refresh the context handle so - # `sample_state()` and the control channel's live view track - # the threaded state (`replacing` keeps a plan running inside - # a fork() branch from capturing the shared live view — see - # `set_sample_state`) + # a solver may return a different TaskState object than it + # was passed (eg. a deepcopy it made, or a state it got from + # fork()) — refresh the context handle so `sample_state()` + # and the control channel's live view follow the threaded + # state. The `replacing` CAS covers a separate case: when + # this plan is itself running *inside* a fork() branch, its + # states are branch copies and the refresh must not move the + # shared live view (see `set_sample_state`) set_sample_state(state, replacing=prev_state) # check for completed From 321170ac1bc689dd2f7204d70b4de472ccc235e7 Mon Sep 17 00:00:00 2001 From: Ransom Richardson Date: Tue, 14 Jul 2026 15:04:31 -0400 Subject: [PATCH 12/13] docs(solver): consolidate set_sample_state rationale into its docstring Drop the per-call-site explanations (Chain/Plan step loops, @solver wrapper) in favor of the full story on set_sample_state itself. Co-Authored-By: Claude Fable 5 --- src/inspect_ai/solver/_chain.py | 8 -------- src/inspect_ai/solver/_plan.py | 8 -------- src/inspect_ai/solver/_solver.py | 5 ----- src/inspect_ai/solver/_task_state.py | 28 +++++++++++++++++----------- 4 files changed, 17 insertions(+), 32 deletions(-) diff --git a/src/inspect_ai/solver/_chain.py b/src/inspect_ai/solver/_chain.py index de5a1a9e9b..7a6e1031e0 100644 --- a/src/inspect_ai/solver/_chain.py +++ b/src/inspect_ai/solver/_chain.py @@ -86,14 +86,6 @@ async def __call__( async with solver_transcript(slv, state) as st: state = await slv(state, generate) st.complete(state) - # a solver may return a different TaskState object than it was - # passed (eg. a deepcopy it made, or a state it got from fork()) - # — refresh the context handle so `sample_state()` and the - # control channel's live view follow the threaded state. The - # `replacing` CAS covers a separate case: when this chain is - # itself running *inside* a fork() branch, its states are branch - # copies and the refresh must not move the shared live view - # (see `set_sample_state`) set_sample_state(state, replacing=prev_state) if state.completed: break diff --git a/src/inspect_ai/solver/_plan.py b/src/inspect_ai/solver/_plan.py index 5c9f042631..874420c9c1 100644 --- a/src/inspect_ai/solver/_plan.py +++ b/src/inspect_ai/solver/_plan.py @@ -106,14 +106,6 @@ async def __call__( state = await solver(state, generate) st.complete(state) - # a solver may return a different TaskState object than it - # was passed (eg. a deepcopy it made, or a state it got from - # fork()) — refresh the context handle so `sample_state()` - # and the control channel's live view follow the threaded - # state. The `replacing` CAS covers a separate case: when - # this plan is itself running *inside* a fork() branch, its - # states are branch copies and the refresh must not move the - # shared live view (see `set_sample_state`) set_sample_state(state, replacing=prev_state) # check for completed diff --git a/src/inspect_ai/solver/_solver.py b/src/inspect_ai/solver/_solver.py index de8493af50..28b51dd626 100644 --- a/src/inspect_ai/solver/_solver.py +++ b/src/inspect_ai/solver/_solver.py @@ -214,10 +214,6 @@ async def call_with_state( ) -> TaskState: prev_state = state state = await original_call(state, generate) - # solvers can run inside a fork() branch, whose copied - # context still reaches the shared ActiveSample — CAS via - # `replacing` so a branch lineage can't capture the live - # view (see `set_sample_state`) set_sample_state(state, replacing=prev_state) return state @@ -234,7 +230,6 @@ async def registered_solver( ) -> TaskState: prev_state = state state = await solver(state, generate) - # CAS-guarded for the same fork() reason as above set_sample_state(state, replacing=prev_state) return state diff --git a/src/inspect_ai/solver/_task_state.py b/src/inspect_ai/solver/_task_state.py index 95131420ac..7569bdac0e 100644 --- a/src/inspect_ai/solver/_task_state.py +++ b/src/inspect_ai/solver/_task_state.py @@ -474,17 +474,23 @@ def sample_state() -> TaskState | None: def set_sample_state(state: TaskState, *, replacing: TaskState | None = None) -> None: """Set the context's current `TaskState`. - Also keeps the control channel's live-conversation handle - (`ActiveSample.live_state`) current when a solver replaces the - `TaskState` object. The ContextVar is context-isolated, but the handle - is a plain attribute on the shared `ActiveSample` — reachable from - `fork()` subtasks, whose branch conversations must not be served as the - sample's main thread. Step-threading callers (the `Chain` / `Plan` step - loops) therefore pass ``replacing`` (the state the step superseded) to - make the handle refresh a compare-and-swap: it only lands if the handle - currently points at ``replacing``, so a lineage that never owned the - handle (a fork branch threads a deepcopy) can't capture it. Without - ``replacing`` the refresh is unconditional (sample start, pre-scoring). + Called wherever the current `TaskState` may have been replaced with a + different object — a solver can return a deepcopy or a state it got from + `fork()` rather than the state it was passed — so that `sample_state()` + and the control channel's live-conversation handle + (`ActiveSample.live_state`) follow the threaded state. + + The ContextVar is context-isolated, but the handle is a plain attribute + on the shared `ActiveSample` — reachable from `fork()` subtasks, whose + branch conversations must not be served as the sample's main thread. + Callers that can themselves be running inside a fork() branch (the + `Chain` / `Plan` step loops, the `@solver` call wrapper) therefore pass + ``replacing`` (the state the step superseded) to make the handle refresh + a compare-and-swap: it only lands if the handle currently points at + ``replacing``, so a lineage that never owned the handle (a fork branch + threads a deepcopy) can't capture it. Without ``replacing`` the refresh + is unconditional — reserved for callers that are by definition on the + sample's main thread (sample start, pre-scoring). """ _sample_state.set(state) from inspect_ai.log._samples import set_active_sample_state From e9248019e73c4733fc7e169675653f44fccf79b8 Mon Sep 17 00:00:00 2001 From: Ransom Richardson Date: Tue, 14 Jul 2026 15:26:46 -0400 Subject: [PATCH 13/13] fix(control): pretty-print raw events for ctl sample events --full MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The table renderer expects the compact projection's flattened fields (unix-float timestamp, top-level tokens/stop_reason/completion), so raw events rendered as a mostly-empty table — less visible data with --full than without. Match the sample messages behavior: in terminal mode, --full pretty-prints the raw JSON (footer/cursor unchanged). Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 1 + src/inspect_ai/_cli/ctl.py | 11 ++++++++--- tests/_control/test_ctl.py | 35 +++++++++++++++++++++++++++++++++-- 3 files changed, 42 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 28804d911e..17f5fa7646 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ - Control Channel: New `inspect ctl sample messages TASK SID [EPOCH]` reads a running (or buffered-but-unlogged) sample's current conversation as a snapshot, with `--tail`/`--all`/`--full`. - Control Channel: `inspect ctl` sample commands given an exact full task id now stop discovery at the process hosting it, skipping busy-retry delays from older sibling processes. - Control Channel: `inspect ctl config --key NAME LIMIT` retunes any named `concurrency()` limit mid-flight, without the registering code opting in; the config view lists the registered keys. +- Control Channel: `inspect ctl sample events --full` now pretty-prints the raw events instead of rendering a mostly-empty summary table. - Bedrock: Assistant text and reasoning are now preserved alongside tool calls in the same turn instead of being dropped. (#4457) - Bugfix: Crash recovery now reconstructs `sample.messages` from the first model role's conversation when a solver runs multiple agents concurrently, instead of returning whichever agent's model call happened to fire last. (#4414) - Bugfix: Several raised errors (unexpected Anthropic input block, unsupported image data type, missing sample-buffer database) now interpolate the offending value into the message instead of showing the literal placeholder text. diff --git a/src/inspect_ai/_cli/ctl.py b/src/inspect_ai/_cli/ctl.py index c2d1da2072..ffd63b9ae1 100644 --- a/src/inspect_ai/_cli/ctl.py +++ b/src/inspect_ai/_cli/ctl.py @@ -1644,7 +1644,7 @@ def _run_sample_events( click.echo(json_lib.dumps(page, indent=2)) return - _print_events(page) + _print_events(page, full=full) @_envelope_failures @@ -3572,10 +3572,15 @@ def _target(current: Any, key: str) -> str: click.echo(f" note: {note}") -def _print_events(page: dict[str, Any]) -> None: +def _print_events(page: dict[str, Any], *, full: bool) -> None: """Render a page of transcript events (table) plus a cursor footer.""" events = page.get("events") or [] - if not events: + if full: + # Raw mode is for machine consumption; the human rendering is the + # compact projection (whose flattened fields the table expects), so + # just pretty-print the raw events. + click.echo(json_lib.dumps(events, indent=2)) + elif not events: click.echo("(no events)") else: rows: list[tuple[str, ...]] = [] diff --git a/tests/_control/test_ctl.py b/tests/_control/test_ctl.py index 9950c61935..4ced150979 100644 --- a/tests/_control/test_ctl.py +++ b/tests/_control/test_ctl.py @@ -431,7 +431,7 @@ def test_print_events_table_and_footer( "next": "CURSORX", "done": False, } - _print_events(page) + _print_events(page, full=False) out = capsys.readouterr().out assert "event" in out.splitlines()[0] # table header # per-type summaries @@ -445,13 +445,44 @@ def test_print_events_table_and_footer( def test_print_events_empty_and_done(capsys: pytest.CaptureFixture[str]) -> None: from inspect_ai._cli.ctl import _print_events - _print_events({"events": [], "next": "X", "done": True}) + _print_events({"events": [], "next": "X", "done": True}, full=False) out = capsys.readouterr().out assert "(no events)" in out assert "done" in out assert "next:" not in out # a done stream offers no resume cursor +def test_print_events_full_pretty_prints_raw( + capsys: pytest.CaptureFixture[str], +) -> None: + """Full mode emits raw JSON: nested fields the summary table can't render.""" + from inspect_ai._cli.ctl import _print_events + + page = { + "events": [ + { + "event": "model", + # raw model_dump timestamps are ISO strings, not unix floats + "timestamp": "2026-07-14T19:10:05+00:00", + "model": "openai/gpt", + "output": {"usage": {"total_tokens": 42}, "completion": "hello"}, + } + ], + "next": "CURSORX", + "done": False, + } + _print_events(page, full=True) + out = capsys.readouterr().out + # nested raw fields survive (the compact table would have dropped them) + assert '"total_tokens": 42' in out + assert '"completion": "hello"' in out + assert "2026-07-14T19:10:05+00:00" in out + # the cursor footer still supports polling loops + assert "1 event" in out + assert "more" in out + assert "next: CURSORX" in out + + def test_keep_alive_footer_on_when_all_tasks_keep_alive( capsys: pytest.CaptureFixture[str], ) -> None: