From d69d6341ef3ac00c96d97f80040c9e121415b369 Mon Sep 17 00:00:00 2001 From: karim Date: Fri, 16 Oct 2026 18:13:11 +0200 Subject: [PATCH] feat(web-product): live state on the workbench rows, the run surface and the harness card (F75, F95, F97) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit F75 — a workbench conversation row carries its live state: the agent's turn from the session hot state and the runs the conversation waits on with their phase, through the existing useApiQuery and entity-event broker paths on the ['space', spaceId, …] keys, never a per-row fetch. F95 — the run surface's harness task shows the same live card the chat shows — last tool line, elapsed time, turns — from the step's activity stream, one component mounted in both places, not a second rendering. F97 — the harness activity card leads with the agent's last status or plain-text message and the time since its last activity, then the files it changed, with the tool lines collapsed behind a disclosure; exit codes are shown only where the harness reports them truthfully (F84 is open, so a command's non-zero exit is not shown as failure). Tests state each property: a row with a running commission renders its phase; the run surface mounts the card for a harness task carrying activity; the card's lead line is the last status or message and the tool lines are collapsed by default. Any extension of the activity event shape is in the Zod schema first and derived from there. --- .../src/__tests__/harnessEvents.test.ts | 91 +++++++ apps/aflow-executor-host/src/harnessEvents.ts | 69 ++++- .../src/services/ProjectionWorker.ts | 71 ++++- .../__tests__/projectionSessionStatus.test.ts | 174 ++++++++++++ docs/plans/315-local-first-run-ergonomics.md | 6 +- .../src/metricsAggregator.ts | 3 + .../src/workflowRunProgress.ts | 14 +- .../src/workflowTaskProgressConsumer.ts | 32 ++- packages/redis/src/hotState/events.ts | 22 +- packages/redis/src/hotState/liveStream.ts | 18 +- .../schemas/src/__tests__/cybernetic.test.ts | 5 +- packages/schemas/src/runtime/entityEvents.ts | 7 + .../src/runtime/harnessActivity.test.ts | 25 ++ .../schemas/src/runtime/harnessActivity.ts | 33 ++- .../schemas/src/runtime/streamMessages.ts | 18 ++ .../src/bootstrap/hostLaneFeed.test.ts | 19 +- .../__tests__/sessionListHotStatus.test.ts | 73 +++++ .../__tests__/sessionLiveDeltas.test.ts | 41 +++ .../services/__tests__/sessionWakeup.test.ts | 34 +++ .../server-runtime/src/services/pubsub.ts | 5 + .../src/services/sessionLiveDeltas.ts | 48 +++- .../src/services/sessionTail.ts | 3 + .../src/services/sessionWakeup.ts | 40 ++- .../server-runtime/src/services/sessions.ts | 50 +++- .../ui/components/harness-activity-card.css | 69 +++++ .../harness-activity-card.render.test.ts | 169 ++++++++++++ .../components/harness-activity-card.test.ts | 101 +++++++ .../ui/components/harness-activity-card.tsx | 252 ++++++++++++++++-- .../components/run-timeline/StepGroupCard.tsx | 10 +- .../src/ui/components/workbench/Workbench.tsx | 10 + .../WorkbenchConversationRow.test.ts | 121 +++++++++ .../workbench/WorkbenchConversationRow.tsx | 46 ++++ .../WorkbenchRecentConversations.tsx | 34 ++- .../workbench/conversationLiveState.test.ts | 102 +++++++ .../workbench/conversationLiveState.ts | 104 ++++++++ .../TaskExecutionDetails.tsx | 21 +- .../WorkflowRunSurfaceContainer.tsx | 53 +++- .../hooks/use-coalesced-invalidation.test.ts | 18 +- .../src/ui/hooks/use-session-metadata.ts | 17 +- 39 files changed, 1903 insertions(+), 125 deletions(-) create mode 100644 apps/aflow-orchestrator/src/services/__tests__/projectionSessionStatus.test.ts create mode 100644 packages/server-runtime/src/services/__tests__/sessionListHotStatus.test.ts create mode 100644 packages/web-product/src/ui/components/harness-activity-card.render.test.ts create mode 100644 packages/web-product/src/ui/components/workbench/WorkbenchConversationRow.test.ts create mode 100644 packages/web-product/src/ui/components/workbench/conversationLiveState.test.ts create mode 100644 packages/web-product/src/ui/components/workbench/conversationLiveState.ts 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']); }); }