diff --git a/apps/aflow-executor-host/src/__tests__/harnessEvents.test.ts b/apps/aflow-executor-host/src/__tests__/harnessEvents.test.ts index 8423cdcf..5f8b5acf 100644 --- a/apps/aflow-executor-host/src/__tests__/harnessEvents.test.ts +++ b/apps/aflow-executor-host/src/__tests__/harnessEvents.test.ts @@ -241,3 +241,94 @@ describe('what a tool call reads as', () => { }); }); }); + +/** + * Proof of life: what the agent says it is doing, the files it changes, and + * which results are a command's exit status — read off the fields of each call, + * so a renamed tool costs nothing. + */ +describe('what a watcher reads first', () => { + function readCalls( + calls: Array<{ id: string; name: string; input: unknown }>, + results: Array<{ id: string; isError: boolean }> = [], + ): HarnessActivityLine[] { + const lines: HarnessActivityLine[] = []; + const reader = createHarnessEventReader('claude-stream-json', (l) => lines.push(l), 0); + for (const call of calls) { + reader.push( + `${JSON.stringify({ + type: 'assistant', + message: { content: [{ type: 'tool_use', ...call }] }, + })}\n`, + ); + } + for (const result of results) { + reader.push( + `${JSON.stringify({ + type: 'user', + message: { + content: [ + { + type: 'tool_result', + tool_use_id: result.id, + content: 'out', + is_error: result.isError, + }, + ], + }, + })}\n`, + ); + } + return lines; + } + + it('names the file a call changes, and not one it only reads', () => { + const lines = readCalls([ + { id: 'a', name: 'Edit', input: { file_path: 'src/a.ts', old_string: 'x', new_string: 'y' } }, + { id: 'b', name: 'Write', input: { file_path: 'src/b.ts', content: 'z' } }, + { id: 'c', name: 'Read', input: { file_path: 'src/c.ts' } }, + { id: 'd', name: 'Grep', input: { pattern: 'parse', path: 'src' } }, + ]); + expect(lines.map((l) => (l.kind === 'tool' ? (l.changes ?? null) : 'other'))).toEqual([ + 'src/a.ts', + 'src/b.ts', + null, + null, + ]); + }); + + it('says what the agent marked in progress, once per change of plan', () => { + const plan = (active: string) => ({ + todos: [ + { content: 'Read the parser', status: 'completed', activeForm: 'Reading the parser' }, + { content: active, status: 'in_progress', activeForm: `${active}…` }, + ], + }); + const lines = readCalls([ + { id: 'a', name: 'TodoWrite', input: plan('Run the tests') }, + { id: 'b', name: 'TodoWrite', input: plan('Run the tests') }, + { id: 'c', name: 'TodoWrite', input: plan('Fix the guard') }, + ]); + expect(lines.filter((l) => l.kind === 'intent').map((l) => l.text)).toEqual([ + 'Run the tests…', + 'Fix the guard…', + ]); + }); + + it("marks a command's result as one, so its exit status can be judged apart", () => { + const lines = readCalls( + [ + { id: 'a', name: 'Bash', input: { command: 'yarn test' } }, + { id: 'b', name: 'Read', input: { file_path: 'missing.ts' } }, + ], + [ + { id: 'a', isError: true }, + { id: 'b', isError: true }, + ], + ); + const results = lines.filter((l) => l.kind === 'tool_result'); + expect(results[0]).toMatchObject({ tool: 'Bash', ok: false, command: true }); + expect(results[1]).toMatchObject({ tool: 'Read', ok: false }); + expect(results[1]).not.toHaveProperty('command'); + }); +}); diff --git a/apps/aflow-executor-host/src/harnessEvents.ts b/apps/aflow-executor-host/src/harnessEvents.ts index ddea318c..fb4fdeeb 100644 --- a/apps/aflow-executor-host/src/harnessEvents.ts +++ b/apps/aflow-executor-host/src/harnessEvents.ts @@ -83,6 +83,50 @@ export function summarizeToolUse(name: string, input: unknown): string { return name; } +/** + * The file a call changes, when it changes one. + * + * Fields again, not names: a call that names a path and carries what to put + * there is a change, whatever the harness calls the tool. A call naming a path + * alone is a read. + */ +const PATH_FIELDS = ['file_path', 'notebook_path', 'path'] as const; +const CONTENT_FIELDS = ['content', 'new_string', 'edits', 'new_source'] as const; + +export function changedFileOf(input: unknown): string | undefined { + if (!isRecord(input)) return undefined; + if (!CONTENT_FIELDS.some((field) => input[field] !== undefined)) return undefined; + for (const field of PATH_FIELDS) { + const value = input[field]; + if (typeof value === 'string' && value.trim() !== '') return clip(value, SUMMARY_MAX); + } + return undefined; +} + +/** A call that hands the shell a command line. */ +export function isCommandCall(input: unknown): boolean { + return isRecord(input) && typeof input['command'] === 'string'; +} + +/** + * What the agent says it is doing: the item of its plan it marked in progress. + * + * A harness that keeps a plan rewrites the whole of it on every change, so the + * call carries every item and the one in progress is the news. Its present-tense + * wording is preferred where the harness writes one. + */ +export function intentOf(input: unknown): string | undefined { + if (!isRecord(input) || !Array.isArray(input['todos'])) return undefined; + for (const item of input['todos']) { + if (!isRecord(item) || item['status'] !== 'in_progress') continue; + for (const field of ['activeForm', 'content'] as const) { + const value = item[field]; + if (typeof value === 'string' && value.trim() !== '') return clip(value, SUMMARY_MAX); + } + } + return undefined; +} + /** A tool result arrives as a string or as content blocks; both read as text. */ function textOfContent(content: unknown): string { if (typeof content === 'string') return content; @@ -125,8 +169,10 @@ export function createHarnessEventReader( const passedThrough: string[] = []; // A tool result names the call it answers, not the tool that made it, so the // name is carried across from the call. Without it a feed says a result came - // back and never which of five in-flight calls it belongs to. - const toolNames = new Map(); + // back and never which of five in-flight calls it belongs to. Whether the + // call ran a command travels with it for the same reason. + const toolCalls = new Map(); + let lastIntent = ''; const at = (): number => Math.max(0, Date.now() - startedAt); @@ -152,14 +198,24 @@ export function createHarnessEventReader( } if (block['type'] === 'tool_use' && typeof block['name'] === 'string') { const name = block['name']; + const input = block['input']; const id = block['id']; - if (typeof id === 'string') toolNames.set(id, name); + if (typeof id === 'string') toolCalls.set(id, { name, command: isCommandCall(input) }); + const changes = changedFileOf(input); onLine({ kind: 'tool', at: at(), tool: name, - text: summarizeToolUse(name, block['input']), + text: summarizeToolUse(name, input), + ...(changes !== undefined ? { changes } : {}), }); + // A plan rewritten without moving on says nothing new, and repeating + // the same intent would date it as fresh news. + const intent = intentOf(input); + if (intent !== undefined && intent !== lastIntent) { + lastIntent = intent; + onLine({ kind: 'intent', at: at(), text: intent }); + } } } }; @@ -170,13 +226,14 @@ export function createHarnessEventReader( for (const block of content) { if (!isRecord(block) || block['type'] !== 'tool_result') continue; const id = block['tool_use_id']; - const tool = (typeof id === 'string' ? toolNames.get(id) : undefined) ?? 'tool'; + const call = typeof id === 'string' ? toolCalls.get(id) : undefined; onLine({ kind: 'tool_result', at: at(), - tool, + tool: call?.name ?? 'tool', ok: block['is_error'] !== true, text: summarizeToolResult(block['content']), + ...(call?.command === true ? { command: true } : {}), }); } }; diff --git a/apps/aflow-orchestrator/src/services/ProjectionWorker.ts b/apps/aflow-orchestrator/src/services/ProjectionWorker.ts index f332c027..c03776cb 100644 --- a/apps/aflow-orchestrator/src/services/ProjectionWorker.ts +++ b/apps/aflow-orchestrator/src/services/ProjectionWorker.ts @@ -1,3 +1,4 @@ +import { randomUUID } from 'node:crypto'; import type { Redis } from 'ioredis'; import type postgres from 'postgres'; import type { PostgresJsDatabase } from 'drizzle-orm/postgres-js'; @@ -16,6 +17,7 @@ import { compareStreamIds, deleteRecoveryData, publishActionCenterWake, + appendEntityEvent, type ProjectionCandidate, type SessionHotState, type SessionEvent, @@ -145,6 +147,17 @@ const EVENT_FLUSH_PAGE = 1000; const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; +/** + * Whether a persisted event moved the session between statuses — a turn + * beginning, resting, resuming or ending. Every such event is a `Session*` + * lifecycle event; step and run events move nothing a conversation list shows. + */ +export function movesSessionStatus( + event: Pick | null | undefined, +): boolean { + return event?.eventType.startsWith('Session') === true; +} + function isUuid(value: string): boolean { return UUID_RE.test(value); } @@ -278,7 +291,7 @@ export function createProjectionWorker( tx: PostgresJsDatabase, tenantId: string, runId: string, - ): Promise<{ evictedGap: boolean }> { + ): Promise<{ evictedGap: boolean; statusMoved: boolean }> { const [row] = await tx .select({ cursor: sessions.lastFlushedEventStreamId, status: sessions.status }) .from(sessions) @@ -297,6 +310,7 @@ export function createProjectionWorker( let cursor = initialCursor; const now = new Date(); let evictedGap = false; + let statusMoved = false; for (;;) { const { entries, lastId, oldestId } = await readDurableEventEntries( @@ -340,6 +354,7 @@ export function createProjectionWorker( } } if (entries.length > 0) { + if (!statusMoved) statusMoved = entries.some((entry) => movesSessionStatus(entry.event)); await tx .insert(eventLog) .values( @@ -359,7 +374,46 @@ export function createProjectionWorker( .set({ lastFlushedEventStreamId: cursor }) .where(eq(sessions.sessionId, runId)); } - return { evictedGap }; + return { evictedGap, statusMoved }; + } + + /** + * Tell the space a conversation's status moved, once the move is durable. + * + * Driven by the lifecycle events this projection just persisted rather than + * by the projection itself, which runs on every step: a turn of forty tool + * calls is two signals — it began, it rested — not forty. Top-level sessions + * only, because those are the conversations a list shows; a child's turn is + * its parent's. Best-effort: the list refetches on its own staleness. + */ + async function publishSessionStatus( + tenantId: string, + runId: string, + runState: SessionHotState, + ): Promise { + if (!runState.spaceId || runState.parentSessionId) return; + try { + await appendEntityEvent(redis, { + tenantId, + spaceId: runState.spaceId, + event: { + eventId: randomUUID(), + eventType: 'entity.session.status', + spaceId: runState.spaceId, + tenantId, + timestamp: Date.now(), + causedBySessionId: runId, + payload: { sessionId: runId, status: runState.status }, + summary: `Conversation ${runState.status.toLowerCase()}`, + }, + }); + } catch (err) { + log.warn(`[ProjectionWorker] session status signal failed for run ${runId}`, { + tenantId, + runId, + error: err instanceof Error ? err.message : String(err), + }); + } } // ── Full flush (terminal runs) ────────────────────────────────────────── @@ -378,7 +432,7 @@ export function createProjectionWorker( // all, and the loss is silent until someone tries. const hotStateSnapshot = await buildHotStateSnapshot(redis, tenantId, runState); let fenceWon = false; - await withTenantSchema(db, tenantContext, async (tx) => { + const statusMoved = await withTenantSchema(db, tenantContext, async (tx) => { const upserted = await tx .insert(sessions) .values({ @@ -438,7 +492,8 @@ export function createProjectionWorker( }) .returning({ sessionId: sessions.sessionId }); fenceWon = upserted.length > 0; - if (fenceWon) await flushDurableEvents(tx, tenantId, runId); + if (!fenceWon) return false; + return (await flushDurableEvents(tx, tenantId, runId)).statusMoved; }); if (!fenceWon) { @@ -472,6 +527,7 @@ export function createProjectionWorker( await bumpAttentionGeneration(redis, tenantId, runState.spaceId); } } + if (statusMoved) await publishSessionStatus(tenantId, runId, runState); // After the upsert, because the mark it claims lives on the row that upsert // writes; not swallowed, because a throw here leaves the candidate @@ -617,9 +673,10 @@ export function createProjectionWorker( // from the status upsert, so a failing insert leaves the status row // current — the throw still withholds the acknowledgement, which is what // re-drives the events. - await withTenantSchema(db, tenantContext, async (tx) => { - await flushDurableEvents(tx, tenantId, runId); - }); + const { statusMoved } = await withTenantSchema(db, tenantContext, (tx) => + flushDurableEvents(tx, tenantId, runId), + ); + if (statusMoved) await publishSessionStatus(tenantId, runId, runState); // The candidate is acknowledged by the caller; a later mutation re-arms it. // Leaving it armed instead would re-upsert every running session every diff --git a/apps/aflow-orchestrator/src/services/__tests__/projectionSessionStatus.test.ts b/apps/aflow-orchestrator/src/services/__tests__/projectionSessionStatus.test.ts new file mode 100644 index 00000000..8e3451e8 --- /dev/null +++ b/apps/aflow-orchestrator/src/services/__tests__/projectionSessionStatus.test.ts @@ -0,0 +1,174 @@ +/** + * The space hears `entity.session.status` when a conversation's turn begins, + * rests or ends — once the move is durable, and only for a move: the + * projection runs on every step, and a turn of forty tool calls is two signals, + * not forty. It is what lets the Workbench's conversation list say which + * conversations are live without a subscription per row (Plan 315 F75). + */ +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import type { SessionHotState } from '@aflow/redis'; + +const TENANT = 'a0000000-0000-4000-8000-000000000915'; +const SPACE = '5e1d0000-0000-4000-8000-000000000915'; +const SESSION = 'c01d0000-0000-4000-8000-000000000915'; +const PARENT = 'c01d0000-0000-4000-8000-0000000009aa'; + +const hot = vi.hoisted(() => ({ + state: undefined as SessionHotState | undefined, + /** The event types in the stream past the projection's cursor. */ + pending: [] as string[], + signals: [] as Array<{ spaceId: string; eventType: string; payload: unknown }>, +})); + +vi.mock('@aflow/redis', async () => { + const actual = await vi.importActual('@aflow/redis'); + return { + ...actual, + claimProjectionCandidates: () => + Promise.resolve([ + { tenantId: TENANT, runId: SESSION, version: 1, leaseUntilMs: Date.now() + 30_000 }, + ]), + getSessionStateSafe: () => Promise.resolve({ ok: true, state: hot.state }), + readDurableEventEntries: (_r: unknown, _t: string, _s: string, cursor: string) => + Promise.resolve( + cursor === '0' && hot.pending.length > 0 + ? { + entries: hot.pending.map((eventType, i) => ({ + id: `${String(i + 1)}-0`, + event: { + eventId: `e${String(i)}`, + eventType, + sessionId: SESSION, + timestamp: Date.now(), + }, + envelope: {}, + })), + lastId: `${String(hot.pending.length)}-0`, + oldestId: '1-0', + } + : { entries: [], lastId: cursor, oldestId: null }, + ), + appendEntityEvent: ( + _redis: unknown, + params: { spaceId: string; event: { eventType: string; payload: unknown } }, + ) => { + hot.signals.push({ + spaceId: params.spaceId, + eventType: params.event.eventType, + payload: params.event.payload, + }); + return Promise.resolve('1-0'); + }, + ackProjection: () => Promise.resolve(true), + publishActionCenterWake: () => undefined, + deleteRecoveryData: () => Promise.resolve(), + }; +}); +vi.mock('@aflow/database', async () => { + const actual = await vi.importActual('@aflow/database'); + const chain = (result: unknown): unknown => + new Proxy(() => undefined, { + get: (_target, key) => + key === 'then' + ? (resolve: (value: unknown) => void) => resolve(result) + : () => chain(result), + }); + const tx = { + insert: () => chain([{ sessionId: SESSION }]), + select: () => chain([{ cursor: null, status: null }]), + update: () => chain([]), + }; + return { + ...actual, + withTenantSchema: (_db: unknown, _ctx: unknown, fn: (t: unknown) => Promise) => fn(tx), + }; +}); +vi.mock('@aflow/cybernetic-runtime', async () => { + const actual = await vi.importActual( + '@aflow/cybernetic-runtime', + ); + return { ...actual, bumpAttentionGeneration: () => Promise.resolve() }; +}); +vi.mock('../hotStateSnapshot.js', () => ({ buildHotStateSnapshot: () => Promise.resolve(null) })); +vi.mock('../pauseNotificationRouter.js', () => ({ + routePauseNotifications: () => Promise.resolve(), +})); +vi.mock('../errorReportAssembly.js', () => ({ assembleErrorReport: () => Promise.resolve() })); + +const { createProjectionWorker, movesSessionStatus } = await import('../ProjectionWorker.js'); + +function session(status: string, extra: Partial = {}): SessionHotState { + const now = Date.now(); + return { + tenantId: TENANT, + sessionId: SESSION, + spaceId: SPACE, + target: { kind: 'platform-role', systemRole: 'cybernetic-helmsman' }, + agentVersion: '1', + status, + createdAt: now, + startedAt: now, + lastUpdatedAt: now, + ...extra, + } as unknown as SessionHotState; +} + +async function project(state: SessionHotState, pending: string[]): Promise { + hot.state = state; + hot.pending = pending; + const worker = createProjectionWorker({ redis: {} as never, db: {} as never }); + await worker.runOnce(); +} + +beforeEach(() => { + hot.signals = []; +}); + +describe("a conversation's status reaches the space", () => { + it('says a turn began once the resume is durable', async () => { + await project(session('RUNNING'), ['SessionResumed', 'StepScheduled']); + + expect(hot.signals).toEqual([ + { + spaceId: SPACE, + eventType: 'entity.session.status', + payload: { sessionId: SESSION, status: 'RUNNING' }, + }, + ]); + }); + + it('says a turn rested on the operator once the pause is durable', async () => { + await project(session('PAUSED'), ['StepSucceeded', 'SessionPaused']); + + expect(hot.signals).toEqual([ + { + spaceId: SPACE, + eventType: 'entity.session.status', + payload: { sessionId: SESSION, status: 'PAUSED' }, + }, + ]); + }); + + it('says nothing for a projection of steps alone, however many', async () => { + await project(session('RUNNING'), ['StepScheduled', 'StepStarted', 'StepSucceeded']); + await project(session('RUNNING'), []); + + expect(hot.signals).toEqual([]); + }); + + it("says nothing for a child session, whose turn is its parent's", async () => { + await project(session('RUNNING', { parentSessionId: PARENT }), ['SessionStarted']); + + expect(hot.signals).toEqual([]); + }); +}); + +describe('which persisted events move a status', () => { + it('is every session lifecycle event and nothing else', () => { + expect(movesSessionStatus({ eventType: 'SessionStarted' })).toBe(true); + expect(movesSessionStatus({ eventType: 'SessionCompleted' })).toBe(true); + expect(movesSessionStatus({ eventType: 'StepSucceeded' })).toBe(false); + expect(movesSessionStatus({ eventType: 'WorkflowTaskUpdate' })).toBe(false); + expect(movesSessionStatus(null)).toBe(false); + }); +}); diff --git a/docs/plans/315-local-first-run-ergonomics.md b/docs/plans/315-local-first-run-ergonomics.md index b82c422c..9b38c86d 100644 --- a/docs/plans/315-local-first-run-ergonomics.md +++ b/docs/plans/315-local-first-run-ergonomics.md @@ -244,7 +244,7 @@ Production-build heap and mount time for a long conversation with the Workbench | F72 | executor | A restart of the host executor ends every process it started and discards every session's checkout, so a merge into the stack's checkout ends every commission, check and review in flight, and `tsx watch` kills five seconds after its signal | P4e (D17: the host executor drains on SIGUSR2, which only the dev stack's watcher sends, and waits for the exit; SIGTERM still ends everything) | | F73 | executor | A checkout takes its third-party dependencies from the folder's installation, and that installation lags the lockfile whenever the operator has not run `yarn install` since pulling — the first real check on a merged tree failed on `playwright-core`, a dependency the lockfile had carried for a day and the folder had never installed | P4e (the check compares the folder's installation state with the checkout's lockfile and refuses by name when they differ, before building anything; the folder's `yarn install` stays the operator's) | | F74 | executor | A drained restart waits for the steps in flight but not for a session between its turns: on exit it discards every kept session's checkout, so a merge between the turns of a continued commission ends that session and whatever its coding agent left uncommitted | Deferred, not built (a kept session that survives a restart, its checkout left on disk and taken up again by the next executor; until then a stream continuing a session merges only once that commission has published) | -| F75 | workbench | The workbench's recent conversations do not say whether a conversation is active — Helmsman mid-turn, or a commission or publication it started still running — so the operator opens each one to find out | P2 (a conversation's row carries its live state: the agent's turn, the runs it waits on and their phase, from the session hot state and the attention list) | +| F75 | workbench | The workbench's recent conversations do not say whether a conversation is active — Helmsman mid-turn, or a commission or publication it started still running — so the operator opens each one to find out | P2 (a conversation's row carries its live state: the agent's turn, the runs it waits on and their phase, from the session hot state and the attention list) — delivered: the session list reads each status from the hot state, the projection emits `entity.session.status` when a turn begins or rests, and the row joins the Workbench's active-run feed; no request per row | | F76 | loop | A commission that runs past `host.harness.run`'s default timeout is killed mid-turn with its work on disk and no result, and the conversation that started it reported the kill as a 27-second off-script run that created another stream's branch — an account that matched nothing in the run's own activity log | P4e (a killed commission returns `killed: { afterMs, byTimeout }` beside its patch so the next brief continues it; a result the agent cannot explain is reported as the data it carries, never as a story) | | F77 | op | A commission's `maxTurns` is a number the briefing agent picks, and a brief that asked for a generous budget got 40 — a slice of this size spends that reading the code — so the run ended before its first edit and the result could only say so; a cap sized to a guess is the length-cap mistake again | P4e (the operation's default turn budget is sized for the work it exists to run, several hundred, named with its reason; a run that reaches the cap says `turnCapReached` as data beside its patch) | | F78 | readiness | A model provider refusing for credit shows in the conversation that met it, with the error and a retry that works, and nowhere else: another stream, or a driver reading run and session state, learns of it only by trying that provider again | P2 (the space's readiness carries the provider's last refusal beside the missing-key line, cleared by the next accepted call, so every conversation and every watcher sees it once) | @@ -264,9 +264,9 @@ Production-build heap and mount time for a long conversation with the Workbench | F92 | op | Helmsman's first read of a harness result guesses a field that is not there (`/data`), spending one step per commission on a `MEMORY_NOT_FOUND` that lists the real fields (320-F5) | P4e (the result's shape is in the operation's output description and the commission skill's result guidance, so the first read names a field that exists) | | F93 | platform | No check exercises an operation against the capability profile a real local space holds (Personal Safe), so a capability-migration gap shows only on the first live call (320-F17) | P4e (a contract test holds every registered operation's capability against each system profile's allowed set and fails on an operation no local profile can call) | | F94 | loop | A cancelled publication wakes the conversation that started it, and Helmsman answered the wakeup by commissioning a fix round nobody asked for, from the review findings it had — thirty minutes of coding agent on a shape the operator had not chosen | P4e (a cancellation's wakeup says who cancelled and that nothing is asked; a commission is started only from a message in the conversation) | -| F95 | workbench | A commission started as a run shows, in the workbench's run view, a coding-agent task that is `running` with `0ms` and "has not reported yet" for its whole half hour: the harness's live feed reaches the chat card of a direct call and no surface of a run, so the operator cannot tell a working commission from a stuck one | P2 (the run view's harness task shows the same live feed the chat card shows — the agent's last tool line, elapsed time, turns — from the step's activity stream) | +| F95 | workbench | A commission started as a run shows, in the workbench's run view, a coding-agent task that is `running` with `0ms` and "has not reported yet" for its whole half hour: the harness's live feed reaches the chat card of a direct call and no surface of a run, so the operator cannot tell a working commission from a stuck one | P2 (the run view's harness task shows the same live feed the chat card shows — the agent's last tool line, elapsed time, turns — from the step's activity stream) — delivered: a run's live-delta wake names its task steps, so a view opened mid-task reads their buffers; the run container folds the frames and mounts `HarnessStepCard`, the chat's card, timed from the task's start | | F96 | loop | Helmsman's link to a commission run is a `sessions/` address, which the web answers with "Failed to fetch run"; the run lives under `runs/` | P2 (the link a run result carries is built from the run's own route, by the platform, not composed by the agent) | -| F97 | workbench | The harness activity card is a scroll of tool invocations and their exit codes, unreadable in real time and misleading while F84 marks every command a failure; what an operator watching a commission needs is proof of life, what the agent says it is doing, any plain-text message it writes, and the files it is changing | P2 (the card leads with the agent's last status or thought and the time since the last activity, then the files touched, with the tool lines collapsed behind a disclosure; exit codes shown only once F84 makes them true) | +| F97 | workbench | The harness activity card is a scroll of tool invocations and their exit codes, unreadable in real time and misleading while F84 marks every command a failure; what an operator watching a commission needs is proof of life, what the agent says it is doing, any plain-text message it writes, and the files it is changing | P2 (the card leads with the agent's last status or thought and the time since the last activity, then the files touched, with the tool lines collapsed behind a disclosure; exit codes shown only once F84 makes them true) — delivered: the activity line carries `intent`, a call's `changes` and a result's `command`; a command's result shows no verdict until F84 is retired | | F98 | executor | Under `open` the sandbox opened the machine's whole loopback to every coding agent and check, and the stack's Redis listens on 6379 with no password and holds the `aflow:write-approval:*` grants the push gate reads, so code an agent wrote could mint a grant and push past the gate | P4e (D19: no port list — the machine's loopback is never a job's under either posture; the `open` launcher refuses this machine on every port, and the job's loopback is the namespace's own on Linux and none on macOS); the second half is P4e (D20): every service of the stack requires a credential, Redis the machine's password, which `REDIS_URL` is given | | F99 | skill | A loop pull request's title and body are the last publication's title and a one-paragraph summary written in the brief, so a branch of eight commits is titled by its final fix and its body says nothing a reader can act on; a pull request written by hand (#64) carries what changed as a table where one fits, the decisions as bullets, the known gaps, and what was verified and what was not | P4e (Local Publish composes the pull request from the branch as a whole: the slice's title, what changes from every commit's message, the review's findings closed and left open, the check's results as verification, and the gaps the commissions recorded; on an appended push it rewrites the body rather than leaving the first one) | | F100 | loop | A publication's `patchRef` and `baseSha` are retyped by Helmsman from a commission's result, and two publications of one patch failed at the commit on a sha with one wrong character and then on the wrong base, while the result the commission run holds was right both times | P4e (Local Publish takes the commission run's id and reads `patchRef`, `baseSha` and `merge.from` from that run's result itself; a brief names a run, not shas) | diff --git a/packages/cybernetic-runtime/src/metricsAggregator.ts b/packages/cybernetic-runtime/src/metricsAggregator.ts index 97f82c71..5c7c8e18 100644 --- a/packages/cybernetic-runtime/src/metricsAggregator.ts +++ b/packages/cybernetic-runtime/src/metricsAggregator.ts @@ -261,10 +261,13 @@ export async function aggregateMetricsBucket( // `entity.session.described` is a description OF activity, not activity: // counting it would make a space look busier the more the Clerk ran. + // `entity.session.status` likewise: the turn it reports is counted where + // the turn is. case 'entity.task.dispatched': case 'entity.task.completed': case 'entity.run.updated': case 'entity.session.described': + case 'entity.session.status': break; case 'entity.mode.transition': { diff --git a/packages/cybernetic-runtime/src/workflowRunProgress.ts b/packages/cybernetic-runtime/src/workflowRunProgress.ts index 3916cf98..acfaa42b 100644 --- a/packages/cybernetic-runtime/src/workflowRunProgress.ts +++ b/packages/cybernetic-runtime/src/workflowRunProgress.ts @@ -10,6 +10,7 @@ import { type WorkflowTaskSurfaceUpdatePayload, type WorkflowRunTaskStatus, type LiveDeltaChannel, + type LiveDeltaTaskStep, } from '@aflow/schemas'; import type { PayloadStore } from '@aflow/payload-store'; import { appendSessionEvent, publishLiveDeltaWake, type SessionEvent } from '@aflow/redis'; @@ -145,11 +146,18 @@ export async function emitWorkflowProgress( * delta is a value superseded by its step's terminal event, and appending one * per flush would spend the session event stream's bounded capacity on text * that is never replayed. A step a workflow dispatched has no session of its - * own, so this is the only way its buffer reaches a reader. + * own, so this is the only way its buffer reaches a reader — and the wake names + * the steps it is about, so a reader that joined after the task started still + * finds them. */ export async function publishWorkflowLiveDeltaWake( deps: EmitWorkflowProgressDeps, - args: { tenantId: string; runId: string; channel: LiveDeltaChannel }, + args: { + tenantId: string; + runId: string; + channel: LiveDeltaChannel; + taskSteps: readonly LiveDeltaTaskStep[]; + }, ): Promise { const [waiters, originatingSessionId] = await Promise.all([ loadPendingWaiters(deps.db, args.tenantId, args.runId), @@ -157,7 +165,7 @@ export async function publishWorkflowLiveDeltaWake( ]); const targets = new Set(waiters.map((w) => w.waiterSessionId)); if (originatingSessionId) targets.add(originatingSessionId); - await publishLiveDeltaWake(deps.redis, args.tenantId, [...targets], args.channel); + await publishLiveDeltaWake(deps.redis, args.tenantId, [...targets], args.channel, args.taskSteps); } /** diff --git a/packages/cybernetic-runtime/src/workflowTaskProgressConsumer.ts b/packages/cybernetic-runtime/src/workflowTaskProgressConsumer.ts index c78da61e..9ae95e19 100644 --- a/packages/cybernetic-runtime/src/workflowTaskProgressConsumer.ts +++ b/packages/cybernetic-runtime/src/workflowTaskProgressConsumer.ts @@ -6,6 +6,7 @@ import { StreamKeys, WORKFLOW_TASK_LIVE_DELTA_EVENT_TYPE, type LiveDeltaChannel, + type LiveDeltaTaskStep, type WorkflowTaskSurfaceUpdatePayload, type WorkflowTaskActivityPayload, type StepExecutionId, @@ -230,8 +231,18 @@ async function runOneIteration(args: { let totalEmitted = 0; // Coalesced per run and channel. A wake is a signal and the reader answers it // by reading the whole buffer, so a burst of appended lines needs exactly one - // — and the fan-out it skips is two indexed queries apiece. - const wakes = new Map(); + // — and the fan-out it skips is two indexed queries apiece. The steps it + // covers are kept, every one: a run's tasks can stream side by side, and the + // wake is how a watcher that joined late learns which buffers to read. + const wakes = new Map< + string, + { + tenantId: string; + runId: string; + channel: LiveDeltaChannel; + taskSteps: Map; + } + >(); for (const [streamKey, entries] of result as Array<[string, Array<[string, string[]]>]>) { for (const [eventId, fieldPairs] of entries) { const fields = parseFieldPairs(fieldPairs); @@ -239,11 +250,19 @@ async function runOneIteration(args: { const wake = decodeLiveDeltaWake(fields); if (wake) { if (tenantId) { - wakes.set(`${tenantId}\u0000${wake.runId}\u0000${wake.channel}`, { + const key = `${tenantId}\u0000${wake.runId}\u0000${wake.channel}`; + const held = wakes.get(key) ?? { tenantId, runId: wake.runId, channel: wake.channel, + taskSteps: new Map(), + }; + held.taskSteps.set(wake.stepExecutionId, { + runId: wake.runId, + taskId: wake.taskId, + stepExecutionId: wake.stepExecutionId, }); + wakes.set(key, held); totalEmitted += 1; } cursors.set(streamKey, eventId); @@ -279,7 +298,12 @@ async function runOneIteration(args: { } } for (const wake of wakes.values()) { - await publishWorkflowLiveDeltaWake(deps, wake).catch((err: unknown) => { + await publishWorkflowLiveDeltaWake(deps, { + tenantId: wake.tenantId, + runId: wake.runId, + channel: wake.channel, + taskSteps: [...wake.taskSteps.values()], + }).catch((err: unknown) => { console.warn( `[workflowTaskProgressConsumer] live-delta wake failed runId=${wake.runId}: ${err instanceof Error ? err.message : String(err)}`, ); diff --git a/packages/redis/src/hotState/events.ts b/packages/redis/src/hotState/events.ts index d80cdd57..c2967e15 100644 --- a/packages/redis/src/hotState/events.ts +++ b/packages/redis/src/hotState/events.ts @@ -1,5 +1,5 @@ import type { Redis } from 'ioredis'; -import { StreamKeys } from '@aflow/schemas'; +import { StreamKeys, type LiveDeltaTaskStep } from '@aflow/schemas'; import { compareStreamIds } from '../streams/streamId.js'; import type { SessionEvent } from './schemas.js'; import { HOT_STATE_TTL_SECONDS, SessionEventSchema } from './schemas.js'; @@ -10,9 +10,23 @@ import { sessionCandidateMember } from './candidateMember.js'; // Pub/Sub Notification (best-effort wakeup for SSE) // ============================================================================ -/** The wake payload, shared by the atomic append path and the standalone one. */ -export function sessionWakePayload(runId: string, eventType: string): string { - return JSON.stringify({ type: 'event', runId, eventType, ts: Date.now() }); +/** + * The wake payload, shared by the atomic append path and the standalone one. + * `taskSteps` names the run task steps a live-delta wake is about; see + * `LiveDeltaTaskStepSchema`. + */ +export function sessionWakePayload( + runId: string, + eventType: string, + taskSteps: readonly LiveDeltaTaskStep[] = [], +): string { + return JSON.stringify({ + type: 'event', + runId, + eventType, + ts: Date.now(), + ...(taskSteps.length > 0 ? { taskSteps } : {}), + }); } // ============================================================================ diff --git a/packages/redis/src/hotState/liveStream.ts b/packages/redis/src/hotState/liveStream.ts index 860daed8..8b239cbb 100644 --- a/packages/redis/src/hotState/liveStream.ts +++ b/packages/redis/src/hotState/liveStream.ts @@ -1,5 +1,10 @@ import type { Redis } from 'ioredis'; -import { LiveDeltaChannelSchema, StreamKeys, type LiveDeltaChannel } from '@aflow/schemas'; +import { + LiveDeltaChannelSchema, + StreamKeys, + type LiveDeltaChannel, + type LiveDeltaTaskStep, +} from '@aflow/schemas'; import { HOT_STATE_TTL_SECONDS } from './schemas.js'; import { sessionWakePayload } from './events.js'; @@ -88,22 +93,25 @@ export async function appendLiveDelta( /** * Wake the watchers of a step whose own job named no session. * - * Same payload on the same channel as the append's own wake, so a reader cannot - * tell the two apart — which is the point: the live plane keeps one shape, and - * only who resolves the audience differs. A signal, never content. + * Same payload on the same channel as the append's own wake, plus the steps it + * is about: a session's own step is named by its hot state, but a run's task + * step is named only by the run, and a reader that subscribed after the task + * started has no other way to learn it. A signal, never content — identifiers + * of buffers whose audience this caller already resolved. */ export async function publishLiveDeltaWake( redis: Redis, tenantId: string, sessionIds: readonly string[], channel: LiveStreamChannel, + taskSteps: readonly LiveDeltaTaskStep[] = [], ): Promise { if (sessionIds.length === 0) return; const pipeline = redis.pipeline(); for (const sessionId of sessionIds) { pipeline.publish( StreamKeys.pubsubChannel(tenantId, sessionId), - sessionWakePayload(sessionId, `LiveDelta:${channel}`), + sessionWakePayload(sessionId, `LiveDelta:${channel}`, taskSteps), ); } await pipeline.exec(); diff --git a/packages/schemas/src/__tests__/cybernetic.test.ts b/packages/schemas/src/__tests__/cybernetic.test.ts index cecc85b0..c1bb1c2d 100644 --- a/packages/schemas/src/__tests__/cybernetic.test.ts +++ b/packages/schemas/src/__tests__/cybernetic.test.ts @@ -215,9 +215,10 @@ describe('EntityEventEnvelopeSchema', () => { expect(result.traceId).toBe('abc123def456'); }); - it('accepts all 50 event types (includes 102h + 104b + 104c + 104e + 104g + 111 + 163 + 170 + 183a/183b follow-up events, and the conversation descriptor)', () => { + it('accepts all 51 event types (includes 102h + 104b + 104c + 104e + 104g + 111 + 163 + 170 + 183a/183b follow-up events, the conversation descriptor and its status)', () => { const types = EntityEventTypeSchema.options; - expect(types.length).toBe(50); + expect(types.length).toBe(51); + expect(types).toContain('entity.session.status'); expect(types).toContain('entity.schedule.armed'); expect(types).toContain('entity.coach.withdrawn'); expect(types).toContain('entity.eval.completed'); diff --git a/packages/schemas/src/runtime/entityEvents.ts b/packages/schemas/src/runtime/entityEvents.ts index 4a2e636e..3ebfe378 100644 --- a/packages/schemas/src/runtime/entityEvents.ts +++ b/packages/schemas/src/runtime/entityEvents.ts @@ -97,6 +97,13 @@ export const EntityEventTypeSchema = z.enum([ // conversation list refreshes without every row holding a subscription of // its own. Payload carries { sessionId }; it is never an unread message. 'entity.session.described', + + // A conversation's session moved between statuses — a turn began, rested on + // the operator, or ended — and the move is now durable. Space-level for the + // same reason as `described`: the conversation list says which conversations + // are live without a subscription per row. Payload carries { sessionId, + // status }; it is a description of activity, not activity. + 'entity.session.status', ]); export type EntityEventType = z.infer; diff --git a/packages/schemas/src/runtime/harnessActivity.test.ts b/packages/schemas/src/runtime/harnessActivity.test.ts index c30f5a2a..fd8fde00 100644 --- a/packages/schemas/src/runtime/harnessActivity.test.ts +++ b/packages/schemas/src/runtime/harnessActivity.test.ts @@ -37,6 +37,31 @@ describe('an activity line', () => { expect(result).toMatchObject({ kind: 'tool_result', ok: false }); }); + it('carries what the agent says it is doing, the file a call changes, and a command', () => { + expect( + HarnessActivityLineSchema.parse({ kind: 'intent', at: 5, text: 'Running the tests' }), + ).toEqual({ kind: 'intent', at: 5, text: 'Running the tests' }); + expect( + HarnessActivityLineSchema.parse({ + kind: 'tool', + at: 6, + tool: 'Edit', + text: 'Edit src/parser.ts', + changes: 'src/parser.ts', + }), + ).toMatchObject({ changes: 'src/parser.ts' }); + expect( + HarnessActivityLineSchema.parse({ + kind: 'tool_result', + at: 7, + tool: 'Bash', + ok: false, + command: true, + text: 'exit 1', + }), + ).toMatchObject({ command: true, ok: false }); + }); + it('refuses a tool line with no tool, and a result with no verdict', () => { expect( HarnessActivityLineSchema.safeParse({ kind: 'tool', at: 1, text: 'something happened' }) diff --git a/packages/schemas/src/runtime/harnessActivity.ts b/packages/schemas/src/runtime/harnessActivity.ts index e13dafb9..b1511ea3 100644 --- a/packages/schemas/src/runtime/harnessActivity.ts +++ b/packages/schemas/src/runtime/harnessActivity.ts @@ -2,8 +2,9 @@ * What a coding harness is doing while a step runs, one line at a time. * * A harness works for minutes at a stretch and says nothing until it is done. - * The activity feed is that wait made legible: the tools it reaches for, the - * narration between them, how each call came back. It is a live value, not a + * The activity feed is that wait made legible: what it says it is doing, the + * tools it reaches for, the narration between them, the files it changes, how + * each call came back. It is a live value, not a * record — superseded by the step's own result, never replayed from a cursor — * so a line is shaped for reading rather than for storage. * @@ -50,6 +51,17 @@ export const HarnessActivityLineSchema = z.discriminatedUnion('kind', [ .max(ACTIVITY_TEXT_MAX) .describe("The harness's own narration between tool calls, in its words."), }), + z.object({ + kind: z.literal('intent'), + at, + text: z + .string() + .max(ACTIVITY_TEXT_MAX) + .describe( + 'What the agent says it is doing now — the step of its own plan it has marked in ' + + 'progress, in its words. The line a person watching the run reads first.', + ), + }), z.object({ kind: z.literal('tool'), at, @@ -64,6 +76,15 @@ export const HarnessActivityLineSchema = z.discriminatedUnion('kind', [ 'One line naming the tool and the most telling thing it was given — the file, the ' + 'command, the pattern — so a reader sees what is being done, not that something is.', ), + changes: z + .string() + .max(ACTIVITY_TEXT_MAX) + .optional() + .describe( + 'The file this call writes or edits, when it changes one. Read off the call by the ' + + 'executor that knows the harness, so a surface lists the files being changed without ' + + 'knowing any tool by name.', + ), }), z.object({ kind: z.literal('tool_result'), @@ -75,6 +96,14 @@ export const HarnessActivityLineSchema = z.discriminatedUnion('kind', [ 'False when the tool itself reported an error. A failed tool call is ordinary — the ' + 'harness reads it and carries on — so it is a property of the line, never of the step.', ), + command: z + .boolean() + .optional() + .describe( + "True when the call ran a shell command, so `ok` is that command's exit status. Kept " + + 'apart from other tools because an exit status is only as truthful as the sandbox the ' + + 'command ran in, and a surface decides whether to show it on that basis.', + ), text: z .string() .max(ACTIVITY_TEXT_MAX) diff --git a/packages/schemas/src/runtime/streamMessages.ts b/packages/schemas/src/runtime/streamMessages.ts index d5a9c2fe..98d8f075 100644 --- a/packages/schemas/src/runtime/streamMessages.ts +++ b/packages/schemas/src/runtime/streamMessages.ts @@ -40,6 +40,24 @@ import { SimulatedFulfillmentReportSchema } from '../simulation/fulfillmentRepor export const LiveDeltaChannelSchema = z.enum(['text', 'thinking', 'activity']); export type LiveDeltaChannel = z.infer; +/** + * The step a workflow run's wake is about, carried on the wake itself. + * + * A step a run dispatched belongs to no session, so a watcher learns which + * buffer to read from the run's `WorkflowTaskUpdate`. A watcher that subscribed + * after that update — a run view opened mid-task, which tails from the + * session's latest position — never sees it, and would read nothing for the + * whole of the task. Naming the step on the wake lets it start from the next + * flush. Identifiers only: the audience is resolved server-side from the run's + * watchers, and the bytes stay in the buffer. + */ +export const LiveDeltaTaskStepSchema = z.object({ + runId: z.string().min(1), + taskId: z.string().min(1), + stepExecutionId: z.string().min(1), +}); +export type LiveDeltaTaskStep = z.infer; + /** * Redis stream key patterns. */ diff --git a/packages/server-runtime/src/bootstrap/hostLaneFeed.test.ts b/packages/server-runtime/src/bootstrap/hostLaneFeed.test.ts index 14ad90dd..a86aea7c 100644 --- a/packages/server-runtime/src/bootstrap/hostLaneFeed.test.ts +++ b/packages/server-runtime/src/bootstrap/hostLaneFeed.test.ts @@ -415,7 +415,12 @@ describe.skipIf(!AVAILABLE)('a workflow-dispatched host step feeding a watching try { await publishWorkflowLiveDeltaWake( { db: watchedBy(SESSION_ID), redis: requireAdmin() }, - { tenantId: TENANT_ID, runId: RUN_ID, channel: 'activity' }, + { + tenantId: TENANT_ID, + runId: RUN_ID, + channel: 'activity', + taskSteps: [{ runId: RUN_ID, taskId: TASK_ID, stepExecutionId: STEP_EXECUTION_ID }], + }, ); await until( 'the wake to arrive on the session channel', @@ -430,13 +435,25 @@ describe.skipIf(!AVAILABLE)('a workflow-dispatched host step feeding a watching // `LiveDelta:` prefix is what keeps a 150ms flush from driving the durable // drain. expect(received).toHaveLength(1); + // And the step it is about, which is all a run view opened after the task + // started has to go on. expect(JSON.parse(received[0] ?? '{}')).toMatchObject({ type: 'event', runId: SESSION_ID, eventType: 'LiveDelta:activity', + taskSteps: [{ runId: RUN_ID, taskId: TASK_ID, stepExecutionId: STEP_EXECUTION_ID }], }); }, 30_000); + it('reads the buffer for a run task step it learned from the wake alone', async () => { + const reader = createLiveDeltaReader(requireAdmin(), TENANT_ID, SESSION_ID); + reader.track([{ runId: RUN_ID, taskId: TASK_ID, stepExecutionId: STEP_EXECUTION_ID }]); + + const frames = await reader.read(new AbortController().signal); + expect(frames).toHaveLength(1); + expect(frames[0]?.delta).toBe(await bufferedFeed()); + }, 30_000); + it('reads the buffer for a run task step it learned from a task update', async () => { const reader = createLiveDeltaReader(requireAdmin(), TENANT_ID, SESSION_ID); reader.observe(taskUpdateEvent()); diff --git a/packages/server-runtime/src/services/__tests__/sessionListHotStatus.test.ts b/packages/server-runtime/src/services/__tests__/sessionListHotStatus.test.ts new file mode 100644 index 00000000..e56cbbf9 --- /dev/null +++ b/packages/server-runtime/src/services/__tests__/sessionListHotStatus.test.ts @@ -0,0 +1,73 @@ +/** + * A conversation list reads each session's status from its hot state where the + * hash still exists, so a list that says whether a conversation is mid-turn is + * not a projection cycle late to it (Plan 315 F75). One round trip for the + * page, one field per session, and nothing it cannot read changes what the + * projected row says. + */ +import { describe, expect, it } from 'vitest'; +import { StreamKeys } from '@aflow/schemas'; + +import { readHotSessionStatuses } from '../sessions.js'; + +const TENANT = 'tenant-1'; + +function fakeRedis(fields: Record) { + const reads: string[] = []; + let execs = 0; + return { + reads, + execs: () => execs, + redis: { + pipeline() { + const results: Array<[Error | null, unknown]> = []; + const chain = { + hget(key: string, field: string) { + reads.push(`${key}#${field}`); + results.push([null, fields[key] ?? null]); + return chain; + }, + exec() { + execs += 1; + return Promise.resolve(results); + }, + }; + return chain; + }, + }, + }; +} + +describe('the hot status of listed sessions', () => { + it('reads every listed session in one round trip', async () => { + const { redis, reads, execs } = fakeRedis({ + [StreamKeys.sessionStateKey(TENANT, 'a')]: 'RUNNING', + [StreamKeys.sessionStateKey(TENANT, 'b')]: 'PAUSED', + }); + + const statuses = await readHotSessionStatuses(redis as never, TENANT, ['a', 'b', 'c']); + + expect(execs()).toBe(1); + expect(reads).toEqual([ + `${StreamKeys.sessionStateKey(TENANT, 'a')}#status`, + `${StreamKeys.sessionStateKey(TENANT, 'b')}#status`, + `${StreamKeys.sessionStateKey(TENANT, 'c')}#status`, + ]); + // `c` has no hash any more: its projected row is the truth for it. + expect([...statuses]).toEqual([ + ['a', 'RUNNING'], + ['b', 'PAUSED'], + ]); + }); + + it('leaves out a value that is not a status', async () => { + const { redis } = fakeRedis({ [StreamKeys.sessionStateKey(TENANT, 'a')]: 'HALF_DONE' }); + expect((await readHotSessionStatuses(redis as never, TENANT, ['a'])).size).toBe(0); + }); + + it('costs nothing for an empty page', async () => { + const { redis, execs } = fakeRedis({}); + expect((await readHotSessionStatuses(redis as never, TENANT, [])).size).toBe(0); + expect(execs()).toBe(0); + }); +}); diff --git a/packages/server-runtime/src/services/__tests__/sessionLiveDeltas.test.ts b/packages/server-runtime/src/services/__tests__/sessionLiveDeltas.test.ts index a3bd47dd..685fcd7d 100644 --- a/packages/server-runtime/src/services/__tests__/sessionLiveDeltas.test.ts +++ b/packages/server-runtime/src/services/__tests__/sessionLiveDeltas.test.ts @@ -290,6 +290,47 @@ describe("the reader follows a run's task steps too", () => { expect(await r.read(signal)).toEqual([]); }); + it('follows a step the wake names, for a reader that joined after the task started', async () => { + // A run view opened mid-task tails from the session's latest position, so + // the `running` update is behind it. The wake is the only thing it hears. + const { redis, r } = watching(); + redis.append(WORKER_STEP, 'activity', 'streamed before the view opened\n'); + r.track([{ runId: RUN, taskId: 'review', stepExecutionId: WORKER_STEP }]); + + expect(await r.read(signal)).toEqual([ + { + stepExecutionId: WORKER_STEP, + channel: 'activity', + offset: 0, + delta: 'streamed before the view opened\n', + }, + ]); + + redis.append(WORKER_STEP, 'activity', 'and since\n'); + r.track([{ runId: RUN, taskId: 'review', stepExecutionId: WORKER_STEP }]); + expect(await r.read(signal)).toEqual([ + { + stepExecutionId: WORKER_STEP, + channel: 'activity', + offset: 'streamed before the view opened\n'.length, + delta: 'and since\n', + }, + ]); + }); + + it('ignores a late wake for a step whose task it saw end', async () => { + const { redis, r } = watching(); + r.track([{ runId: RUN, taskId: 'review', stepExecutionId: WORKER_STEP }]); + redis.append(WORKER_STEP, 'activity', 'line\n'); + expect(await r.read(signal)).toHaveLength(1); + + r.observe( + taskEvent({ status: 'succeeded', taskType: 'operation', workerSessionId: WORKER_STEP }), + ); + r.track([{ runId: RUN, taskId: 'review', stepExecutionId: WORKER_STEP }]); + expect(await r.read(signal)).toEqual([]); + }); + it('leaves an agent task alone — its worker session is a session, not a step', async () => { const { redis, r } = watching(); r.observe(taskEvent({ status: 'running', taskType: 'agent', workerSessionId: WORKER_STEP })); diff --git a/packages/server-runtime/src/services/__tests__/sessionWakeup.test.ts b/packages/server-runtime/src/services/__tests__/sessionWakeup.test.ts index c90edb9b..02575c12 100644 --- a/packages/server-runtime/src/services/__tests__/sessionWakeup.test.ts +++ b/packages/server-runtime/src/services/__tests__/sessionWakeup.test.ts @@ -93,6 +93,40 @@ describe('session wakeup', () => { controller.abort(); }); + it('hands over the task steps a run wake names, gathered across coalesced wakes', async () => { + const { subscriber, publish } = fakeSubscriber(); + const controller = new AbortController(); + const wakeup = await subscribeSessionWakeup({ + pubsubSubscriber: subscriber, + tenantId: TENANT, + sessionId: SESSION, + signal: controller.signal, + }); + + const step = (id: string) => ({ runId: 'run-1', taskId: `task-${id}`, stepExecutionId: id }); + publish({ + type: 'event', + runId: SESSION, + eventType: 'LiveDelta:activity', + taskSteps: [step('a')], + } as PubSubNotification); + publish({ + type: 'event', + runId: SESSION, + eventType: 'LiveDelta:activity', + // A malformed entry is dropped; the well-formed ones still arrive. + taskSteps: [step('a'), step('b'), { runId: 'run-1' }], + } as PubSubNotification); + + expect(await wakeup.next()).toEqual({ durable: false, taskSteps: [step('a'), step('b')] }); + + publish({ type: 'event', runId: SESSION, eventType: 'LiveDelta:text' } as PubSubNotification); + expect(await wakeup.next()).toEqual({ durable: false }); + + await wakeup.close(); + controller.abort(); + }); + it('treats a durable event notification as a durable wake', async () => { const { subscriber, publish } = fakeSubscriber(); const controller = new AbortController(); diff --git a/packages/server-runtime/src/services/pubsub.ts b/packages/server-runtime/src/services/pubsub.ts index 3ddc215d..5d3d3598 100644 --- a/packages/server-runtime/src/services/pubsub.ts +++ b/packages/server-runtime/src/services/pubsub.ts @@ -21,6 +21,11 @@ export interface RunEventNotification { eventId: string; eventType: string; timestamp: string; + /** + * The run task steps a live-delta wake is about. Parsed as JSON and never + * typed by that alone: a reader validates it against `LiveDeltaTaskStepSchema`. + */ + taskSteps?: unknown; } export interface RunStatusNotification { diff --git a/packages/server-runtime/src/services/sessionLiveDeltas.ts b/packages/server-runtime/src/services/sessionLiveDeltas.ts index e12c4dad..4186f3c4 100644 --- a/packages/server-runtime/src/services/sessionLiveDeltas.ts +++ b/packages/server-runtime/src/services/sessionLiveDeltas.ts @@ -5,6 +5,7 @@ import { type ApiSessionEvent, type LiveDeltaChannel, type LiveDeltaFrame, + type LiveDeltaTaskStep, type SessionId, type TenantId, } from '@aflow/schemas'; @@ -30,7 +31,9 @@ import { readLiveDeltaFrom } from '@aflow/redis'; * run dispatched to an operation belongs to no session at all, so the session * parked on the run is the only reader its buffer has; the run records that * step in the task row's `workerSessionId`, which arrives here on the same - * `WorkflowTaskUpdate` events the surface is built from. + * `WorkflowTaskUpdate` events the surface is built from — or, for a reader + * that subscribed after that update went out, on the run's live-delta wake, + * which names the steps it is about (`track`). */ export type { LiveDeltaFrame }; @@ -70,6 +73,12 @@ function offsetKey(stepExecutionId: string, channel: LiveDeltaChannel): string { export interface LiveDeltaReader { /** Feed every drained durable event so the reader tracks step lifecycle. */ observe(event: ApiSessionEvent): void; + /** + * Follow the run task steps a live-delta wake named. A step whose task this + * reader already saw end stays ended: a wake can arrive after the terminal + * update, and reading its buffer again would re-send a finished feed. + */ + track(taskSteps: readonly LiveDeltaTaskStep[]): void; /** Read whatever the in-flight step has buffered since the last call. */ read(signal: AbortSignal): Promise; } @@ -91,6 +100,22 @@ export function createLiveDeltaReader( for (const channel of LIVE_CHANNELS) offsets.delete(offsetKey(stepExecutionId, channel)); }; + const markTerminated = (stepExecutionId: string): void => { + terminated.add(stepExecutionId); + if (terminated.size > TERMINATED_WINDOW) { + const oldest = terminated.values().next().value; + if (oldest !== undefined) terminated.delete(oldest); + } + }; + + const follow = (key: string, step: string): void => { + const held = taskSteps.get(key); + if (held !== undefined && held !== step) forget(held); + if (taskSteps.has(key) || taskSteps.size < MAX_TRACKED_TASK_STEPS) { + taskSteps.set(key, step); + } + }; + return { observe(event: ApiSessionEvent): void { const taskUpdate = event.data.workflowTaskUpdate; @@ -105,14 +130,16 @@ export function createLiveDeltaReader( taskUpdate.taskType !== 'agent' && taskUpdate.taskType !== 'human' && !TERMINAL_TASK_STATUSES.has(taskUpdate.status); - if (held !== undefined && held !== step) forget(held); if (streams && step !== undefined) { - if (taskSteps.has(key) || taskSteps.size < MAX_TRACKED_TASK_STEPS) { - taskSteps.set(key, step); - } + terminated.delete(step); + follow(key, step); } else { taskSteps.delete(key); if (held !== undefined) forget(held); + if (TERMINAL_TASK_STATUSES.has(taskUpdate.status)) { + if (step !== undefined) markTerminated(step); + if (held !== undefined) markTerminated(held); + } } return; } @@ -124,10 +151,13 @@ export function createLiveDeltaReader( // in `readLiveDeltaFrom` handles the buffer's DEL-and-reappend. Marking it // terminated here would drop the retried attempt's stream entirely. if (event.eventType === 'StepFailed' && event.metadata?.['willRetry'] === true) return; - terminated.add(stepExecutionId); - if (terminated.size > TERMINATED_WINDOW) { - const oldest = terminated.values().next().value; - if (oldest !== undefined) terminated.delete(oldest); + markTerminated(stepExecutionId); + }, + + track(named: readonly LiveDeltaTaskStep[]): void { + for (const { runId, taskId, stepExecutionId } of named) { + if (terminated.has(stepExecutionId)) continue; + follow(`${runId}\u0000${taskId}`, stepExecutionId); } }, diff --git a/packages/server-runtime/src/services/sessionTail.ts b/packages/server-runtime/src/services/sessionTail.ts index f0dfaaef..3dab51aa 100644 --- a/packages/server-runtime/src/services/sessionTail.ts +++ b/packages/server-runtime/src/services/sessionTail.ts @@ -764,6 +764,9 @@ export function createSessionTailService(deps: SessionTailDeps): SessionTailServ const signalKind = await wakeupSource.next(); if (signalKind === 'abort' || signal.aborted) return; if (signalKind.durable) yield* drainDurable(); + // After the drain, so a terminal update in the same burst is observed + // before a wake that still names its step. + if (signalKind.taskSteps) liveReader?.track(signalKind.taskSteps); yield* readLive(); } } finally { diff --git a/packages/server-runtime/src/services/sessionWakeup.ts b/packages/server-runtime/src/services/sessionWakeup.ts index e271499b..34db78d1 100644 --- a/packages/server-runtime/src/services/sessionWakeup.ts +++ b/packages/server-runtime/src/services/sessionWakeup.ts @@ -1,4 +1,9 @@ -import type { SessionId, TenantId } from '@aflow/schemas'; +import { + LiveDeltaTaskStepSchema, + type LiveDeltaTaskStep, + type SessionId, + type TenantId, +} from '@aflow/schemas'; import type { PubSubSubscriber, PubSubNotification } from './pubsub.js'; /** @@ -18,8 +23,24 @@ import type { PubSubSubscriber, PubSubNotification } from './pubsub.js'; * the subscriber wakes its sessions on reconnect — so a timer that re-read the * stream twice a second per subscription would be paying, forever and per * connection, for a gap that no longer exists. + * + * A live-delta wake from a workflow run also names the task steps it is about + * (`taskSteps`), accumulated across coalesced wakes like `durable` is, because + * a reader that subscribed after a task started has no other way to learn its + * step. Absent when no wake named any. */ -export type WakeupSignal = 'abort' | { durable: boolean }; +export type WakeupSignal = 'abort' | { durable: boolean; taskSteps?: LiveDeltaTaskStep[] }; + +/** The steps a notification names, keeping each one the schema accepts. */ +export function taskStepsOf(notification: PubSubNotification): LiveDeltaTaskStep[] { + if (notification.type !== 'event' || !Array.isArray(notification.taskSteps)) return []; + const steps: LiveDeltaTaskStep[] = []; + for (const entry of notification.taskSteps as unknown[]) { + const parsed = LiveDeltaTaskStepSchema.safeParse(entry); + if (parsed.success) steps.push(parsed.data); + } + return steps; +} export interface SessionWakeup { /** Resolves on the next wakeup, or `'abort'` once the caller's signal fires. */ @@ -55,10 +76,13 @@ export async function subscribeSessionWakeup(opts: { // durable event coalesced behind a burst of live flushes is never missed. let wakeupPending = false; let durablePending = false; + // Keyed by step, so a burst of flushes from one task names it once. + const taskStepsPending = new Map(); let resolveWakeup: (() => void) | null = null; - const wakeup = (durable: boolean) => { + const wakeup = (durable: boolean, taskSteps: readonly LiveDeltaTaskStep[] = []) => { wakeupPending = true; if (durable) durablePending = true; + for (const step of taskSteps) taskStepsPending.set(step.stepExecutionId, step); const r = resolveWakeup; resolveWakeup = null; if (r) r(); @@ -69,7 +93,13 @@ export async function subscribeSessionWakeup(opts: { wakeupPending = false; const durable = durablePending; durablePending = false; - resolve({ durable }); + if (taskStepsPending.size === 0) { + resolve({ durable }); + return; + } + const taskSteps = [...taskStepsPending.values()]; + taskStepsPending.clear(); + resolve({ durable, taskSteps }); }; if (wakeupPending) { drain(); @@ -103,7 +133,7 @@ export async function subscribeSessionWakeup(opts: { if (pubsubSubscriber) { try { subscription = await pubsubSubscriber.subscribeToRun(tenantId, sessionId, (notification) => { - wakeup(isDurableNotification(notification)); + wakeup(isDurableNotification(notification), taskStepsOf(notification)); }); } catch (err) { unsubscribeReconnect?.(); diff --git a/packages/server-runtime/src/services/sessions.ts b/packages/server-runtime/src/services/sessions.ts index 1c08702e..d352cf66 100644 --- a/packages/server-runtime/src/services/sessions.ts +++ b/packages/server-runtime/src/services/sessions.ts @@ -14,8 +14,10 @@ import { AgentDefinitionSchema, MAX_INLINE_DEFINITION_BYTES, projectTargetColumns, + SessionStatusSchema, StreamKeys, } from '@aflow/schemas'; +import type { Redis } from 'ioredis'; import type { TenantId, SessionId, @@ -811,6 +813,38 @@ function createMockSessionService(): SessionService { // Helpers // ============================================================================ +/** + * Each listed session's status as its hot state holds it, in one round trip. + * + * One field per session, read raw: a listing must never be the reader that + * quarantines a live session's hash over a parse it does not need. A session + * whose hash has gone — rested past its TTL, or flushed — is absent, and its + * projected row is the truth for it. + */ +export async function readHotSessionStatuses( + redis: Pick, + tenantId: string, + sessionIds: readonly string[], +): Promise> { + const statuses = new Map(); + if (sessionIds.length === 0) return statuses; + const pipeline = redis.pipeline(); + for (const sessionId of sessionIds) { + pipeline.hget(StreamKeys.sessionStateKey(tenantId, sessionId), 'status'); + } + let results: Array<[Error | null, unknown]> | null; + try { + results = await pipeline.exec(); + } catch { + return statuses; + } + sessionIds.forEach((sessionId, i) => { + const parsed = SessionStatusSchema.safeParse(results?.[i]?.[1]); + if (parsed.success) statuses.set(sessionId, parsed.data); + }); + return statuses; +} + async function hydrateRequestedInput( store: PayloadStore, ref: string, @@ -1519,8 +1553,17 @@ function createRealSessionService(ctx: AppContext): SessionService { runs = filtered; } + const listed = runs.slice(0, query.limit); + const hotStatuses = redis + ? await readHotSessionStatuses( + redis, + query.tenantId, + listed.map((run) => run.sessionId), + ) + : new Map(); + return { - sessions: runs.slice(0, query.limit).map((run) => ({ + sessions: listed.map((run) => ({ sessionId: run.sessionId as SessionId, target: projectTargetColumns({ targetKind: run.targetKind, @@ -1529,7 +1572,10 @@ function createRealSessionService(ctx: AppContext): SessionService { targetInlineDefRef: run.targetInlineDefRef, }), agentVersion: run.agentVersion, - status: run.status as SessionStatus, + // The hot state where it is still held: the row is its projection, a + // cycle behind, and a list that says whether a conversation is mid-turn + // should not be the one surface that is late to it. + status: hotStatuses.get(run.sessionId) ?? (run.status as SessionStatus), createdAt: run.startedAt.toISOString(), // The execution clock, which the projection advances on every step. // What a conversation list sorts on is `lastActivityAt`. diff --git a/packages/web-product/src/ui/components/harness-activity-card.css b/packages/web-product/src/ui/components/harness-activity-card.css index 33cb8b86..204447bf 100644 --- a/packages/web-product/src/ui/components/harness-activity-card.css +++ b/packages/web-product/src/ui/components/harness-activity-card.css @@ -39,6 +39,75 @@ padding-top: var(--space-3); } +/* What the agent last said leads the body, in prose and at reading size: it is + the one line written for the person watching. Held to a few lines so a long + paragraph cannot push the files and the disclosure out of the card. */ +.harness-activity__lead { + display: flex; + flex-direction: column; + gap: var(--space-1); + margin-bottom: var(--space-2); +} + +.harness-activity__lead-text { + font-size: var(--font-size-sm); + color: var(--color-text-primary); + white-space: pre-wrap; + overflow-wrap: anywhere; + display: -webkit-box; + -webkit-line-clamp: 4; + -webkit-box-orient: vertical; + overflow: hidden; +} + +.harness-activity__lead-text--status { + color: var(--color-text-muted); +} + +.harness-activity__files { + margin-bottom: var(--space-2); +} + +.harness-activity__file-list { + list-style: none; + margin: var(--space-1) 0 0; + padding: 0; + font-family: var(--font-family-mono); + font-size: var(--font-size-xs); + line-height: 1.6; +} + +.harness-activity__file { + white-space: nowrap; + overflow: hidden; + text-overflow: ellipsis; +} + +.harness-activity__file-more { + color: var(--color-text-muted); +} + +.harness-activity__tools { + all: unset; + box-sizing: border-box; + display: flex; + align-items: center; + gap: var(--space-1); + width: 100%; + margin-bottom: var(--space-1); + color: var(--color-text-muted); + font-size: var(--font-size-xs); + cursor: pointer; +} + +.harness-activity__tools-label { + flex: 1 1 auto; + min-width: 0; + white-space: nowrap; + overflow: hidden; + text-overflow: ellipsis; +} + .harness-feed__wrap { position: relative; } diff --git a/packages/web-product/src/ui/components/harness-activity-card.render.test.ts b/packages/web-product/src/ui/components/harness-activity-card.render.test.ts new file mode 100644 index 00000000..4b8e1b5e --- /dev/null +++ b/packages/web-product/src/ui/components/harness-activity-card.render.test.ts @@ -0,0 +1,169 @@ +/** + * @vitest-environment jsdom + */ +/** + * The harness card as a person sees it, rendered: where it is mounted, what it + * leads with, and what it keeps folded. + * + * The run surface's harness task mounts the same card the chat does, fed from + * the step's activity stream through the surface's provider — a commission's + * task row reading "has not reported yet" for half an hour while its feed was + * flowing is the failure this pins (Plan 315 F95). The card leads with what the + * agent last said and folds its tool lines (F97). + */ +import { act, createElement, type ReactNode } from 'react'; +import { createRoot, type Root } from 'react-dom/client'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { HarnessActivityLine } from '@aflow/schemas'; +import type { HarnessActivityState } from '@aflow/run-view'; + +vi.mock('./providers.js', () => ({ + useApi: () => ({ apiUrl: 'http://api.test', headers: () => ({}) }), +})); + +vi.mock('../lib/fetch-payload.js', () => ({ + // Nothing in these tests has a stored payload; a fetch is a test error. + fetchPayload: () => Promise.reject(new Error('HTTP 404')), +})); + +const { HarnessActivityProvider, HarnessStepCard } = await import('./harness-activity-card.js'); +const { TaskExecutionDetails } = await import('./workflow-run-surface/TaskExecutionDetails.js'); + +const STEP = '00000000-0000-4000-8000-000000000123'; + +const FEED: HarnessActivityLine[] = [ + { kind: 'status', at: 0, text: 'Model claude-opus-5-5' }, + { kind: 'thought', at: 1_000, text: 'Looking at how the parser handles empty input.' }, + { kind: 'tool', at: 1_400, tool: 'Read', text: 'Read src/parser.ts' }, + { + kind: 'tool', + at: 2_000, + tool: 'Edit', + text: 'Edit src/parser.ts', + changes: 'src/parser.ts', + }, + { kind: 'intent', at: 2_100, text: 'Running the parser tests' }, + { kind: 'tool', at: 2_200, tool: 'Bash', text: 'Bash yarn test:file parser.test.ts' }, + { + kind: 'tool_result', + at: 9_000, + tool: 'Bash', + ok: false, + command: true, + text: 'sandbox: cannot write /tmp/cwd-1234', + }, +]; + +function feedOf(lines: HarnessActivityLine[]): Record { + return { + [STEP]: { + lines, + partial: '', + consumedBytes: 1, + lastActivityAtMs: Date.now() - 4_000, + settled: false, + }, + }; +} + +let container: HTMLDivElement; +let root: Root; + +beforeEach(() => { + (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true; + container = document.createElement('div'); + document.body.appendChild(container); + root = createRoot(container); +}); + +afterEach(() => { + act(() => { + root.unmount(); + }); + container.remove(); +}); + +function render(node: ReactNode, feeds: Record): void { + act(() => { + root.render(createElement(HarnessActivityProvider, { feeds, live: true }, node)); + }); +} + +function leadText(): string | null { + return container.querySelector('.harness-activity__lead-text')?.textContent ?? null; +} + +function toolsDisclosure(): HTMLButtonElement | null { + return container.querySelector('button.harness-activity__tools'); +} + +describe('the harness card', () => { + it('leads with what the agent last said, then the files it is changing', () => { + render(createElement(HarnessStepCard, { stepExecutionId: STEP, running: true }), feedOf(FEED)); + + expect(leadText()).toBe('Running the parser tests'); + expect(container.textContent).toMatch(/last active 4\.\ds ago/); + expect(container.querySelector('.harness-activity__file')?.textContent).toBe('src/parser.ts'); + }); + + it('keeps the tool lines folded until asked, naming the call they end on', () => { + render(createElement(HarnessStepCard, { stepExecutionId: STEP, running: true }), feedOf(FEED)); + + const disclosure = toolsDisclosure(); + expect(disclosure?.getAttribute('aria-expanded')).toBe('false'); + expect(disclosure?.textContent).toContain('3 tool calls · last: Bash yarn test:file'); + expect(container.querySelector('.harness-feed')).toBeNull(); + + act(() => { + disclosure?.click(); + }); + expect(toolsDisclosure()?.getAttribute('aria-expanded')).toBe('true'); + expect(container.querySelector('.harness-feed')?.textContent).toContain('Read'); + }); + + it("does not show a command's non-zero exit as a failure", () => { + render(createElement(HarnessStepCard, { stepExecutionId: STEP, running: true }), feedOf(FEED)); + act(() => { + toolsDisclosure()?.click(); + }); + const resultRow = container.querySelector('.harness-feed__row--result'); + expect(resultRow?.textContent).toContain('sandbox: cannot write'); + expect(resultRow?.textContent).toContain('↳'); + }); + + it('leads with the last plain-text message where the agent states no plan', () => { + render( + createElement(HarnessStepCard, { stepExecutionId: STEP, running: true }), + feedOf(FEED.slice(0, 4)), + ); + expect(leadText()).toBe('Looking at how the parser handles empty input.'); + }); +}); + +describe("a run's harness task", () => { + const task = { + taskId: 'implement', + label: 'Implement the change', + status: 'running', + attempt: 1, + taskType: 'operation', + operationId: 'host.harness.run', + workerSessionId: STEP, + startedAt: new Date(Date.now() - 30 * 60_000).toISOString(), + } as unknown as Parameters[0]['task']; + + it("mounts the chat's card fed from the step's activity stream", () => { + render(createElement(TaskExecutionDetails, { task }), feedOf(FEED)); + + expect(container.querySelector('.harness-activity')).not.toBeNull(); + expect(leadText()).toBe('Running the parser tests'); + expect(container.textContent).not.toContain('has not reported yet'); + }); + + it('counts a quiet task from its start rather than reading 0ms', () => { + render(createElement(TaskExecutionDetails, { task }), {}); + + expect(container.textContent).toContain('30.0m'); + expect(container.textContent).not.toContain('0ms'); + }); +}); diff --git a/packages/web-product/src/ui/components/harness-activity-card.test.ts b/packages/web-product/src/ui/components/harness-activity-card.test.ts index d04df071..36fb7fd3 100644 --- a/packages/web-product/src/ui/components/harness-activity-card.test.ts +++ b/packages/web-product/src/ui/components/harness-activity-card.test.ts @@ -11,14 +11,21 @@ import type { HarnessActivityLine } from '@aflow/schemas'; import { activityCounts, + changedFiles, emptyFeedLine, foldHeaderText, isPinnedToBottom, + lastToolLine, + leadLine, leadingSection, lineNeedsExpansion, liveSummaryLabel, parseActivityFeed, + resultMark, + runElapsedMs, + sinceActivityLabel, splitToolLine, + toolDisclosureLabel, } from './harness-activity-card.js'; import { harnessCardTitle, readHarnessId, readResultHarness } from '../lib/harness-title.js'; @@ -79,6 +86,100 @@ describe('what a feed says about itself', () => { }); }); +/** + * Proof of life. A person watching a commission reads what the agent says it is + * doing, how long since it did anything, and the files it is changing — the + * calls are folded behind them. + */ +describe('what the card leads with', () => { + it('leads with the last thing the agent said, never a tool call', () => { + expect(leadLine(FEED)).toEqual({ + kind: 'thought', + at: 61_500, + text: 'The empty case throws. Adding the guard.', + }); + const working: HarnessActivityLine[] = [ + ...FEED, + { kind: 'intent', at: 62_000, text: 'Running the parser tests' }, + { kind: 'tool', at: 62_100, tool: 'Bash', text: 'Bash yarn test:file parser.test.ts' }, + ]; + expect(leadLine(working)).toMatchObject({ kind: 'intent', text: 'Running the parser tests' }); + }); + + it('falls back to the last fact the run reported before the agent has spoken', () => { + const lines: HarnessActivityLine[] = [ + { kind: 'status', at: 0, text: 'Model claude-opus-5-5' }, + { kind: 'tool', at: 300, tool: 'Read', text: 'Read src/parser.ts' }, + ]; + expect(leadLine(lines)).toMatchObject({ kind: 'status', text: 'Model claude-opus-5-5' }); + }); + + it('has no lead line for a feed of calls alone', () => { + expect(leadLine([{ kind: 'tool', at: 1, tool: 'Read', text: 'Read a.ts' }])).toBeUndefined(); + }); + + it('lists the files changed once each, the most recent first', () => { + const lines: HarnessActivityLine[] = [ + { kind: 'tool', at: 1, tool: 'Edit', text: 'Edit a.ts', changes: 'a.ts' }, + { kind: 'tool', at: 2, tool: 'Read', text: 'Read c.ts' }, + { kind: 'tool', at: 3, tool: 'Write', text: 'Write b.ts', changes: 'b.ts' }, + { kind: 'tool', at: 4, tool: 'Edit', text: 'Edit a.ts', changes: 'a.ts' }, + ]; + expect(changedFiles(lines)).toEqual(['a.ts', 'b.ts']); + expect(changedFiles(FEED)).toEqual([]); + }); + + it('names the folded calls by their count and the one they end on', () => { + expect(lastToolLine(FEED)?.text).toBe('Bash yarn test:file parser.test.ts'); + expect(toolDisclosureLabel(FEED)).toBe( + '2 tool calls · last: Bash yarn test:file parser.test.ts', + ); + expect(toolDisclosureLabel(FEED.slice(0, 2))).toBe('2 events'); + }); + + it('reads the time since the harness last wrote anything', () => { + expect(sinceActivityLabel(10_000, 9_500)).toBe('active now'); + expect(sinceActivityLabel(100_000, 88_000)).toBe('last active 12.0s ago'); + }); + + it('counts a run from its start, not from a feed that has not reported', () => { + // The F95 symptom: a run thirty minutes in reading `0ms` because its feed + // was empty. The step's own start is the clock wherever it is known. + const empty = activityCounts([]); + expect(runElapsedMs({ counts: empty, nowMs: 1_800_000, startedAtMs: 0 })).toBe(1_800_000); + expect(liveSummaryLabel(empty, 1_800_000)).toBe('30.0m'); + }); + + it('keeps counting through a quiet spell where only the feed tells the time', () => { + const counts = activityCounts(FEED); + expect(runElapsedMs({ counts, nowMs: 50_000, lastActivityAtMs: 20_000 })).toBe(91_500); + expect(runElapsedMs({ counts, nowMs: 50_000 })).toBe(61_500); + }); +}); + +/** + * A command's exit status is not yet true inside a commission (Plan 315 F84): + * a command that succeeded can still exit non-zero on a sandbox refusal. The + * card shows no verdict for a command until the lane makes it true. + */ +describe("how a call's result is marked", () => { + it("shows no verdict for a command's exit, either way", () => { + expect( + resultMark({ kind: 'tool_result', at: 1, tool: 'Bash', ok: false, command: true, text: '' }), + ).toBe('unjudged'); + expect( + resultMark({ kind: 'tool_result', at: 1, tool: 'Bash', ok: true, command: true, text: '' }), + ).toBe('unjudged'); + }); + + it("shows another tool's error, which is its own report", () => { + expect(resultMark({ kind: 'tool_result', at: 1, tool: 'Read', ok: false, text: '' })).toBe( + 'failed', + ); + expect(resultMark({ kind: 'tool_result', at: 1, tool: 'Read', ok: true, text: '' })).toBe('ok'); + }); +}); + /** * A finished run is read for its answer; a running one for what it is doing. * Leading with a transcript of how the answer was reached buries the answer. diff --git a/packages/web-product/src/ui/components/harness-activity-card.tsx b/packages/web-product/src/ui/components/harness-activity-card.tsx index c418a4e9..0f964d90 100644 --- a/packages/web-product/src/ui/components/harness-activity-card.tsx +++ b/packages/web-product/src/ui/components/harness-activity-card.tsx @@ -4,12 +4,17 @@ * A harness step, while it works and once it is done. * * A harness runs for minutes and says nothing until the end, so the wait is - * the whole of the experience: what it is reading, what it tried, how the call - * came back. This is that wait, under the step rather than inside the agent's - * message — an executor's view of its own work is not the run speaking. + * the whole of the experience. What a person watching needs from it is proof + * of life: what the agent says it is doing, or the last thing it wrote, and how + * long ago it last did anything; then the files it is changing. The tool calls + * are the record of how, folded behind a disclosure — a scroll of invocations + * cannot be read in real time. This is that wait, under the step rather than + * inside the agent's message — an executor's view of its own work is not the + * run speaking. * - * One component, two places: the step row in the timeline and the task row in - * a run. Both show the same thing, because they are the same thing. + * One component, every place: the chat, the step row in the timeline and the + * task row in a run all mount `HarnessStepCard`. They show the same thing, + * because they are the same thing. * * Nothing vanishes when the step ends. The answer takes the top and the feed * folds to its counts under it, so a reader who arrives late reads what came of @@ -33,6 +38,7 @@ import { Badge, Icon, Spinner, Text } from '@aflow/design-system'; import { fetchPayload } from '../lib/fetch-payload.js'; import { harnessCardTitle, readHarnessId, readResultHarness } from '../lib/harness-title.js'; import { useApi } from './providers.js'; +import { useNowTicker } from './workflow-run-surface/useNowTicker.js'; import { HostHarnessBody, HostHarnessHeader, @@ -92,40 +98,59 @@ export function HarnessActivityProvider({ ); } -export function useHarnessActivityLines( +/** One step's feed as the surface's provider holds it, when it holds one. */ +export function useHarnessActivityFeed( stepExecutionId: string | undefined, -): HarnessActivityLine[] { +): HarnessActivityState | undefined { const { feeds } = useContext(HarnessActivityContext); - if (stepExecutionId === undefined) return NO_LINES; - return feeds[stepExecutionId]?.lines ?? NO_LINES; + return stepExecutionId === undefined ? undefined : feeds[stepExecutionId]; } -/** Every feed this surface holds — for a row that reads many at once. */ /** * The card for one step, reading that step's feed from the surface's provider. - * The chat renders it while the step runs and again on the step's result, so - * the feed a person watched is the one folded above the result. + * + * The only way a surface mounts the card: the chat renders it while the step + * runs and again on the step's result, the run timeline under a step row, and a + * run's view under its task — so a commission watched from its run reads the + * same card as a call watched from the chat, from the same feed. */ export function HarnessStepCard({ stepExecutionId, running, result, + harness, + startedAt, }: { stepExecutionId: string | undefined; running: boolean; result?: HostHarnessResult | undefined; + harness?: string | undefined; + /** When the step started, where the surface knows it — the elapsed clock. */ + startedAt?: string | number | undefined; }): ReactNode { - const lines = useHarnessActivityLines(stepExecutionId); + const feed = useHarnessActivityFeed(stepExecutionId); + const startedAtMs = + typeof startedAt === 'number' + ? startedAt + : startedAt === undefined + ? Number.NaN + : Date.parse(startedAt); return ( 0 + ? { lastActivityAtMs: feed.lastActivityAtMs } + : {})} + {...(Number.isNaN(startedAtMs) ? {} : { startedAtMs })} /> ); } +/** Every feed this surface holds — for a row that reads many at once. */ export function useHarnessActivityFeeds(): Record { return useContext(HarnessActivityContext).feeds; } @@ -225,8 +250,95 @@ export function lineNeedsExpansion(line: HarnessActivityLine): boolean { return line.text.includes('\n') || line.text.length > ROW_FITS_CHARS; } -export function liveSummaryLabel(counts: ActivityCounts): string { - const parts = [formatDuration(counts.elapsedMs)]; +/** + * The line the card leads with: the last thing the agent said — what it says it + * is doing, or what it wrote in plain text — or, before it has said anything, + * the last fact the run reported about itself. + * + * A tool call never leads. A scroll of invocations is the record of the work, + * not an account of it, and is what a person watching cannot read in real time. + */ +export function leadLine(lines: readonly HarnessActivityLine[]): HarnessActivityLine | undefined { + let status: HarnessActivityLine | undefined; + for (let i = lines.length - 1; i >= 0; i -= 1) { + const line = lines[i]; + if (line === undefined) continue; + if (line.kind === 'intent' || line.kind === 'thought') return line; + if (line.kind === 'status' && status === undefined) status = line; + } + return status; +} + +/** The files the run has changed, the most recently touched first. */ +export function changedFiles(lines: readonly HarnessActivityLine[]): string[] { + const seen = new Set(); + const files: string[] = []; + for (let i = lines.length - 1; i >= 0; i -= 1) { + const line = lines[i]; + if (line?.kind !== 'tool' || line.changes === undefined || seen.has(line.changes)) continue; + seen.add(line.changes); + files.push(line.changes); + } + return files; +} + +/** The last call the run made — what the folded tool lines say they end on. */ +export function lastToolLine( + lines: readonly HarnessActivityLine[], +): Extract | undefined { + for (let i = lines.length - 1; i >= 0; i -= 1) { + const line = lines[i]; + if (line?.kind === 'tool') return line; + } + return undefined; +} + +/** + * How a call's result is marked. + * + * A command's exit status is not yet one a reader can trust (Plan 315 F84): + * inside a commission a command that succeeded can still exit non-zero on a + * sandbox refusal, so every check the agent ran would read as a failure. Until + * the lane makes that status true, a command's result is shown without a + * verdict. Every other tool's error is its own report, and is shown as one. + */ +export function resultMark( + line: Extract, +): 'ok' | 'failed' | 'unjudged' { + if (line.command === true) return 'unjudged'; + return line.ok ? 'ok' : 'failed'; +} + +/** + * How long the run has been going, as of `nowMs`. + * + * From the step's start where the surface knows it. Otherwise from the feed: + * the last line's offset, plus the wall-clock time since that line arrived — so + * a run that has been quiet for a minute does not read as a minute younger. + */ +export function runElapsedMs(opts: { + counts: ActivityCounts; + nowMs: number; + startedAtMs?: number | undefined; + lastActivityAtMs?: number | undefined; +}): number { + if (opts.startedAtMs !== undefined) return Math.max(0, opts.nowMs - opts.startedAtMs); + if (opts.lastActivityAtMs === undefined) return opts.counts.elapsedMs; + return opts.counts.elapsedMs + Math.max(0, opts.nowMs - opts.lastActivityAtMs); +} + +/** The time since the harness last wrote anything — the card's proof of life. */ +export function sinceActivityLabel(nowMs: number, lastActivityAtMs: number): string { + const ms = Math.max(0, nowMs - lastActivityAtMs); + if (ms < 2_000) return 'active now'; + return `last active ${formatDuration(ms)} ago`; +} + +export function liveSummaryLabel( + counts: ActivityCounts, + elapsedMs: number = counts.elapsedMs, +): string { + const parts = [formatDuration(elapsedMs)]; if (counts.turns > 0) parts.push(`${String(counts.turns)} turn${counts.turns === 1 ? '' : 's'}`); if (counts.toolCalls > 0) { parts.push(`${String(counts.toolCalls)} tool call${counts.toolCalls === 1 ? '' : 's'}`); @@ -385,7 +497,7 @@ function ActivityLineRow({ expanded: boolean; onToggle: () => void; }): ReactNode { - if (line.kind === 'thought') { + if (line.kind === 'thought' || line.kind === 'intent') { return
{line.text}
; } if (line.kind === 'status') { @@ -394,16 +506,17 @@ function ActivityLineRow({ const { tool, detail } = splitToolLine(line); const rowClass = `harness-feed__row${line.kind === 'tool_result' ? ' harness-feed__row--result' : ''}`; + const mark = line.kind === 'tool_result' ? resultMark(line) : null; const body = ( <> - {line.kind === 'tool_result' ? ( + {mark === 'ok' || mark === 'failed' ? ( ) : ( - › + {mark === 'unjudged' ? '↳' : '›'} )} {tool} @@ -490,10 +603,48 @@ export interface HarnessActivityCardProps { folder?: string | undefined; /** Whose feed this is — the handle an expanded row is addressed against. */ stepExecutionId?: string | undefined; + /** When the harness last wrote anything, by this tab's clock. Live feeds only. */ + lastActivityAtMs?: number | undefined; + /** When the step started, where the surface knows it. */ + startedAtMs?: number | undefined; } const NO_EXPANDED_LINES: ReadonlySet = new Set(); +/** At most this many changed files are named before the rest are counted. */ +const FILES_SHOWN = 6; + +function ChangedFilesLine({ files }: { files: string[] }): ReactNode { + const shown = files.slice(0, FILES_SHOWN); + const more = files.length - shown.length; + return ( +
+ + {`Changing ${String(files.length)} file${files.length === 1 ? '' : 's'}`} + +
    + {shown.map((file) => ( +
  • + {file} +
    • + ))} + {more > 0 &&
  • {`+${String(more)} more`}
    • } +
+
+ ); +} + +/** The folded tool lines say how many there are and the call they end on. */ +export function toolDisclosureLabel(lines: readonly HarnessActivityLine[]): string { + const counts = activityCounts([...lines]); + const head = + counts.toolCalls > 0 + ? `${String(counts.toolCalls)} tool call${counts.toolCalls === 1 ? '' : 's'}` + : `${String(counts.events)} event${counts.events === 1 ? '' : 's'}`; + const last = lastToolLine(lines); + return last === undefined ? head : `${head} · last: ${last.text}`; +} + export function HarnessActivityCard({ lines, running, @@ -502,23 +653,34 @@ export function HarnessActivityCard({ harness, folder, stepExecutionId, + lastActivityAtMs, + startedAtMs, }: HarnessActivityCardProps): ReactNode { // Open while it runs, folded once it is done: a running step is the only // thing on screen worth watching, and a finished one has a result to read. const [openOverride, setOpenOverride] = useState(null); const open = openOverride ?? running; + // The tool lines stay folded whether or not the run is: the lead line and + // the files are what a person watching reads, and the calls are for the one + // who goes looking. + const [toolsOpen, setToolsOpen] = useState(false); const storedRef = activityRef ?? result?.activityRef; const stored = useStoredFeed(storedRef, open && lines.length === 0); const shown = lines.length > 0 ? lines : stored.lines; const counts = activityCounts(shown); const live = useHarnessFeedsLive(); + const nowMs = useNowTicker(running); + const lead = leadLine(shown); + const files = changedFiles(shown); + const elapsedMs = runElapsedMs({ counts, nowMs, startedAtMs, lastActivityAtMs }); const [expandedLines, setExpandedLines] = useState>(NO_EXPANDED_LINES); // A row is remembered by its position in one step's feed, so another step's // feed would open rows nobody asked for. useEffect(() => { setExpandedLines(NO_EXPANDED_LINES); + setToolsOpen(false); }, [stepExecutionId]); const toggleLine = useCallback((index: number) => { setExpandedLines((current) => { @@ -562,19 +724,53 @@ export function HarnessActivityCard({ )} - {running ? liveSummaryLabel(counts) : foldHeaderText(counts, storedRef !== undefined)} + {running + ? liveSummaryLabel(counts, elapsedMs) + : foldHeaderText(counts, storedRef !== undefined)} {open && (
{shown.length > 0 ? ( - + <> + {(lead !== undefined || (running && lastActivityAtMs !== undefined)) && ( +
+ {lead !== undefined && ( +
+ {lead.text} +
+ )} + {running && lastActivityAtMs !== undefined && ( + + {sinceActivityLabel(nowMs, lastActivityAtMs)} + + )} +
+ )} + {files.length > 0 && } + + {toolsOpen && ( + + )} + ) : stored.loading ? ( ) : ( diff --git a/packages/web-product/src/ui/components/run-timeline/StepGroupCard.tsx b/packages/web-product/src/ui/components/run-timeline/StepGroupCard.tsx index 3fc77a7f..97d663cf 100644 --- a/packages/web-product/src/ui/components/run-timeline/StepGroupCard.tsx +++ b/packages/web-product/src/ui/components/run-timeline/StepGroupCard.tsx @@ -9,8 +9,7 @@ import { useApi } from '../providers.js'; import { HARNESS_RUN_OPERATION, - HarnessActivityCard, - useHarnessActivityLines, + HarnessStepCard, useHarnessName, useHarnessResult, } from '../harness-activity-card.js'; @@ -260,19 +259,18 @@ const HARNESS_RUNNING_STATUSES = new Set([ ]); function HarnessStepActivity({ group }: { group: StepGroup }) { - const lines = useHarnessActivityLines(group.stepExecutionId); const running = HARNESS_RUNNING_STATUSES.has(group.status); const result = useHarnessResult(running ? undefined : group.outputRef); const harness = useHarnessName(group.inputRef); return (
-
); diff --git a/packages/web-product/src/ui/components/workbench/Workbench.tsx b/packages/web-product/src/ui/components/workbench/Workbench.tsx index e41bf56c..99703c7b 100644 --- a/packages/web-product/src/ui/components/workbench/Workbench.tsx +++ b/packages/web-product/src/ui/components/workbench/Workbench.tsx @@ -82,6 +82,14 @@ export function Workbench({ [skillsQuery.data], ); + // Every workflow's name, the platform's own included: a conversation's + // commission or publication is a run of a system workflow, which the skill + // list above leaves out. + const workflowNames = useMemo( + () => new Map((skillsQuery.data?.workflows ?? []).map((w) => [w.slug, w.name] as const)), + [skillsQuery.data], + ); + // What each paused run is waiting on, said as a person. Derived from the // open items rather than the run row, so the label and the card that // resolves it can never disagree about who is being asked. @@ -145,6 +153,8 @@ export function Workbench({ diff --git a/packages/web-product/src/ui/components/workbench/WorkbenchConversationRow.test.ts b/packages/web-product/src/ui/components/workbench/WorkbenchConversationRow.test.ts new file mode 100644 index 00000000..b9093624 --- /dev/null +++ b/packages/web-product/src/ui/components/workbench/WorkbenchConversationRow.test.ts @@ -0,0 +1,121 @@ +/** + * @vitest-environment jsdom + */ +/** + * A conversation's row in the Workbench, rendered: it says whether the + * conversation is live — the agent mid-turn, a commission it started still + * running and at what phase — so the operator need not open each one to find + * out (Plan 315 F75). + */ +import { act, createElement } from 'react'; +import { createRoot, type Root } from 'react-dom/client'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import type { Session } from '../../lib/types.js'; + +vi.mock('next/navigation', () => ({ useRouter: () => ({ push: () => undefined }) })); + +const { WorkbenchConversationRow } = await import('./WorkbenchConversationRow.js'); +const { conversationLiveState, runsByConversation } = await import('./conversationLiveState.js'); + +const SESSION: Session = { + sessionId: '00000000-0000-4000-8000-0000000000c1', + agentId: 'cybernetic-helmsman', + agentVersion: '1', + status: 'PAUSED', + createdAt: '2026-10-16T09:00:00.000Z', + lastActivityAt: '2026-10-16T10:00:00.000Z', + metadata: { + title: 'Fix the parser', + summary: 'Guarding the empty input case', + titleSource: 'generated', + revision: 1, + } as Session['metadata'], +}; + +let container: HTMLDivElement; +let root: Root; + +beforeEach(() => { + (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true; + container = document.createElement('div'); + document.body.appendChild(container); + root = createRoot(container); +}); + +afterEach(() => { + act(() => { + root.unmount(); + }); + container.remove(); +}); + +function renderRow(session: Session, runs: Parameters[0]): void { + const live = conversationLiveState( + session, + runsByConversation(runs, new Map([['local-code-change', 'Local Code Change']])), + ); + act(() => { + root.render( + createElement(WorkbenchConversationRow, { + session, + ownerName: null, + presence: [], + personFor: () => undefined, + unread: 0, + unreadLabel: '0', + onOpen: () => undefined, + onRename: () => undefined, + onRegenerate: () => undefined, + canEdit: false, + live, + }), + ); + }); +} + +function liveLine(): string | null { + return container.querySelector('[data-testid="conversation-live"]')?.textContent ?? null; +} + +const COMMISSION = { + runId: 'run-1', + status: 'running', + workflowSlug: 'local-code-change', + workflowRevision: 1, + startedAt: '2026-10-16T10:00:00.000Z', + completedAt: null, + durationMs: null, + totalCostCents: null, + totalTokens: null, + evalScore: null, + evalVerdict: null, + taskCount: 3, + pausedReason: null, + initiatedByUserId: null, + sessionId: SESSION.sessionId, + rootSessionId: SESSION.sessionId, +}; + +describe("a conversation's row", () => { + it('renders the phase of a commission the conversation is running', () => { + renderRow(SESSION, [COMMISSION]); + expect(liveLine()).toContain('Local Code Change · running'); + }); + + it('says what a paused commission is waiting on', () => { + renderRow(SESSION, [{ ...COMMISSION, status: 'paused', pausedReason: 'needs_decision' }]); + expect(liveLine()).toContain('Local Code Change · Needs a decision'); + }); + + it('says the agent is working while it is mid-turn', () => { + renderRow({ ...SESSION, status: 'RUNNING' }, []); + expect(liveLine()).toContain('Working'); + }); + + it('says nothing live about a conversation at rest', () => { + renderRow(SESSION, []); + expect(liveLine()).toBeNull(); + expect(container.textContent).toContain('Fix the parser'); + }); +}); diff --git a/packages/web-product/src/ui/components/workbench/WorkbenchConversationRow.tsx b/packages/web-product/src/ui/components/workbench/WorkbenchConversationRow.tsx index 6dcd337d..26068775 100644 --- a/packages/web-product/src/ui/components/workbench/WorkbenchConversationRow.tsx +++ b/packages/web-product/src/ui/components/workbench/WorkbenchConversationRow.tsx @@ -7,6 +7,7 @@ import { ParticipantBadge } from '../room/ParticipantBadge.js'; import { conversationTitle } from '../../lib/conversation-title.js'; import type { Session } from '../../lib/types.js'; import { formatAgo } from './WorkbenchRunRow.js'; +import { isLive, type ConversationLiveState } from './conversationLiveState.js'; export interface ConversationPresence { userId: string; @@ -31,6 +32,47 @@ export interface WorkbenchConversationRowProps { * that can only fail — the chat header hides them on the same test. */ canEdit: boolean; + /** Whether the agent is mid-turn, and the runs it started that are still going. */ + live?: ConversationLiveState | undefined; +} + +/** At most this many runs are named on a row before the rest are counted. */ +const RUNS_SHOWN = 2; + +/** + * What is live in a conversation, on a line of its own: the agent's turn, then + * each run it is waiting on with the phase that run is at. Absent when nothing + * is — a resting conversation is the ordinary state and says nothing. + */ +export function ConversationLiveLine({ live }: { live: ConversationLiveState }) { + if (!isLive(live)) return null; + const shown = live.runs.slice(0, RUNS_SHOWN); + const more = live.runs.length - shown.length; + return ( + + {live.inTurn ? Working : null} + {shown.map((run) => ( + + {`${run.name} · ${run.phase}`} + + ))} + {more > 0 ? ( + + {`+${String(more)} more`} + + ) : null} + + ); } /** @@ -59,6 +101,7 @@ export function WorkbenchConversationRow({ onRename, onRegenerate, canEdit, + live, }: WorkbenchConversationRowProps) { const [editing, setEditing] = useState(false); const [draft, setDraft] = useState(''); @@ -189,6 +232,9 @@ export function WorkbenchConversationRow({ {formatAgo(activeAt)} + {/* Whether it is live comes before what it is about: it is the reason + to open this conversation now rather than later. */} + {live ? : null} {/* Absent, not blank, when there is no synopsis: an unnamed row collapses to one line rather than reserving space for nothing. */} {summary ? ( diff --git a/packages/web-product/src/ui/components/workbench/WorkbenchRecentConversations.tsx b/packages/web-product/src/ui/components/workbench/WorkbenchRecentConversations.tsx index e2b515fe..b18c5814 100644 --- a/packages/web-product/src/ui/components/workbench/WorkbenchRecentConversations.tsx +++ b/packages/web-product/src/ui/components/workbench/WorkbenchRecentConversations.tsx @@ -7,21 +7,25 @@ import { Column, Row, Spinner, Text } from '@aflow/design-system'; import { useApiQuery } from '../../hooks/useApiQuery.js'; import { formatUnread, useRoomActivity } from '../../hooks/use-room-activity.js'; -import { useSessionMetadataRefresh, useSessionRename } from '../../hooks/use-session-metadata.js'; +import { useSessionListRefresh, useSessionRename } from '../../hooks/use-session-metadata.js'; import { useSpacePeople } from '../../hooks/use-space-people.js'; import { useCurrentUser } from '../user-avatar.js'; import { useSpace } from '../providers.js'; import { spaceRoute } from '../../lib/space-routes.js'; import type { Session } from '../../lib/types.js'; +import type { SpaceWorkflowRunSummary } from '../../hooks/use-space-workflow-runs.js'; import { WorkbenchConversationRow } from './WorkbenchConversationRow.js'; +import { conversationLiveState, runsByConversation } from './conversationLiveState.js'; import { WorkbenchSection } from './WorkbenchSection.js'; /** * Recent conversations (Plan 228 §3.4): **Helmsman only**. Runner sessions are * reflected in Skill runs (and never start on their own), so listing them here * would double-count. A Helmsman chat doesn't "end", so idle/paused is the - * normal resting state and labeling it would be noise. The exception is - * failure: a FAILED session is rare and actionable, so it gets a badge. + * normal resting state and labeling it would be noise. The exceptions are the + * states someone would otherwise open the conversation to find out: failure, + * which is rare and actionable, and being live — the Helmsman mid-turn, or a + * commission or publication it started still running, with that run's phase. * Clicking resumes in place. * * Rows lead with the conversation's name and its synopsis. Recency alone @@ -51,14 +55,26 @@ interface InvitationItem { /** How many of the shared list this section shows. */ const RECENT_CONVERSATIONS_SHOWN = 8; +const NO_RUNS: SpaceWorkflowRunSummary[] = []; +const NO_NAMES: ReadonlyMap = new Map(); + export function WorkbenchRecentConversations({ spaceId, spaceSlug, onRevealChat, + activeRuns = NO_RUNS, + workflowNames = NO_NAMES, }: { spaceId: string; spaceSlug: string; onRevealChat?: () => void; + /** + * The space's running and paused runs — the Workbench's own feed, handed down + * rather than asked for again, so a row's runs cost no request of their own. + */ + activeRuns?: readonly SpaceWorkflowRunSummary[]; + /** Skill names by slug, for naming a conversation's runs. */ + workflowNames?: ReadonlyMap; }) { const router = useRouter(); const personFor = useSpacePeople(spaceId); @@ -98,9 +114,14 @@ export function WorkbenchRecentConversations({ const activity = useRoomActivity(spaceId, sessionIds); const { rename, regenerate } = useSessionRename(spaceId); - // A conversation finishing its name is usually not the one anyone has open, - // so the refresh rides the space channel rather than a per-row subscription. - useSessionMetadataRefresh(spaceId); + // A conversation finishing its name, or starting and ending a turn, is + // usually not the one anyone has open, so the refresh rides the space channel + // rather than a per-row subscription. + useSessionListRefresh(spaceId); + const runsFor = useMemo( + () => runsByConversation(activeRuns, workflowNames), + [activeRuns, workflowNames], + ); const sessions = useMemo(() => { const list = query.data?.sessions ?? []; @@ -220,6 +241,7 @@ export function WorkbenchRecentConversations({ .filter((p) => p.userId !== currentUser?.userId)} personFor={personFor} canEdit={canEdit} + live={conversationLiveState(s, runsFor)} // Awareness, not attention: this says the room moved without // you, and stays quieter than the approval and mention cards // that ask you to act. diff --git a/packages/web-product/src/ui/components/workbench/conversationLiveState.test.ts b/packages/web-product/src/ui/components/workbench/conversationLiveState.test.ts new file mode 100644 index 00000000..833c6628 --- /dev/null +++ b/packages/web-product/src/ui/components/workbench/conversationLiveState.test.ts @@ -0,0 +1,102 @@ +/** + * Whether a conversation in the Workbench's list is live, read off the session + * list and the space's active-run feed the Workbench already holds — never a + * request per row. + */ +import { describe, expect, it } from 'vitest'; + +import type { SpaceWorkflowRunSummary } from '../../hooks/use-space-workflow-runs.js'; +import { + conversationLiveState, + isLive, + nameFromSlug, + runsByConversation, +} from './conversationLiveState.js'; + +const CONVERSATION = '00000000-0000-4000-8000-0000000000c1'; +const OTHER = '00000000-0000-4000-8000-0000000000c2'; +const DELEGATE = '00000000-0000-4000-8000-0000000000d1'; + +function run(overrides: Partial): SpaceWorkflowRunSummary { + return { + runId: 'run-1', + status: 'running', + workflowSlug: 'local-code-change', + workflowRevision: 1, + startedAt: '2026-10-16T10:00:00.000Z', + completedAt: null, + durationMs: null, + totalCostCents: null, + totalTokens: null, + evalScore: null, + evalVerdict: null, + taskCount: 3, + pausedReason: null, + initiatedByUserId: null, + sessionId: CONVERSATION, + rootSessionId: CONVERSATION, + ...overrides, + }; +} + +const NAMES = new Map([['local-code-change', 'Local Code Change']]); + +describe("a conversation's live state", () => { + it('is in a turn while the agent is working, and resting otherwise', () => { + const none = new Map(); + expect(conversationLiveState({ sessionId: CONVERSATION, status: 'RUNNING' }, none).inTurn).toBe( + true, + ); + expect( + conversationLiveState({ sessionId: CONVERSATION, status: 'WAITING_ON_CHILD' }, none).inTurn, + ).toBe(true); + const resting = conversationLiveState({ sessionId: CONVERSATION, status: 'PAUSED' }, none); + expect(resting.inTurn).toBe(false); + expect(isLive(resting)).toBe(false); + }); + + it('carries the runs it started with their phase, paused first', () => { + const runsFor = runsByConversation( + [ + run({ runId: 'commission' }), + run({ + runId: 'publication', + workflowSlug: 'local-publish', + status: 'paused', + pausedReason: 'needs_decision', + }), + ], + NAMES, + ); + const state = conversationLiveState({ sessionId: CONVERSATION, status: 'PAUSED' }, runsFor); + expect(isLive(state)).toBe(true); + expect(state.runs).toEqual([ + { runId: 'publication', name: 'Local publish', status: 'paused', phase: 'Needs a decision' }, + { runId: 'commission', name: 'Local Code Change', status: 'running', phase: 'running' }, + ]); + }); + + it('counts a run a delegate started as the root conversation’s', () => { + const runsFor = runsByConversation([run({ sessionId: DELEGATE })], NAMES); + expect(runsFor.get(CONVERSATION)?.map((r) => r.runId)).toEqual(['run-1']); + expect(runsFor.get(DELEGATE)?.map((r) => r.runId)).toEqual(['run-1']); + }); + + it("says nothing about another conversation's runs, or a run that ended", () => { + const runsFor = runsByConversation( + [ + run({ sessionId: OTHER, rootSessionId: OTHER }), + run({ runId: 'done', status: 'succeeded' }), + ], + NAMES, + ); + expect( + conversationLiveState({ sessionId: CONVERSATION, status: 'PAUSED' }, runsFor).runs, + ).toEqual([]); + }); + + it('names a run whose skill the space did not name from its slug', () => { + expect(nameFromSlug('local-publish')).toBe('Local publish'); + expect(nameFromSlug('code_review')).toBe('Code review'); + }); +}); diff --git a/packages/web-product/src/ui/components/workbench/conversationLiveState.ts b/packages/web-product/src/ui/components/workbench/conversationLiveState.ts new file mode 100644 index 00000000..70b066e0 --- /dev/null +++ b/packages/web-product/src/ui/components/workbench/conversationLiveState.ts @@ -0,0 +1,104 @@ +import type { SpaceWorkflowRunSummary } from '../../hooks/use-space-workflow-runs.js'; +import type { Session } from '../../lib/types.js'; +import { humanizePauseReason } from './WorkbenchRunRow.js'; + +/** + * Whether a conversation is live, read off what the Workbench already holds. + * + * Two things make a conversation live, and a row says both: the agent is in a + * turn — its session's hot status, which the list carries — or a run it started + * is still going, a commission or a publication, from the space's active-run + * feed. Neither is fetched per row: the list and the feed are each one query, + * refreshed by the space's entity events. + */ +export interface ConversationRunPhase { + runId: string; + /** The skill the run is of, as the space names it. */ + name: string; + status: 'running' | 'paused'; + /** What the run is at, said as a person would: `running`, `Needs a decision`. */ + phase: string; +} + +export interface ConversationLiveState { + /** True while the agent is taking a turn — working, or waiting on a delegate. */ + inTurn: boolean; + /** The runs this conversation started that have not ended, paused first. */ + runs: ConversationRunPhase[]; +} + +/** Statuses in which the agent is doing something rather than resting on you. */ +const IN_TURN_STATUSES: ReadonlySet = new Set([ + 'QUEUED', + 'RUNNING', + 'WAITING_ON_CHILD', + 'CANCELLING', +]); + +/** A run's slug read as a name, where the space's list does not name it. */ +export function nameFromSlug(slug: string): string { + const words = slug.replace(/[_-]+/g, ' ').trim(); + return words.length === 0 ? slug : words.charAt(0).toUpperCase() + words.slice(1); +} + +export function runPhase(run: Pick): string { + if (run.status === 'paused') { + return run.pausedReason ? humanizePauseReason(run.pausedReason) : 'paused'; + } + return 'running'; +} + +/** + * The runs a conversation waits on, keyed by conversation. + * + * A run belongs to the conversation that steers it: the session that started + * it, or the root of the chain it was started from — a commission a delegate + * started is still the operator's conversation's work. + */ +export function runsByConversation( + runs: readonly SpaceWorkflowRunSummary[], + workflowNames: ReadonlyMap, +): Map { + const byConversation = new Map(); + for (const run of runs) { + const status = + run.status === 'paused' ? 'paused' : run.status === 'running' ? 'running' : undefined; + if (status === undefined) continue; + const phase: ConversationRunPhase = { + runId: run.runId, + name: workflowNames.get(run.workflowSlug) ?? nameFromSlug(run.workflowSlug), + status, + phase: runPhase(run), + }; + const owners = new Set( + [run.rootSessionId, run.sessionId].filter((id): id is string => typeof id === 'string'), + ); + for (const owner of owners) { + const list = byConversation.get(owner) ?? []; + list.push(phase); + byConversation.set(owner, list); + } + } + // A paused run is waiting on someone, which is the more urgent of the two. + for (const list of byConversation.values()) { + list.sort((a, b) => (a.status === b.status ? 0 : a.status === 'paused' ? -1 : 1)); + } + return byConversation; +} + +const NO_RUNS: ConversationRunPhase[] = []; + +export function conversationLiveState( + session: Pick, + runsFor: ReadonlyMap, +): ConversationLiveState { + return { + inTurn: IN_TURN_STATUSES.has(session.status), + runs: runsFor.get(session.sessionId) ?? NO_RUNS, + }; +} + +/** Whether there is anything live to say about the conversation at all. */ +export function isLive(state: ConversationLiveState): boolean { + return state.inTurn || state.runs.length > 0; +} diff --git a/packages/web-product/src/ui/components/workflow-run-surface/TaskExecutionDetails.tsx b/packages/web-product/src/ui/components/workflow-run-surface/TaskExecutionDetails.tsx index ab13ffe2..b5b29a4f 100644 --- a/packages/web-product/src/ui/components/workflow-run-surface/TaskExecutionDetails.tsx +++ b/packages/web-product/src/ui/components/workflow-run-surface/TaskExecutionDetails.tsx @@ -10,8 +10,7 @@ import { Pill } from './WorkflowRunSurfaceParts.js'; import { CodingTranscriptView } from './CodingTranscriptView.js'; import { HARNESS_RUN_OPERATION, - HarnessActivityCard, - useHarnessActivityLines, + HarnessStepCard, useHarnessName, useHarnessResult, } from '../harness-activity-card.js'; @@ -256,25 +255,25 @@ const HARNESS_RUNNING_STATUSES = new Set(['scheduled', 'running']); /** * The harness card for a task in a run. * - * The same card the chat timeline shows under a step, keyed by the step the run - * recorded for the task; the stored feed on the result takes over once the run - * has written it. + * The same card the chat shows for a direct call, keyed by the step the run + * recorded for the task and fed from that step's activity stream — which the + * surface's provider folds wherever the run is shown (`WorkflowRunSurfaceContainer` + * outside the chat). The stored feed on the result takes over once the run has + * written it. */ -function HarnessTaskActivity({ task }: { task: WorkflowSurfaceTaskState }) { +export function HarnessTaskActivity({ task }: { task: WorkflowSurfaceTaskState }) { const running = HARNESS_RUNNING_STATUSES.has(task.status); - const stepExecutionId = harnessFeedStepId(task); - const lines = useHarnessActivityLines(stepExecutionId); const result = useHarnessResult(running ? undefined : task.outputRef); const harness = useHarnessName(task.inputRef); return (
-
); diff --git a/packages/web-product/src/ui/components/workflow-run-surface/WorkflowRunSurfaceContainer.tsx b/packages/web-product/src/ui/components/workflow-run-surface/WorkflowRunSurfaceContainer.tsx index 2e379a13..a997a4c1 100644 --- a/packages/web-product/src/ui/components/workflow-run-surface/WorkflowRunSurfaceContainer.tsx +++ b/packages/web-product/src/ui/components/workflow-run-surface/WorkflowRunSurfaceContainer.tsx @@ -5,7 +5,8 @@ import { useQueryClient } from '@tanstack/react-query'; import { runViewReducer, initialRunViewState } from '@aflow/run-view'; import type { SessionEvent } from '../../lib/types.js'; import { useApiQuery } from '../../hooks/useApiQuery.js'; -import { useSessionEventListener } from '../../hooks/use-session-events.js'; +import { useSessionEventListener, useSessionLiveDeltas } from '../../hooks/use-session-events.js'; +import { HarnessActivityProvider } from '../harness-activity-card.js'; import { WorkflowRunSurface } from './WorkflowRunSurface.js'; import { useWorkflowRunUsageRefresh } from './useWorkflowRunUsageRefresh.js'; import { useWorkflowRunPauseRefresh } from './useWorkflowRunPauseRefresh.js'; @@ -62,6 +63,23 @@ export function WorkflowRunSurfaceContainer({ runId, spaceId }: WorkflowRunSurfa replayHistory: true, }); + // 4. The live frames of the same session — a harness task's activity feed + // among them, since the server reads a run's task steps for every session + // watching the run. Folded by the same reducer the chat uses and handed to + // the surface through the same provider, so a commission's task row shows + // the card a direct call shows in the chat rather than a feed that never + // reports. Passive, behind the cursor-bearing acquire above. + useSessionLiveDeltas(data ? originatingSessionId : null, (frame) => { + dispatch({ + type: 'LIVE_DELTA', + stepExecutionId: frame.stepExecutionId, + channel: frame.channel, + offset: frame.offset, + delta: frame.delta, + timestamp: new Date().toISOString(), + }); + }); + const runState = reducerState.workflowRuns[runId]; const queryClient = useQueryClient(); @@ -89,18 +107,25 @@ export function WorkflowRunSurfaceContainer({ runId, spaceId }: WorkflowRunSurfa } return ( - undefined} - // Only surface a connection state when we actually have a session to - // tail; a run with no originating session is a static snapshot. - {...(originatingSessionId ? { sseConnected: isConnected } : {})} - /> + // A run with no session to tail has no live channel to be down, so its + // stored feeds read as they are rather than as a dropped connection. + + undefined} + // Only surface a connection state when we actually have a session to + // tail; a run with no originating session is a static snapshot. + {...(originatingSessionId ? { sseConnected: isConnected } : {})} + /> + ); } diff --git a/packages/web-product/src/ui/hooks/use-coalesced-invalidation.test.ts b/packages/web-product/src/ui/hooks/use-coalesced-invalidation.test.ts index 20d4879c..14a3401b 100644 --- a/packages/web-product/src/ui/hooks/use-coalesced-invalidation.test.ts +++ b/packages/web-product/src/ui/hooks/use-coalesced-invalidation.test.ts @@ -45,7 +45,7 @@ vi.mock('./useApiQuery.js', () => ({ const { ENTITY_EVENT_REFETCH_WINDOW_MS } = await import('./use-coalesced-invalidation.js'); const { useSpaceWorkflowRuns, spaceWorkflowRunsKey } = await import('./use-space-workflow-runs.js'); -const { useSessionMetadataRefresh } = await import('./use-session-metadata.js'); +const { useSessionListRefresh } = await import('./use-session-metadata.js'); (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true; @@ -54,7 +54,7 @@ const BURST = 100; function Workbench(): null { useSpaceWorkflowRuns(SPACE_ID); - useSessionMetadataRefresh(SPACE_ID); + useSessionListRefresh(SPACE_ID); return null; } @@ -132,6 +132,20 @@ describe('refreshing on the space channel', () => { expect(invalidationsOf(spaceWorkflowRunsKey(SPACE_ID))).toBe(0); }); + it('refetches the conversation list when a turn begins or rests, once for the burst', async () => { + deliver([ + event('entity.session.status', { payload: { sessionId: 's1', status: 'RUNNING' } }), + event('entity.session.status', { payload: { sessionId: 's2', status: 'PAUSED' } }), + event('entity.session.described'), + ]); + + await act(async () => { + await vi.advanceTimersByTimeAsync(ENTITY_EVENT_REFETCH_WINDOW_MS); + }); + + expect(invalidationsOf(['space', SPACE_ID, 'sessions'])).toBe(1); + }); + it('still refreshes each skill a burst touched, once each', async () => { deliver([ ...Array.from({ length: BURST }, () => event('entity.run.updated', { workflowSlug: 'a' })), diff --git a/packages/web-product/src/ui/hooks/use-session-metadata.ts b/packages/web-product/src/ui/hooks/use-session-metadata.ts index b5b61ede..fe866377 100644 --- a/packages/web-product/src/ui/hooks/use-session-metadata.ts +++ b/packages/web-product/src/ui/hooks/use-session-metadata.ts @@ -9,19 +9,26 @@ import type { SessionMetadata } from '../lib/types.js'; export { UNTITLED_CONVERSATION, conversationTitle } from '../lib/conversation-title.js'; +/** The space signals that change what a conversation list shows. */ +const SESSION_LIST_SIGNALS: ReadonlySet = new Set([ + 'entity.session.described', + 'entity.session.status', +]); + /** - * Refresh conversation names when the background job finishes one. + * Refresh the conversation list when a conversation is named, or when its turn + * begins or rests. * * Listens on the space channel rather than the open session's own stream: the - * conversation being named is usually not the one anyone has open, and a row + * conversation that changed is usually not the one anyone has open, and a row * in a list nobody subscribed to would otherwise stay stale until the next * refetch. One subscription for the whole list, through the refcounted broker, - * and one refetch for a burst of names. + * and one refetch for a burst of signals. */ -export function useSessionMetadataRefresh(spaceId: string | null): void { +export function useSessionListRefresh(spaceId: string | null): void { const invalidate = useCoalescedInvalidation(); useSpaceEntityEventListener(spaceId, (event) => { - if (!spaceId || event.eventType !== 'entity.session.described') return; + if (!spaceId || !SESSION_LIST_SIGNALS.has(event.eventType)) return; invalidate(['space', spaceId, 'sessions']); }); }