Show live task progress while agents work (#3227)

* feat(agents): surface live task progress

Keep current task state above the composer and turn task snapshots into semantic timeline activity. Normalize provider task protocols while preserving genuine plan flows.

* fix(agents): admit tasks through stream sequencing

* test(agents): expect normalized task status
This commit is contained in:
Mohamed Boudra
2026-08-11 18:23:30 +02:00
committed by GitHub
parent b2ff359b60
commit db6520bf36
37 changed files with 1342 additions and 60 deletions
@@ -3557,6 +3557,59 @@ describe("processAgentStreamEvent", () => {
expect(result.sideEffects).toEqual([]);
});
it("publishes task snapshots only from accepted timeline events", () => {
const tasks = [
{ id: "inspect", text: "Inspect", status: "in_progress" as const, completed: false },
];
const event: AgentStreamEventPayload = {
type: "timeline",
provider: "codex",
item: {
type: "todo",
items: tasks,
},
};
const currentCursor: TimelineCursor = {
epoch: "epoch-1",
startSeq: 1,
endSeq: 4,
};
const accepted = processAgentStreamEvent({
...baseStreamInput,
event,
seq: 5,
epoch: "epoch-1",
currentCursor,
});
const stale = processAgentStreamEvent({
...baseStreamInput,
event,
seq: 4,
epoch: "epoch-1",
currentCursor,
});
const wrongEpoch = processAgentStreamEvent({
...baseStreamInput,
event,
seq: 5,
epoch: "epoch-2",
currentCursor,
});
const gapped = processAgentStreamEvent({
...baseStreamInput,
event,
seq: 8,
epoch: "epoch-1",
currentCursor,
});
expect(accepted.taskSnapshot).toEqual(tasks);
expect(stale.taskSnapshot).toBeUndefined();
expect(wrongEpoch.taskSnapshot).toBeUndefined();
expect(gapped.taskSnapshot).toBeUndefined();
});
it("detects gap and emits catch-up side effect", () => {
const existingCursor: TimelineCursor = {
epoch: "epoch-1",
@@ -1,6 +1,6 @@
import type { AgentStreamEventPayload } from "@getpaseo/protocol/messages";
import { selectAgentTimelineState, useSessionStore } from "@/stores/session-store";
import type { AssistantMessageItem, StreamItem } from "@/types/stream";
import type { AssistantMessageItem, StreamItem, TodoEntry } from "@/types/stream";
import type { TurnLivenessTransition } from "@/timeline/turn-liveness";
import {
applyStreamEvent,
@@ -1436,6 +1436,7 @@ export interface ProcessAgentStreamEventOutput {
cursor: TimelineCursor | null;
cursorChanged: boolean;
acknowledgedClientMessageIds: string[];
taskSnapshot?: TodoEntry[];
sideEffects: AgentStreamReducerSideEffect[];
}
@@ -1642,6 +1643,9 @@ export function processAgentStreamEvent(
cursor: sequencing.nextTimelineCursor,
cursorChanged: sequencing.cursorChanged,
acknowledgedClientMessageIds: streamResult.acknowledgedClientMessageIds ?? [],
...(sequencing.shouldApplyStreamEvent && event.type === "timeline" && event.item.type === "todo"
? { taskSnapshot: event.item.items }
: {}),
sideEffects: sequencing.sideEffects,
};
}
@@ -1667,6 +1671,7 @@ export function processAgentStreamEvents(
let changedTail = false;
let changedHead = false;
let cursorChanged = false;
let taskSnapshot: TodoEntry[] | undefined;
const acknowledgedClientMessageIds = new Set<string>();
const sideEffects: AgentStreamReducerSideEffect[] = [];
@@ -1690,6 +1695,9 @@ export function processAgentStreamEvents(
for (const clientMessageId of result.acknowledgedClientMessageIds) {
acknowledgedClientMessageIds.add(clientMessageId);
}
if (result.taskSnapshot !== undefined) {
taskSnapshot = result.taskSnapshot;
}
if (result.cursorChanged) {
cursor = result.cursor ?? undefined;
@@ -1705,6 +1713,7 @@ export function processAgentStreamEvents(
cursor: cursor ?? null,
cursorChanged,
acknowledgedClientMessageIds: [...acknowledgedClientMessageIds],
...(taskSnapshot !== undefined ? { taskSnapshot } : {}),
sideEffects,
};
}
@@ -1788,6 +1797,7 @@ interface StreamStatePatch {
tail?: StreamItem[];
head?: StreamItem[];
acknowledgedClientMessageIds?: readonly string[];
taskSnapshot?: TodoEntry[];
}
export function deriveAgentStreamTurnLiveness(
@@ -1850,7 +1860,8 @@ export function createSessionAgentStreamReducerQueue(
if (
result.changedTail ||
result.changedHead ||
result.acknowledgedClientMessageIds.length > 0
result.acknowledgedClientMessageIds.length > 0 ||
result.taskSnapshot !== undefined
) {
setAgentStreamState(serverId, agentId, {
...(result.changedTail ? { tail: result.tail } : {}),
@@ -1858,6 +1869,7 @@ export function createSessionAgentStreamReducerQueue(
...(result.acknowledgedClientMessageIds.length > 0
? { acknowledgedClientMessageIds: result.acknowledgedClientMessageIds }
: {}),
...(result.taskSnapshot !== undefined ? { taskSnapshot: result.taskSnapshot } : {}),
});
}
if (result.cursorChanged && result.cursor) {