diff --git a/cli/internal/daemon/protocol/api.gen.go b/cli/internal/daemon/protocol/api.gen.go index 79f070cd..ac8b6124 100644 --- a/cli/internal/daemon/protocol/api.gen.go +++ b/cli/internal/daemon/protocol/api.gen.go @@ -3893,8 +3893,11 @@ type GetContextParams struct { SectionID *ID `form:"section_id,omitempty" json:"section_id,omitempty"` Page *Counter `form:"page,omitempty" json:"page,omitempty"` After *Counter `form:"after,omitempty" json:"after,omitempty"` - Limit *Limit `form:"limit,omitempty" json:"limit,omitempty"` - Next *Next `form:"next,omitempty" json:"next,omitempty"` + + // SinceMs Requests only. Include attempts finished at or after this Unix timestamp in milliseconds. + SinceMs *Counter `form:"since_ms,omitempty" json:"since_ms,omitempty"` + Limit *Limit `form:"limit,omitempty" json:"limit,omitempty"` + Next *Next `form:"next,omitempty" json:"next,omitempty"` } // GetSessionFileParams defines parameters for GetSessionFile. @@ -4932,7 +4935,7 @@ type ClientInterface interface { // GetContext Inspect prepared context or request records // - // Summary never prepares a request. Section requires snapshot_id, section_id, page and forbids request-pagination fields. Requests accepts after/limit/next. Stale snapshot is 410 context_changed, never another section. Requires capability `context: 1`. + // Summary never prepares a request. Section requires snapshot_id, section_id, page and forbids request-pagination fields. Requests accepts after/since_ms/limit/next. Stale snapshot is 410 context_changed, never another section. Requires capability `context: 1`. // // Corresponds with GET /sessions/{session_id}/context (the `GetContext` operationId). GetContext(ctx context.Context, sessionID ID, params *GetContextParams, reqEditors ...RequestEditorFn) (*http.Response, error) @@ -6963,7 +6966,7 @@ func (c *Client) CompactSession(ctx context.Context, sessionID ID, body CompactS // GetContext Inspect prepared context or request records // -// Summary never prepares a request. Section requires snapshot_id, section_id, page and forbids request-pagination fields. Requests accepts after/limit/next. Stale snapshot is 410 context_changed, never another section. Requires capability `context: 1`. +// Summary never prepares a request. Section requires snapshot_id, section_id, page and forbids request-pagination fields. Requests accepts after/since_ms/limit/next. Stale snapshot is 410 context_changed, never another section. Requires capability `context: 1`. // // Corresponds with GET /sessions/{session_id}/context (the `GetContext` operationId). func (c *Client) GetContext(ctx context.Context, sessionID ID, params *GetContextParams, reqEditors ...RequestEditorFn) (*http.Response, error) { @@ -11829,6 +11832,18 @@ func NewGetContextRequest(server string, sessionID ID, params *GetContextParams) } + if params.SinceMs != nil { + + if queryFrag, err := runtime.StyleParamWithOptions("form", true, "since_ms", *params.SinceMs, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationQuery, Type: "integer", Format: ""}); err != nil { + return nil, err + } else { + for _, qp := range strings.Split(queryFrag, "&") { + rawQueryFragments = append(rawQueryFragments, qp) + } + } + + } + if params.Limit != nil { if queryFrag, err := runtime.StyleParamWithOptions("form", true, "limit", *params.Limit, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationQuery, Type: "integer", Format: ""}); err != nil { diff --git a/docs/openapi.yaml b/docs/openapi.yaml index 0b8f0c58..de640675 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -1620,7 +1620,7 @@ paths: '410': $ref: '#/components/responses/Problem410' description: 'Summary never prepares a request. Section requires snapshot_id, section_id, page and - forbids request-pagination fields. Requests accepts after/limit/next. Stale snapshot is 410 context_changed, + forbids request-pagination fields. Requests accepts after/since_ms/limit/next. Stale snapshot is 410 context_changed, never another section. Requires capability `context: 1`.' parameters: - name: session_id @@ -1655,6 +1655,12 @@ paths: required: false schema: $ref: '#/components/schemas/Counter' + - name: since_ms + in: query + required: false + description: Requests only. Include attempts finished at or after this Unix timestamp in milliseconds. + schema: + $ref: '#/components/schemas/Counter' - $ref: '#/components/parameters/Limit' - $ref: '#/components/parameters/Next' /sessions/{session_id}/catalog: diff --git a/robot-docs/provider-requests.md b/robot-docs/provider-requests.md index 9fceec16..3455ebbc 100644 --- a/robot-docs/provider-requests.md +++ b/robot-docs/provider-requests.md @@ -57,8 +57,12 @@ Within one projection the request is append-only, so equal head hash plus equal ## Local API `GET /sessions/{session_id}/context?view=requests` reads paged provider -request records from the store. `after` is a durable request-record ID; -opaque `next` continues the same query. The route requires the daemon token. +request records from the store. `after` is a durable request-record ID. +Optional `since_ms` is a nonnegative Unix timestamp in milliseconds: only +attempts finished at or after it are returned. Records remain oldest ID first. +Opaque `next` continues the same query; repeat its filters, including `since_ms`, +and limit unchanged. Prepared context views reject `since_ms`. The route +requires the daemon token. These records and the server's quota observations use the same stored facts. See the [HTTP contract](../docs/http-api-design.md#prepared-context-and-request-records) for fields and bounds. diff --git a/robot-docs/web.md b/robot-docs/web.md index 383bf5e5..b157c1fb 100644 --- a/robot-docs/web.md +++ b/robot-docs/web.md @@ -30,9 +30,9 @@ The **terminal** (`nodes/TerminalNode.svelte`, `views/TerminalView.svelte`, `job The **trace** (`nodes/TraceNode.svelte`, `views/TraceView.svelte`, `trace.ts`, `TraceTie.svelte`, `tracelink.svelte.ts`) shows one task as albedo ran it: "trace" on the sign-off of a turn that ran tools opens it beside the chat, tied to it like a review. A task is that turn and the turns after it until the person types again (mail from a child, continuations), so a parent's wait for its agents and their replies fall inside it. A trace opened on the session's newest task follows the session instead: every later turn is added as it begins, each message the person sent marked on the axis ("you"), and it reads again whenever the session's phase changes as well as every 3 s while anything in it works, never while the tab is hidden, and every 5 s while a read fails. Its axis runs ahead of now (a dashed line), by a quarter of the task or at least 30 s, and grows only when now reaches the end, so the calls already drawn hold still under the pointer between reads. "earlier turn" adds the turn before the earliest shown (`turnBefore` walks history back to the person's previous message; the count is saved on the window as `earlier`), read in place so the view keeps its zoom. -The trace is read in parallel where it can be: the agents, jobs and the search for where the task's model calls begin (single-row probes, eight at a time) start with its history, and the session's own lane shows before its agents are read. It animates only transform and opacity, and nothing animates while it is out of view. +The trace is read in parallel where it can be: agents, jobs and model calls start with its history, and the session's own lane shows before its agents are read. Each window keeps per-lane history positions, request sequences and known children. Live reads append new entries and calls, close pending tools, and reuse completed child lanes until their activity changes; running jobs and child summaries are reconciled. A followed child keeps growing while its parent waits. Hydrated inputs and traces are kept, with a tool’s later result fetched separately. Earlier-turn changes and failed reads discard the cursors. Initial model-call reads use `since_ms` to skip requests that finished before the task; a daemon rejecting it with 400 is detected once, then uses the older parallel single-row probing search. Later reads use request sequences on either daemon. It animates only transform and opacity, and nothing animates while it is out of view. -Rows come from three places: model calls from the session's request records (`GET /sessions/{id}/context?view=requests`, oldest first; a long session finds the task's first record by probing single rows rather than reading every page before it), tool calls from the transcript (a call's entry to its result's; input a history page only referenced is read afterwards in one `history/content` walk per agent), and agents from `GET /sessions?parent_id=`, three deep, kept when they were alive during the task. Each agent is one lane: its calls never overlap, so model calls (grey, the wait for the first byte lighter) and tool calls (the lane's hue, which keeps clear of red; failures are red) take turns along one strip, and a gap of 1.5 s or more while an agent below works is hatched as waiting. Under the strip, each tool call has one chip naming the most telling thing it did, read from its trace (a web search, then an MCP call, the computer, an artifact, an edit, a command, a search, a read, a listing, a display; the first line of its code when it did none of them), with the rest counted ("+3") and listed on hover and in its drawer. Calls closer than a readable chip (64 px) share one cluster chip that counts their kinds (as many as fit) and zooms to its stretch when pressed; clusters are worked out again as the zoom changes, and no chip runs past the next, so none overlaps. Jobs are packed by where they end on screen, at least 18 px wide, for the same reason. A fold's idle label stays off the ticks, and gives way to the person's "you" when their message ended it. Under the chips, the lane's background jobs (`GET /extensions/run/sessions/{id}/jobs/recent`, kept only while the kernel keeps them) on the same clock, packed into as few tracks as keep them apart; tracks past the first fold into a thin band of ticks, and pointing at them floats them out over the lanes below, so opening them never moves the layout. A running job shimmers, one that failed or timed out is red, one started as a service is dashed, and a click opens its terminal beside the trace. Jobs never keep idle time from folding. The gutter gives each lane's model and how much context its last call sent, and what share came from the cache; a compaction summary's call is drawn in the accent. Calls are often a few pixels wide, so the strip is taller than its bars and finds the call itself (`callAt`): the nearest within 10 px, a tool call close by beating the model call under the pointer. Spawn arrows leave the parent's tool call running when the child was created; reply arrows leave the child's last step before its mail landed. Above the lanes, one bar says where the task's own time went (`spent`): thinking, running code, waiting on agents, waiting on the person (an idle stretch their message ended), and the gaps between steps, with the duration, tokens, calls and failures beside it; all of it follows the playhead. +Rows come from three places: model calls from the session's request records (`GET /sessions/{id}/context?view=requests`, oldest first; `since_ms` finds the task’s first records, with single-row probes only on older daemons), tool calls from the transcript (a call's entry to its result's; input a history page only referenced is read afterwards in one `history/content` walk per agent), and agents from `GET /sessions?parent_id=`, three deep, kept when they were alive during the task. Each agent is one lane: its calls never overlap, so model calls (grey, the wait for the first byte lighter) and tool calls (the lane's hue, which keeps clear of red; failures are red) take turns along one strip, and a gap of 1.5 s or more while an agent below works is hatched as waiting. Under the strip, each tool call has one chip naming the most telling thing it did, read from its trace (a web search, then an MCP call, the computer, an artifact, an edit, a command, a search, a read, a listing, a display; the first line of its code when it did none of them), with the rest counted ("+3") and listed on hover and in its drawer. Calls closer than a readable chip (64 px) share one cluster chip that counts their kinds (as many as fit) and zooms to its stretch when pressed; clusters are worked out again as the zoom changes, and no chip runs past the next, so none overlaps. Jobs are packed by where they end on screen, at least 18 px wide, for the same reason. A fold's idle label stays off the ticks, and gives way to the person's "you" when their message ended it. Under the chips, the lane's background jobs (`GET /extensions/run/sessions/{id}/jobs/recent`, kept only while the kernel keeps them) on the same clock, packed into as few tracks as keep them apart; tracks past the first fold into a thin band of ticks, and pointing at them floats them out over the lanes below, so opening them never moves the layout. A running job shimmers, one that failed or timed out is red, one started as a service is dashed, and a click opens its terminal beside the trace. Jobs never keep idle time from folding. The gutter gives each lane's model and how much context its last call sent, and what share came from the cache; a compaction summary's call is drawn in the accent. Calls are often a few pixels wide, so the strip is taller than its bars and finds the call itself (`callAt`): the nearest within 10 px, a tool call close by beating the model call under the pointer. Spawn arrows leave the parent's tool call running when the child was created; reply arrows leave the child's last step before its mail landed. Above the lanes, one bar says where the task's own time went (`spent`): thinking, running code, waiting on agents, waiting on the person (an idle stretch their message ended), and the gaps between steps, with the duration, tokens, calls and failures beside it; all of it follows the playhead. A press on a call opens a drawer under its lane: what it did (each edit opens the review of its turn at that file, through `onreview`), its code and output (read then, in one batch), or a model call's first byte, tokens sent, cache share and tokens written; ← and → step along the lane, Esc closes. Hovering a tool call in either the trace or the chat lights it in both (`traceLink.entry`, keyed by the chat's entry: the result once there is one), and `TraceTie` draws a line between them over the canvas, pinned to the chat's edge when the call is scrolled out of view; opening one in the trace sets `traceLink.reveal`, which opens its stretch in the chat and scrolls it there. Any stretch of 20 s or more with nothing running is folded into a 2.5 s break labelled with how long it was idle (`clockOf`, a polylinear d3 scale); ticks are round elapsed times within each busy stretch. Zooming, panning and replaying only move CSS variables (`--w` the axis width, `--left` the pan, `--t` the playhead), so none re-renders a bar: a pinch zooms at the pointer, a sideways swipe pans (the canvas leaves both to anything inside `.zooms-itself` while the window is readable), a press on the axis scrubs, a double-click on it or "fit" shows the whole task, a double-click on a call zooms to it, and the replay plays any task in about 12 s. Search matches a call's name, the first 4000 characters of its input and its error; Enter opens each match in turn. `prefers-reduced-motion` drops every animation but the replay. diff --git a/src/albedo/daemon/http_sessions.gleam b/src/albedo/daemon/http_sessions.gleam index 5c2eda7d..130d8a83 100644 --- a/src/albedo/daemon/http_sessions.gleam +++ b/src/albedo/daemon/http_sessions.gleam @@ -957,6 +957,7 @@ pub fn context( "section_id", "page", "after", + "since_ms", "limit", "next", ]), @@ -1001,6 +1002,12 @@ pub fn context( 0, 9_007_199_254_740_991, )) + use since_ms <- result.try(http_api.integer_parameter( + parameters, + "since_ms", + 0, + 9_007_199_254_740_991, + )) use after <- result.try(http_api.page_offset( config.token, req, @@ -1009,7 +1016,7 @@ pub fn context( after, )) use page <- result.try( - requests.page(runtime.ledger(host), id, after, limit + 1) + requests.page(runtime.ledger(host), id, after, since_ms, limit + 1) |> result.map_error(http_api.failure), ) let shown = list.take(page.0, limit) @@ -1030,7 +1037,7 @@ pub fn context( use _ <- result.try( case list.any(parameters, fn(pair) { - list.contains(["after", "limit", "next"], pair.0) + list.contains(["after", "since_ms", "limit", "next"], pair.0) }) { True -> diff --git a/src/albedo/daemon/requests.gleam b/src/albedo/daemon/requests.gleam index a9747642..9c35d08c 100644 --- a/src/albedo/daemon/requests.gleam +++ b/src/albedo/daemon/requests.gleam @@ -462,11 +462,13 @@ pub type Row { ) } -/// Rows after `after`, oldest first, at most `limit`; the id to continue from. +/// Rows after `after` finished at or after `since_ms`, oldest first; +/// at most `limit` rows and the id to continue from. pub fn page( database: store.Store, session: String, after: Int, + since_ms: Int, limit: Int, ) -> Result(#(List(Row), Int), String) { store.snapshot(database, fn(db) { @@ -474,8 +476,13 @@ pub fn page( db, "SELECT " <> columns - <> " FROM provider_requests WHERE session=? AND id>? ORDER BY id LIMIT ?", - [sqlight.text(session), sqlight.int(after), sqlight.int(limit)], + <> " FROM provider_requests WHERE session=? AND id>? AND finished_ms>=? ORDER BY id LIMIT ?", + [ + sqlight.text(session), + sqlight.int(after), + sqlight.int(since_ms), + sqlight.int(limit), + ], row_decoder(), )) let next = diff --git a/test/e2e/requests_test.py b/test/e2e/requests_test.py index bea403db..6a3e71a1 100644 --- a/test/e2e/requests_test.py +++ b/test/e2e/requests_test.py @@ -10,7 +10,9 @@ of this is observable from events or history alone. import json import unittest +import urllib.error from datetime import datetime +from urllib.parse import urlencode from harness import Albedo, Provider, Reply, text @@ -275,6 +277,58 @@ class ProviderRequestsTest(unittest.TestCase): finally: provider.close() + def test_since_ms_filters_finished_attempts_and_binds_pagination(self): + provider = Provider(reply) + try: + with Albedo(provider, protocol="responses") as app: + session = app.session() + for prompt in ("first", "third", "hit the limit"): + app.prompt(session, prompt).close() + app.idle(session) + rows = self.rows(app, session)["items"] + self.assertEqual(len(rows), 3) + since_ms = round(instant(rows[1]["ended_at"]).timestamp() * 1000) + self.assertLess( + instant(rows[0]["ended_at"]), instant(rows[1]["ended_at"]) + ) + query = f"&since_ms={since_ms}&limit=1" + page = self.rows(app, session, query) + self.assertEqual(page["items"], [rows[1]]) + self.assertIsNotNone(page["next"]) + continuation = "&" + urlencode({"next": page["next"]}) + following = self.rows(app, session, query + continuation) + self.assertEqual(following["items"], [rows[2]]) + self.assertIsNone(following["next"]) + self.assertEqual(rows[2]["outcome"], "failed") + after = self.rows(app, session, query + f"&after={rows[1]['sequence']}") + self.assertEqual(after["items"], [rows[2]]) + ended_ms = round(instant(rows[2]["ended_at"]).timestamp() * 1000) + self.assertEqual( + self.rows(app, session, f"&since_ms={ended_ms + 1}")["items"], [] + ) + for invalid in ( + f"&since_ms={since_ms + 1}&limit=1" + continuation, + "&limit=1" + continuation, + "&since_ms=-1", + "&since_ms=1.5", + "&since_ms=9007199254740992", + ): + with self.subTest(query=invalid): + with self.assertRaises(urllib.error.HTTPError) as rejected: + self.rows(app, session, invalid) + self.assertEqual(rejected.exception.code, 400) + rejected.exception.close() + for view in ("summary", "section"): + with self.subTest(view=view): + with self.assertRaises(urllib.error.HTTPError) as rejected: + app.api( + f"/sessions/{session}/context?view={view}&since_ms=0" + ).close() + self.assertEqual(rejected.exception.code, 400) + rejected.exception.close() + finally: + provider.close() + def test_the_latest_usage_reports_its_calls_span_and_rate(self): provider = Provider(reply) try: diff --git a/web/scripts/trace.test.mjs b/web/scripts/trace.test.mjs new file mode 100644 index 00000000..10f0e9e1 --- /dev/null +++ b/web/scripts/trace.test.mjs @@ -0,0 +1,148 @@ +// Cursor merges and old-daemon negotiation run in the browser, outside daemon E2E. +import assert from 'node:assert/strict'; +import { readFile } from 'node:fs/promises'; +import { stripTypeScriptTypes } from 'node:module'; +import test from 'node:test'; + +const source = stripTypeScriptTypes((await readFile(new URL('../src/lib/trace.ts', import.meta.url), 'utf8')).replace(/^import .*;$/gm, '')); +let imports = 0; +const iso = (ms) => new Date(ms).toISOString(); +const session = (id = 'root') => ({ id, name: id, created_at: iso(1000), activity_at: iso(1000), status: { phase: 'generating' } }); +const entry = (position, kind, extra = {}) => ({ id: `e${position}`, position, kind, created_at: iso(1000 + position), turn_id: 'turn', turn_type: 'user', content: [], ...extra }); +const row = (sequence, kind = 'turn') => ({ sequence, kind, started_at: iso(1000), ended_at: iso(1100), model: 'model', prompt_tokens: {}, completion_tokens: {}, cached_tokens: {} }); + +async function fixture({ old = false } = {}) { + const calls = []; + const histories = new Map([['root', [entry(1, 'user'), entry(2, 'tool_call', { tool: { tool_call_id: 'call', name: 'python' } })]]]); + const requests = new Map([['root', [row(1)]]]); + const children = new Map(); + const content = new Map(); + let fail = false; + class ApiError extends Error { constructor(status) { super('unsupported'); this.status = status; } } + const api = { + ApiError, + busy: (s) => s?.status?.phase === 'generating', + jsonField: () => undefined, + readContentBatch: async (_id, ids) => { calls.push(['content', ids]); return new Map(ids.map((id) => [id, content.get(id)])); }, + listChildren: async (id) => { calls.push(['children', id]); return children.get(id) ?? []; }, + listJobs: async (id) => { calls.push(['jobs', id]); return []; }, + readHistory: async (id, options = {}) => { + calls.push(['history', id, options]); + if (fail) throw new Error('offline'); + const all = histories.get(id) ?? []; + const items = all.filter((e) => options.after === undefined || e.position > options.after); + const limit = options.limit ?? 100; + return { items: options.after === undefined ? items.slice(-limit) : items.slice(0, limit), newer: items.length > limit, older: false }; + }, + readRequests: async (id, after, limit, _signal, since) => { + calls.push(['requests', id, after, since]); + if (old && since !== undefined) throw new ApiError(400); + const rows = (requests.get(id) ?? []).filter((r) => r.sequence > after); + return { items: rows.slice(0, limit), next: rows.length > limit ? 'more' : null }; + }, + }; + globalThis.__traceAPI = api; + const prelude = 'const {ApiError,busy,jsonField,listChildren,listJobs,readHistory,readRequests,readContentBatch} = globalThis.__traceAPI;'; + const mod = await import(`data:text/javascript;base64,${Buffer.from(prelude + source).toString('base64')}#${++imports}`); + delete globalThis.__traceAPI; + const cursor = new mod.TraceCursor(); + const read = (from = 1, root = session()) => mod.readTrace(root, from, 1, true, undefined, undefined, cursor); + return { calls, histories, requests, children, content, details: mod.details, cursor, read, fail: () => { fail = true; } }; +} + +test('refresh reads after cursors, closes a pending tool, adds marks, and advances past background calls', async () => { + const f = await fixture(); + const first = await f.read(); + assert.equal(first.lanes[0].steps.find((s) => s.kind === 'tool').open, true); + f.histories.get('root').push(entry(3, 'tool_result', { tool: { tool_call_id: 'call' } }), entry(4, 'user', { turn_id: 'next' })); + f.requests.get('root').push(row(2), row(3, 'background')); + f.calls.length = 0; + const next = await f.read(); + assert.equal(next.lanes[0].steps.length, 3); + assert.equal(next.lanes[0].steps.find((s) => s.kind === 'tool').result, 'e3'); + assert.equal(next.marks.length, 1); + assert.equal(f.calls.find((c) => c[0] === 'history')[2].after, 2); + assert.equal(f.calls.find((c) => c[0] === 'requests')[2], 1); + f.calls.length = 0; + await f.read(); + assert.equal(f.calls.find((c) => c[0] === 'requests')[2], 3); +}); + +test('discovers new children and retains completed children without rereading their work', async () => { + const f = await fixture(); + await f.read(); + const child = { ...session('child'), activity_at: iso(1100), status: { phase: 'idle' } }; + f.children.set('root', [child]); + f.histories.set('child', [entry(1, 'tool_call', { tool: { tool_call_id: 'c' } }), entry(2, 'tool_result', { tool: { tool_call_id: 'c' } })]); + const next = await f.read(); + assert.equal(next.lanes.length, 2); + f.calls.length = 0; + assert.equal((await f.read()).lanes.length, 2); + assert.equal(f.calls.some((c) => c[1] === 'child'), false); +}); + +test('negotiates an old daemon once across lanes and uses sequence reads thereafter', async () => { + const f = await fixture({ old: true }); + f.children.set('root', [session('child')]); + await f.read(); + await f.read(); + assert.equal(f.calls.filter((c) => c[0] === 'requests' && c[3] !== undefined).length, 1); +}); + +test('earlier turns and failed reads discard cursors', async () => { + const f = await fixture(); + await f.read(); + f.calls.length = 0; + await f.read(0); + assert.equal(f.calls.find((c) => c[0] === 'history')[2].after, -1); + f.fail(); + await assert.rejects(f.read(0), /offline/); + assert.equal(f.cursor.lanes.size, 0); +}); + +test('a followed idle parent includes its still-working child on the first read and refresh', async () => { + const f = await fixture(); + f.children.set('root', [{ ...session('child'), activity_at: iso(1005) }]); + f.histories.set('child', [entry(5, 'tool_call', { tool: { tool_call_id: 'c' } })]); + f.requests.set('child', [{ ...row(2), started_at: iso(1004), ended_at: iso(1010) }]); + const root = { ...session(), status: { phase: 'idle' } }; + const first = await f.read(1, root); + const next = await f.read(1, root); + assert.equal(first.lanes[1].steps.length, 2); + assert.equal(next.lanes[1].steps.length, 2); + assert.equal(first.lanes[1].steps[0].id, next.lanes[1].steps[0].id); +}); + +test('long tasks paginate once, then request only the tail', async () => { + const f = await fixture(); + f.histories.set('root', Array.from({ length: 401 }, (_, i) => entry(i + 1, i ? 'assistant' : 'user'))); + f.requests.set('root', Array.from({ length: 401 }, (_, i) => row(i + 1))); + const first = await f.read(); + assert.equal(first.lanes[0].steps.length, 401); + assert.equal(f.calls.filter((c) => c[0] === 'history').length, 3); + assert.equal(f.calls.filter((c) => c[0] === 'requests').length, 3); + f.calls.length = 0; + const next = await f.read(); + assert.equal(next.lanes[0].steps.length, 401); + assert.equal(f.calls.filter((c) => c[0] === 'history').length, 1); + assert.equal(f.calls.filter((c) => c[0] === 'requests').length, 1); + assert.equal(f.calls.find((c) => c[0] === 'history')[2].after, 401); + assert.equal(f.calls.find((c) => c[0] === 'requests')[2], 401); +}); + +test('hydration keeps an earlier input when its result arrives and skips filled details', async () => { + const f = await fixture(); + const step = { id: 'tool', entry: 'call' }; + const trace = { lanes: [{ id: 'root', steps: [step] }] }; + f.content.set('call', { arguments: '{"code":"print(1)"}' }); + const known = await f.details(trace); + assert.equal(known.get('tool').input.line, 'print(1)'); + step.traced = 'result'; + f.content.set('result', { trace: '{"activities":[],"changes":[]}' }); + const completed = await f.details(trace, undefined, known); + assert.equal(completed.get('tool').input.line, 'print(1)'); + assert.deepEqual(completed.get('tool').did, { activities: [], changes: [] }); + f.calls.length = 0; + assert.equal((await f.details(trace, undefined, completed)).size, 0); + assert.equal(f.calls.length, 0); +}); diff --git a/web/src/lib/api.gen.d.ts b/web/src/lib/api.gen.d.ts index abdee457..14190e44 100644 --- a/web/src/lib/api.gen.d.ts +++ b/web/src/lib/api.gen.d.ts @@ -371,7 +371,7 @@ export interface paths { }; /** * Inspect prepared context or request records - * @description Summary never prepares a request. Section requires snapshot_id, section_id, page and forbids request-pagination fields. Requests accepts after/limit/next. Stale snapshot is 410 context_changed, never another section. Requires capability `context: 1`. + * @description Summary never prepares a request. Section requires snapshot_id, section_id, page and forbids request-pagination fields. Requests accepts after/since_ms/limit/next. Stale snapshot is 410 context_changed, never another section. Requires capability `context: 1`. */ get: operations["getContext"]; put?: never; @@ -5791,6 +5791,8 @@ export interface operations { section_id?: components["schemas"]["ID"]; page?: components["schemas"]["Counter"]; after?: components["schemas"]["Counter"]; + /** @description Requests only. Include attempts finished at or after this Unix timestamp in milliseconds. */ + since_ms?: components["schemas"]["Counter"]; limit?: components["parameters"]["Limit"]; next?: components["parameters"]["Next"]; }; diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index 6998b196..86b53b8f 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -71,8 +71,9 @@ export type RequestRecord = components['schemas']['ProviderRequestRecord']; * A page of a session's provider requests, oldest first: every model call * with its timing and tokens. `after` is a request's sequence. */ -export function readRequests(sessionId: string, after = 0, limit = 200, signal?: AbortSignal): Promise<{ items: RequestRecord[]; next: string | null }> { +export function readRequests(sessionId: string, after = 0, limit = 200, signal?: AbortSignal, since?: number): Promise<{ items: RequestRecord[]; next: string | null }> { const query = new URLSearchParams({ view: 'requests', after: String(after), limit: String(limit) }); + if (since !== undefined) query.set('since_ms', String(since)); return get(`/sessions/${sessionId}/context?${query}`, signal); } diff --git a/web/src/lib/trace.ts b/web/src/lib/trace.ts index 28056d3e..5b867a81 100644 --- a/web/src/lib/trace.ts +++ b/web/src/lib/trace.ts @@ -1,4 +1,4 @@ -import { listChildren, listJobs, readContentBatch, readHistory, readRequests, type HistoryEntry, type Job, type RequestRecord, type Session } from './api'; +import { ApiError, listChildren, listJobs, readContentBatch, readHistory, readRequests, type HistoryEntry, type Job, type RequestRecord, type Session } from './api'; import type { Trace as CallTrace } from './burst'; import { scaleLinear } from 'd3-scale'; import { busy } from './live.svelte'; @@ -222,7 +222,7 @@ const PROBES = 8; * finds where the window starts by probing single rows, several at once, * rather than reading every page before it. */ -async function requestsStart(sessionId: string, start: number, signal?: AbortSignal): Promise { +async function probeRequestsStart(sessionId: string, start: number, signal?: AbortSignal): Promise { const first = await readRequests(sessionId, 0, PAGE, signal); const last = first.items.at(-1); if (!last || !first.next || ended(last) >= start) return { rows: first.items, next: first.next && last ? last.sequence : null }; @@ -249,18 +249,27 @@ async function requestsStart(sessionId: string, start: number, signal?: AbortSig return { rows: [], next: low }; } -/** A session's model calls that overlap a window, read on from where `requestsStart` found it begins. */ -async function requestsBetween(from: Promise, sessionId: string, start: number, end: number, signal?: AbortSignal): Promise { - const found = await from; - const rows = [...found.rows]; - let next = found.rows.length && at(found.rows.at(-1)!.started_at) > end ? null : found.next; - while (next !== null) { - const page = await readRequests(sessionId, next, PAGE, signal); - rows.push(...page.items); - const tail = page.items.at(-1); - next = page.next && tail && at(tail.started_at) <= end ? tail.sequence : null; +// Capability detection is shared by all lanes, including simultaneous first reads. +let sinceSupported: boolean | undefined; +let checkingSince: Promise | undefined; + +async function requestsStart(sessionId: string, start: number, signal?: AbortSignal): Promise { + if (checkingSince) await checkingSince; + if (sinceSupported === false) return probeRequestsStart(sessionId, start, signal); + const read = readRequests(sessionId, 0, PAGE, signal, Math.max(0, Math.floor(start))); + if (sinceSupported === undefined) checkingSince = read.then(() => { sinceSupported = true; }, (error) => { + if (error instanceof ApiError && error.status === 400) sinceSupported = false; + }); + try { + const page = await read; + return { rows: page.items, next: page.next ? page.items.at(-1)!.sequence : null }; + } catch (error) { + if (!(error instanceof ApiError) || error.status !== 400) throw error; + sinceSupported = false; + return probeRequestsStart(sessionId, start, signal); + } finally { + checkingSince = undefined; } - return rows.filter((row) => row.kind !== 'background' && at(row.started_at) <= end && ended(row) >= start); } /** A provider's failure as a sentence: the message inside a JSON error body, or the text as it came. */ @@ -381,39 +390,130 @@ function jobsBetween(jobs: Job[], start: number, end: number, now: number): JobS .sort((a, b) => a.start - b.start); } +interface LaneCursor { + entries: HistoryEntry[]; + rows: RequestRecord[]; + sequence: number; + children?: Session[]; + activity?: string | null; + running?: boolean; + jobs?: Job[]; + tools?: Step[]; + processed?: number; +} + +/** Cursors belong to one trace window and are dropped after a failed read. */ +export class TraceCursor { + key = ''; + lanes = new Map(); + + reset() { + this.key = ''; + this.lanes.clear(); + } +} + +function mergedTools(id: string, lane: LaneCursor, entries: HistoryEntry[], now: number): Step[] { + const fresh = entries.filter((entry) => entry.position > (lane.processed ?? -1)); + const pending = new Set(lane.tools?.filter((step) => step.open).map((step) => step.call)); + const calls = entries.filter((entry) => pending.has(entry.id)); + const updated = toolSteps(id, [...calls, ...fresh], now); + const known = new Map(lane.tools?.map((step) => [step.id, step])); + const starts = new Map(); + for (const entry of entries) if (entry.turn_id && !starts.has(entry.turn_id)) starts.set(entry.turn_id, entry.position); + for (const step of updated) { + if (step.turn) step.turn.from = starts.get(step.turn.id)!; + known.set(step.id, step); + } + lane.processed = entries.at(-1)?.position; + lane.tools = [...known.values()]; + return lane.tools; +} + +function cursorFor(cursor: TraceCursor, id: string): LaneCursor { + let lane = cursor.lanes.get(id); + if (!lane) { + lane = { entries: [], rows: [], sequence: -1 }; + cursor.lanes.set(id, lane); + } + return lane; +} + +async function appendHistory(id: string, lane: LaneCursor, signal?: AbortSignal): Promise { + const fresh = await readTask(id, (lane.entries.at(-1)?.position ?? 0) + 1, null, signal); + lane.entries = [...lane.entries, ...fresh.entries]; + return lane.entries; +} + +function nextRequests(id: string, lane: LaneCursor, start: number, signal?: AbortSignal): Promise { + if (lane.sequence < 0) return requestsStart(id, start, signal); + return readRequests(id, lane.sequence, PAGE, signal).then((page) => ({ rows: page.items, next: page.next ? page.items.at(-1)!.sequence : null })); +} + +async function appendRequests(id: string, lane: LaneCursor, start: number, end: number, signal?: AbortSignal, first = nextRequests(id, lane, start, signal)): Promise { + const found = await first; + let sequence = Math.max(0, lane.sequence); + const track = (rows: RequestRecord[]) => { + for (const row of rows) sequence = Math.max(sequence, row.sequence); + return rows; + }; + track(found.rows); + // Keep background calls out of the lane, but advance past them too. + const rows = [...found.rows]; + let next = found.rows.length && at(found.rows.at(-1)!.started_at) > end ? null : found.next; + while (next !== null) { + const page = await readRequests(id, next, PAGE, signal); + rows.push(...track(page.items)); + const tail = page.items.at(-1); + next = page.next && tail && at(tail.started_at) <= end ? tail.sequence : null; + } + lane.sequence = sequence; + lane.rows = [...lane.rows, ...rows.filter((row) => row.kind !== 'background' && ended(row) >= start)]; + return lane.rows.filter((row) => at(row.started_at) <= end); +} + /** * Reads the trace of `turns` turns from `from` in a session. One that * reaches the newest turn follows the session (`follows` undefined asks * which it is); one that ends on an older turn stays as it is. `onroot` * gets the session's own lane as soon as it is read, before its agents. */ -export async function readTrace( +async function readTraceFrom( session: Session, from: number, turns: number, follows: boolean | undefined, signal?: AbortSignal, onroot?: (trace: Trace) => void, + cursor = new TraceCursor(), ): Promise { + const key = `${session.id}:${from}:${turns}`; + if (cursor.key !== key || follows === false) cursor.reset(); + cursor.key = key; + const saved = cursorFor(cursor, session.id); const now = Date.now(); // What does not wait on the task's history starts with it: the agents, // the jobs, and the search for where its model calls begin, once the // first page says when the task opened. - const children = listChildren(session.id, signal); + const children = listChildren(session.id, signal).then((children) => { saved.children = children; return children; }); const jobs = listJobs(session.id, signal).catch((): Job[] => []); let started = (_time: number) => {}; - const begun = new Promise((resolve) => (started = resolve)).then((time) => requestsStart(session.id, time - 1000, signal)); + const begun = new Promise((resolve) => (started = resolve)).then((time) => nextRequests(session.id, saved, time - 1000, signal)); // Read as a task, the newest turns already hold everything after `from`. - const { entries, newest } = await readTask(session.id, from, follows ? null : turns, signal, started); + const incremental = saved.entries.length > 0 && follows === true; + const { entries, newest } = incremental + ? { entries: await appendHistory(session.id, saved, signal), newest: true } + : await readTask(session.id, from, follows ? null : turns, signal, started); + saved.entries = entries; const following = follows ?? newest; const opened = at(entries[0]?.created_at); started(opened || now); // A followed session that works is still going; otherwise the task ends with its last entry. const closed = entries.reduce((latest, entry) => Math.max(latest, at(entry.created_at) || 0), opened); const going = following && busy(session); - const window = { start: opened, end: going ? now : closed }; - const rows = await requestsBetween(begun, session.id, window.start - 1000, window.end + 1000, signal); - const steps = [...rows.map((row) => modelStep(session.id, row, now)), ...toolSteps(session.id, entries, now)]; + const window = { start: opened, end: going ? now : closed, follows: following }; + const rows = await appendRequests(session.id, saved, window.start - 1000, window.end + 1000, signal, begun); + const steps = [...rows.map((row) => modelStep(session.id, row, now)), ...mergedTools(session.id, saved, entries, now)]; const root = { ...laneOf(session, null, 0, steps, jobsBetween(await jobs, window.start, window.end, now)), busy: going }; window.end = Math.max(window.end, root.end); @@ -433,7 +533,7 @@ export async function readTrace( const lanes: Lane[] = [root]; const links: Link[] = []; const mail = entries.filter((entry) => entry.mail?.sender_session_id); - await agentsOf(root, children, window, now, lanes, links, signal); + await agentsOf(root, children, window, now, lanes, links, cursor, signal); for (const lane of lanes) { if (lane === root) continue; // A reply: the agent's mail landing, and the parent's next step. @@ -450,8 +550,26 @@ export async function readTrace( return assemble(lanes, links); } +/** Reads new rows from a window's cursors; a failed read discards them. */ +export async function readTrace( + session: Session, + from: number, + turns: number, + follows: boolean | undefined, + signal?: AbortSignal, + onroot?: (trace: Trace) => void, + cursor = new TraceCursor(), +): Promise { + try { + return await readTraceFrom(session, from, turns, follows, signal, onroot, cursor); + } catch (error) { + cursor.reset(); + throw error; + } +} + /** Adds the agents a lane started inside the window, depth first, with their spawn links. */ -async function agentsOf(parent: Lane, listed: Promise, window: { start: number; end: number }, now: number, lanes: Lane[], links: Link[], signal?: AbortSignal) { +async function agentsOf(parent: Lane, listed: Promise, window: { start: number; end: number; follows: boolean }, now: number, lanes: Lane[], links: Link[], cursor: TraceCursor, signal?: AbortSignal) { if (parent.depth >= DEPTH) return; const children = (await listed).filter((child) => { const born = at(child.created_at); @@ -459,15 +577,30 @@ async function agentsOf(parent: Lane, listed: Promise, window: { star }); const read = await Promise.all( children.map(async (child) => { - const span = { start: Math.max(window.start, at(child.created_at)), end: window.end }; + const end = window.follows ? Math.max(window.end, at(child.activity_at) || 0, busy(child) ? now : 0) : window.end; + const span = { start: Math.max(window.start, at(child.created_at)), end }; // Its own agents are listed while it is read. - const below = parent.depth + 1 < DEPTH ? listChildren(child.id, signal) : Promise.resolve([]); + const saved = cursorFor(cursor, child.id); + const activity = child.activity_at ?? child.created_at; + const changed = saved.activity !== activity || busy(child) || saved.running || !!saved.tools?.some((step) => step.open); + const discover = changed || !saved.children || saved.children.some(busy); + const below = parent.depth + 1 < DEPTH && discover + ? listChildren(child.id, signal) + : Promise.resolve(saved.children ?? []); + let history = Promise.resolve(saved.entries); + if (!saved.entries.length) history = readWindow(child.id, span.start, span.end, signal); + else if (changed) history = appendHistory(child.id, saved, signal); const [entries, rows, jobs] = await Promise.all([ - readWindow(child.id, span.start, span.end, signal), - requestsBetween(requestsStart(child.id, span.start, signal), child.id, span.start, span.end, signal), - listJobs(child.id, signal).catch((): Job[] => []), + history, + changed ? appendRequests(child.id, saved, span.start, span.end, signal) : Promise.resolve(saved.rows), + changed || saved.jobs?.some((job) => job.state === 'running') ? listJobs(child.id, signal) : Promise.resolve(saved.jobs ?? []), ]); - return { child, below, jobs: jobsBetween(jobs, span.start, span.end, now), steps: [...rows.map((row) => modelStep(child.id, row, now)), ...toolSteps(child.id, entries, now)] }; + saved.entries = entries; + saved.children = await below; + saved.activity = activity; + saved.running = busy(child); + saved.jobs = jobs; + return { child, below, jobs: jobsBetween(jobs, span.start, span.end, now), steps: [...rows.map((row) => modelStep(child.id, row, now)), ...mergedTools(child.id, saved, entries.filter((entry) => at(entry.created_at) <= span.end), now)] }; }), ); for (const { child, below, steps, jobs } of read.sort((a, b) => at(a.child.created_at) - at(b.child.created_at))) { @@ -478,7 +611,7 @@ async function agentsOf(parent: Lane, listed: Promise, window: { star const born = at(child.created_at); const spawner = parent.steps.find((step) => step.kind === 'tool' && step.start <= born && step.end >= born - 500); if (spawner) links.push({ id: `spawn:${lane.id}`, from: spawner.id, to: lane.steps[0].id, kind: 'spawn' }); - await agentsOf(lane, below, window, now, lanes, links, signal); + await agentsOf(lane, below, window, now, lanes, links, cursor, signal); } } @@ -492,11 +625,11 @@ export interface Details { * The parts of tool calls the history pages only referenced, by step id: * read after the trace shows, in one walk per agent. */ -export async function details(trace: Trace, signal?: AbortSignal): Promise> { +export async function details(trace: Trace, signal?: AbortSignal, known?: ReadonlyMap): Promise> { const found = new Map(); await Promise.all( trace.lanes.map(async (lane) => { - const wanted = lane.steps.filter((step) => step.entry || step.traced); + const wanted = lane.steps.filter((step) => (step.entry && !known?.get(step.id)?.input) || (step.traced && !known?.get(step.id)?.did)); for (let index = 0; index < wanted.length; index += 100) { const batch = wanted.slice(index, index + 100); const ids = batch.flatMap((step) => [step.entry, step.traced].filter((id): id is string => !!id)); @@ -504,7 +637,7 @@ export async function details(trace: Trace, signal?: AbortSignal): Promise { if (first && !signal.aborted) trace = root; - }); + }, cursor); if (signal.aborted) return; const known = new Set(steps.map((step) => step.id)); const fresh = next.lanes.flatMap((lane) => lane.steps.map((step) => step.id)).filter((id) => !known.has(id)); @@ -125,7 +126,7 @@ trace = next; failure = ''; if (next.lanes.some((lane) => lane.steps.some((step) => (step.entry || step.traced) && !fetched.has(step.id)))) - details(next, signal).then((found) => found.forEach((detail, id) => fetched.set(id, detail)), () => {}); + details(next, signal, fetched).then((found) => found.forEach((detail, id) => fetched.set(id, detail)), () => {}); if (!reduced && fresh.length) setTimeout(() => fresh.forEach((id) => entering.delete(id)), 1100); } catch (error) { if (!signal.aborted) failure = String((error as Error).message ?? error);