puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-10-30 05:02:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

Phase 0: simplify timeline contract and reconnect behavior

Browse Source
This commit is contained in:
Mohamed Boudra
2026-03-25 20:39:07 +07:00
parent cebde228b1
commit c202bd5ff7
27 changed files with 1331 additions and 1238 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/STORAGE_REVAMP_PLAN.md
+9 -7
View File
@@ -92,9 +92,9 @@ The room has converged on a simplified contract:
- `timeline` event with no `seq` means provisional live-only state and is never returned by catch-up fetches
- committed rows must already be in canonical display/history shape, not raw provider chunk shape
- `fetch-after-seq` returns committed rows only
- subscribe may emit the current provisional head once if a turn is active
- that provisional seed must contain current full provisional items, not replayed deltas
- that provisional seed must arrive before later provisional patches for that subscription
- Phase 0 policy: reconnect does not use a subscribe-time provisional seed
- if a turn is still active after reconnect, the client resumes from future live provisional updates only
- provisional-seed ordering is therefore out of scope for Phase 0 until product explicitly chooses to add that policy
- at most one active provisional assistant stream and one active provisional reasoning stream exist per turn
- tool progress continues to correlate by existing `callId`
- reasoning is live-only for v1 unless product explicitly changes that call later
@@ -170,6 +170,8 @@ It is not:
If emitted, it must arrive before later provisional patch events for that subscription.
Phase 0 chooses the simpler policy: no subscribe-time provisional seed for now.
### Reconnect scenarios
Concrete reconnect example:
@@ -193,7 +195,7 @@ Case B: the assistant message has not finished by the time the client reconnects
2. The client calls `fetch-after-seq(agentId, 120)`.
3. The server may legitimately return nothing new yet, because committed row `121` does not exist yet.
4. Live streaming resumes.
5. A subscribe-time provisional seed may repopulate the current in-progress state if that policy is enabled; otherwise the client continues from fresh live updates only.
5. In Phase 0, the client continues from fresh live updates only.
6. When the assistant message finally completes, it becomes one committed row `121`.
In both cases:
@@ -498,9 +500,9 @@ Dependencies:
Tests required:
- shared message parsing tests for the simplified timeline payloads
- app reducer tests for committed append, provisional merge, and subscribe-time provisional seeding
- session/WebSocket tests for `live -> disconnect -> fetch-after-seq -> subscribe -> provisional seed -> resume`
- explicit subscribe-time provisional-seed ordering test
- app reducer tests for committed append, provisional merge, and reconnect without provisional seeding
- session/WebSocket tests for `live -> disconnect -> fetch-after-seq -> subscribe -> resume`
- explicit no-provisional-seed reconnect test proving that reconnect resumes from committed catch-up plus future live events only
Exit gate:
The client and server can exchange committed and provisional timeline updates without `epoch`, `reset`, or projection metadata, and all timeline catch-up tests pass with the new contract.
packages/app/src/contexts/session-context.tsx
+4 -11
View File
@@ -525,9 +525,8 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
void client
.fetchAgentTimeline(agentId, {
direction: "after",
cursor: { epoch: cursor.epoch, seq: cursor.endSeq },
cursor: { seq: cursor.endSeq },
limit: 0,
projection: "canonical",
})
.catch((error) => {
console.warn("[Session] failed to fetch catch-up timeline on resume", agentId, error);
@@ -753,13 +752,12 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
);
const requestCanonicalCatchUp = useCallback(
(agentId: string, cursor: { epoch: string; endSeq: number }) => {
(agentId: string, cursor: { endSeq: number }) => {
void client
.fetchAgentTimeline(agentId, {
direction: "after",
cursor: { epoch: cursor.epoch, seq: cursor.endSeq },
cursor: { seq: cursor.endSeq },
limit: 0,
projection: "canonical",
})
.catch((error) => {
console.warn("[Session] failed to fetch canonical catch-up timeline", agentId, error);
@@ -862,7 +860,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
}
if (
current &&
current.epoch === result.cursor.epoch &&
current.startSeq === result.cursor.startSeq &&
current.endSeq === result.cursor.endSeq
) {
@@ -967,7 +964,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
const unsubAgentStream = client.on("agent_stream", (message) => {
if (message.type !== "agent_stream") return;
const { agentId, event, timestamp, seq, epoch } = message.payload;
const { agentId, event, timestamp, seq } = message.payload;
const parsedTimestamp = new Date(timestamp);
const streamEvent = event as AgentStreamEventPayload;
if (
@@ -1009,7 +1006,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
const result = processAgentStreamEvent({
event: streamEvent,
seq,
epoch,
currentTail,
currentHead,
currentCursor,
@@ -1033,8 +1029,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
if (
current &&
typeof seq === "number" &&
typeof epoch === "string" &&
current.epoch === epoch &&
seq >= current.startSeq &&
seq <= current.endSeq
) {
@@ -1043,7 +1037,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
}
if (
current &&
current.epoch === nextCursor.epoch &&
current.startSeq === nextCursor.startSeq &&
current.endSeq === nextCursor.endSeq
) {
packages/app/src/contexts/session-stream-reducers.test.ts
+215 -479
View File
@@ -9,13 +9,9 @@ import {
type TimelineCursor,
} from "./session-stream-reducers";
// ---------------------------------------------------------------------------
// Test helpers
// ---------------------------------------------------------------------------
function makeTimelineEntry(seq: number, text: string, type: string = "assistant_message") {
return {
seqStart: seq,
seq,
provider: "claude",
item: { type, text },
timestamp: new Date(1000 + seq).toISOString(),
@@ -33,22 +29,30 @@ function makeTimelineEvent(
} as AgentStreamEventPayload;
}
function makeUserTimelineEvent(text: string): AgentStreamEventPayload {
function makeToolCallEvent(status: "running" | "completed"): AgentStreamEventPayload {
return {
type: "timeline",
provider: "claude",
item: { type: "user_message", text },
} as AgentStreamEventPayload;
item: {
type: "tool_call",
callId: "call-1",
name: "shell",
status,
detail: {
type: "shell",
command: "pwd",
},
error: null,
},
};
}
const baseTimelineInput: ProcessTimelineResponseInput = {
payload: {
agentId: "agent-1",
direction: "after",
reset: false,
epoch: "epoch-1",
startCursor: null,
endCursor: null,
startSeq: null,
endSeq: null,
entries: [],
error: null,
},
@@ -63,7 +67,6 @@ const baseTimelineInput: ProcessTimelineResponseInput = {
const baseStreamInput: ProcessAgentStreamEventInput = {
event: makeTimelineEvent("hello"),
seq: undefined,
epoch: undefined,
currentTail: [],
currentHead: [],
currentCursor: undefined,
@@ -71,10 +74,6 @@ const baseStreamInput: ProcessAgentStreamEventInput = {
timestamp: new Date(2000),
};
// ---------------------------------------------------------------------------
// processTimelineResponse
// ---------------------------------------------------------------------------
describe("processTimelineResponse", () => {
it("returns error path when payload.error is set", () => {
const result = processTimelineResponse({
@@ -93,35 +92,10 @@ describe("processTimelineResponse", () => {
expect(result.tail).toBe(baseTimelineInput.currentTail);
expect(result.head).toBe(baseTimelineInput.currentHead);
expect(result.cursorChanged).toBe(false);
expect(result.sideEffects).toEqual([]);
});
it("returns error with no init resolution when no deferred exists", () => {
const result = processTimelineResponse({
...baseTimelineInput,
isInitializing: true,
hasActiveInitDeferred: false,
payload: {
...baseTimelineInput.payload,
error: "timeout",
},
});
expect(result.error).toBe("timeout");
expect(result.initResolution).toBe(null);
expect(result.clearInitializing).toBe(true);
});
it("replaces tail and clears head when reset=true", () => {
const existingTail: StreamItem[] = [
{
kind: "user_message",
id: "old",
text: "old message",
timestamp: new Date(500),
},
];
const existingHead: StreamItem[] = [
it("replaces tail during bootstrap tail init and schedules committed catch-up", () => {
const provisionalHead: StreamItem[] = [
{
kind: "assistant_message",
id: "head-1",
@@ -132,518 +106,280 @@ describe("processTimelineResponse", () => {
const result = processTimelineResponse({
...baseTimelineInput,
currentTail: existingTail,
currentHead: existingHead,
payload: {
...baseTimelineInput.payload,
reset: true,
startCursor: { seq: 1 },
endCursor: { seq: 3 },
entries: [
makeTimelineEntry(1, "first"),
makeTimelineEntry(2, "second"),
makeTimelineEntry(3, "third"),
],
},
});
expect(result.tail).not.toBe(existingTail);
expect(result.tail.length).toBeGreaterThan(0);
expect(result.head).toEqual([]);
expect(result.cursorChanged).toBe(true);
expect(result.cursor).toEqual({
epoch: "epoch-1",
startSeq: 1,
endSeq: 3,
});
expect(result.error).toBe(null);
expect(result.sideEffects.some((e) => e.type === "flush_pending_updates")).toBe(true);
});
it("sets cursor to null when reset=true but no cursors in payload", () => {
const result = processTimelineResponse({
...baseTimelineInput,
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 5 },
payload: {
...baseTimelineInput.payload,
reset: true,
entries: [],
},
});
expect(result.cursor).toBe(null);
expect(result.cursorChanged).toBe(true);
});
it("performs bootstrap tail init with catch-up side effect", () => {
const result = processTimelineResponse({
...baseTimelineInput,
currentHead: provisionalHead,
isInitializing: true,
hasActiveInitDeferred: true,
initRequestDirection: "tail",
payload: {
...baseTimelineInput.payload,
direction: "tail",
epoch: "epoch-1",
startCursor: { seq: 1 },
endCursor: { seq: 5 },
startSeq: 1,
endSeq: 5,
entries: [makeTimelineEntry(1, "first"), makeTimelineEntry(5, "last")],
},
});
// Bootstrap tail replaces
expect(result.tail.length).toBeGreaterThan(0);
expect(result.head).toEqual([]);
expect(result.cursorChanged).toBe(true);
expect(result.cursor).toEqual({
epoch: "epoch-1",
startSeq: 1,
endSeq: 5,
});
// Should have catch-up side effect
const catchUp = result.sideEffects.find((e) => e.type === "catch_up");
expect(catchUp).toBeDefined();
expect(catchUp!.type === "catch_up" && catchUp!.cursor).toEqual({
epoch: "epoch-1",
endSeq: 5,
const catchUp = result.sideEffects.find((effect) => effect.type === "catch_up");
expect(catchUp).toEqual({
type: "catch_up",
cursor: { endSeq: 5 },
});
});
it("appends incrementally for contiguous seqs", () => {
const existingCursor: TimelineCursor = {
epoch: "epoch-1",
startSeq: 1,
endSeq: 3,
};
const result = processTimelineResponse({
...baseTimelineInput,
currentCursor: existingCursor,
payload: {
...baseTimelineInput.payload,
epoch: "epoch-1",
entries: [makeTimelineEntry(4, "next-1"), makeTimelineEntry(5, "next-2")],
it("prepends older committed history for before pagination", () => {
const currentTail: StreamItem[] = [
{
kind: "assistant_message",
id: "tail-3",
text: "newer",
timestamp: new Date(3000),
},
});
expect(result.tail.length).toBeGreaterThan(0);
expect(result.cursorChanged).toBe(true);
expect(result.cursor).toEqual({
epoch: "epoch-1",
startSeq: 1,
endSeq: 5,
});
expect(result.error).toBe(null);
});
it("detects gap and emits catch-up side effect", () => {
const existingCursor: TimelineCursor = {
epoch: "epoch-1",
startSeq: 1,
endSeq: 3,
};
];
const currentCursor: TimelineCursor = { startSeq: 3, endSeq: 4 };
const result = processTimelineResponse({
...baseTimelineInput,
currentCursor: existingCursor,
payload: {
...baseTimelineInput.payload,
epoch: "epoch-1",
entries: [makeTimelineEntry(10, "far ahead")],
},
});
// Gap should trigger catch-up
const catchUp = result.sideEffects.find((e) => e.type === "catch_up");
expect(catchUp).toBeDefined();
expect(catchUp!.type === "catch_up" && catchUp!.cursor).toEqual({
epoch: "epoch-1",
endSeq: 3,
});
});
it("drops stale entries silently", () => {
const existingCursor: TimelineCursor = {
epoch: "epoch-1",
startSeq: 1,
endSeq: 8,
};
const result = processTimelineResponse({
...baseTimelineInput,
currentCursor: existingCursor,
payload: {
...baseTimelineInput.payload,
epoch: "epoch-1",
entries: [makeTimelineEntry(5, "old"), makeTimelineEntry(7, "also old")],
},
});
// No new items appended (all dropped as stale)
expect(result.tail).toBe(baseTimelineInput.currentTail);
expect(result.cursorChanged).toBe(false);
});
it("drops entries with epoch mismatch", () => {
const existingCursor: TimelineCursor = {
epoch: "epoch-1",
startSeq: 1,
endSeq: 5,
};
const result = processTimelineResponse({
...baseTimelineInput,
currentCursor: existingCursor,
payload: {
...baseTimelineInput.payload,
epoch: "epoch-2",
entries: [makeTimelineEntry(6, "different epoch")],
},
});
expect(result.tail).toBe(baseTimelineInput.currentTail);
expect(result.cursorChanged).toBe(false);
});
it("resolves init when deferred matches direction", () => {
const result = processTimelineResponse({
...baseTimelineInput,
isInitializing: true,
hasActiveInitDeferred: true,
initRequestDirection: "after",
payload: {
...baseTimelineInput.payload,
direction: "after",
entries: [],
},
});
expect(result.initResolution).toBe("resolve");
expect(result.clearInitializing).toBe(true);
});
it("does not resolve init when directions differ (before vs after)", () => {
const result = processTimelineResponse({
...baseTimelineInput,
isInitializing: true,
hasActiveInitDeferred: true,
initRequestDirection: "after",
currentTail,
currentCursor,
payload: {
...baseTimelineInput.payload,
direction: "before",
entries: [],
startSeq: 1,
endSeq: 2,
entries: [
makeTimelineEntry(1, "hello", "user_message"),
makeTimelineEntry(2, "older"),
],
},
});
// "before" direction doesn't match "after" initRequestDirection,
// and "before" is not a bootstrap tail path, so init should NOT resolve
expect(result.initResolution).toBe(null);
expect(result.clearInitializing).toBe(false);
expect(result.cursorChanged).toBe(true);
expect(result.cursor).toEqual({
startSeq: 1,
endSeq: 4,
});
expect(result.tail).toHaveLength(3);
expect(result.tail[0]?.kind).toBe("user_message");
expect(result.tail[1]?.kind).toBe("assistant_message");
expect(result.tail[2]).toBe(currentTail[0]);
});
it("clears initializing even without deferred", () => {
it("replaces stale provisional assistant UI when fetch-after returns committed row 121", () => {
const currentHead: StreamItem[] = [
{
kind: "assistant_message",
id: "head-assistant",
text: "partial",
timestamp: new Date(120000),
},
];
const currentCursor: TimelineCursor = { startSeq: 1, endSeq: 120 };
const result = processTimelineResponse({
...baseTimelineInput,
isInitializing: true,
hasActiveInitDeferred: false,
currentHead,
currentCursor,
payload: {
...baseTimelineInput.payload,
direction: "after",
entries: [],
},
});
expect(result.clearInitializing).toBe(true);
expect(result.initResolution).toBe(null);
});
it("always includes flush_pending_updates side effect on success", () => {
const result = processTimelineResponse({
...baseTimelineInput,
payload: {
...baseTimelineInput.payload,
entries: [],
},
});
expect(result.sideEffects.some((e) => e.type === "flush_pending_updates")).toBe(true);
});
it("initializes cursor when no existing cursor on first entries", () => {
const result = processTimelineResponse({
...baseTimelineInput,
currentCursor: undefined,
payload: {
...baseTimelineInput.payload,
epoch: "epoch-1",
entries: [makeTimelineEntry(1, "first"), makeTimelineEntry(2, "second")],
startSeq: 121,
endSeq: 121,
entries: [makeTimelineEntry(121, "finalized reply")],
},
});
expect(result.head).toEqual([]);
expect(result.cursorChanged).toBe(true);
expect(result.cursor).toEqual({
epoch: "epoch-1",
startSeq: 1,
endSeq: 2,
endSeq: 121,
});
expect(result.tail[result.tail.length - 1]).toMatchObject({
kind: "assistant_message",
text: "finalized reply",
});
});
it("keeps provisional head when reconnect catch-up has no new committed rows yet", () => {
const currentHead: StreamItem[] = [
{
kind: "assistant_message",
id: "head-assistant",
text: "still streaming",
timestamp: new Date(120000),
},
];
const currentCursor: TimelineCursor = { startSeq: 1, endSeq: 120 };
const result = processTimelineResponse({
...baseTimelineInput,
currentHead,
currentCursor,
payload: {
...baseTimelineInput.payload,
direction: "after",
startSeq: null,
endSeq: null,
entries: [],
},
});
expect(result.head).toBe(currentHead);
expect(result.cursorChanged).toBe(false);
expect(result.tail).toBe(baseTimelineInput.currentTail);
});
it("requests catch-up when committed rows arrive with a forward gap", () => {
const currentCursor: TimelineCursor = { startSeq: 1, endSeq: 120 };
const result = processTimelineResponse({
...baseTimelineInput,
currentCursor,
payload: {
...baseTimelineInput.payload,
direction: "after",
startSeq: 125,
endSeq: 125,
entries: [makeTimelineEntry(125, "far ahead")],
},
});
expect(result.cursorChanged).toBe(false);
expect(result.tail).toBe(baseTimelineInput.currentTail);
expect(result.sideEffects).toContainEqual({
type: "catch_up",
cursor: { endSeq: 120 },
});
});
});
// ---------------------------------------------------------------------------
// processAgentStreamEvent
// ---------------------------------------------------------------------------
describe("processAgentStreamEvent", () => {
it("passes through non-timeline events without cursor changes", () => {
const turnEvent: AgentStreamEventPayload = {
type: "turn_completed",
provider: "claude",
};
it("treats seq-less timeline events as provisional head updates", () => {
const result = processAgentStreamEvent({
...baseStreamInput,
event: turnEvent,
event: makeTimelineEvent("partial"),
seq: undefined,
epoch: undefined,
});
expect(result.changedHead).toBe(true);
expect(result.changedTail).toBe(false);
expect(result.head).toHaveLength(1);
expect(result.head[0]).toMatchObject({
kind: "assistant_message",
text: "partial",
});
expect(result.cursorChanged).toBe(false);
expect(result.cursor).toBe(null);
expect(result.sideEffects).toEqual([]);
});
it("accepts timeline event with cursor advance", () => {
const existingCursor: TimelineCursor = {
epoch: "epoch-1",
startSeq: 1,
endSeq: 4,
};
it("appends committed live rows to tail and clears superseded provisional assistant state", () => {
const currentHead: StreamItem[] = [
{
kind: "assistant_message",
id: "head-assistant",
text: "partial",
timestamp: new Date(1000),
},
];
const currentCursor: TimelineCursor = { startSeq: 1, endSeq: 120 };
const result = processAgentStreamEvent({
...baseStreamInput,
event: makeTimelineEvent("new chunk"),
seq: 5,
epoch: "epoch-1",
currentCursor: existingCursor,
event: makeTimelineEvent("finalized reply"),
seq: 121,
currentHead,
currentCursor,
});
expect(result.changedTail).toBe(true);
expect(result.changedHead).toBe(true);
expect(result.head).toEqual([]);
expect(result.cursorChanged).toBe(true);
expect(result.cursor).toEqual({
epoch: "epoch-1",
startSeq: 1,
endSeq: 5,
endSeq: 121,
});
expect(result.tail[result.tail.length - 1]).toMatchObject({
kind: "assistant_message",
text: "finalized reply",
});
expect(result.sideEffects).toEqual([]);
});
it("detects gap and emits catch-up side effect", () => {
const existingCursor: TimelineCursor = {
epoch: "epoch-1",
startSeq: 1,
endSeq: 4,
};
it("replaces provisional tool progress when the committed tool row arrives", () => {
const provisional = processAgentStreamEvent({
...baseStreamInput,
event: makeToolCallEvent("running"),
seq: undefined,
});
const committed = processAgentStreamEvent({
...baseStreamInput,
event: makeToolCallEvent("completed"),
seq: 8,
currentHead: provisional.head,
currentTail: provisional.tail,
currentCursor: { startSeq: 1, endSeq: 7 },
});
expect(committed.head).toEqual([]);
expect(committed.tail).toHaveLength(1);
expect(committed.tail[0]).toMatchObject({
kind: "tool_call",
payload: {
source: "agent",
data: {
callId: "call-1",
status: "completed",
},
},
});
});
it("requests catch-up when a committed live row skips ahead", () => {
const result = processAgentStreamEvent({
...baseStreamInput,
event: makeTimelineEvent("far ahead"),
seq: 10,
epoch: "epoch-1",
currentCursor: existingCursor,
seq: 125,
currentCursor: { startSeq: 1, endSeq: 120 },
});
expect(result.cursorChanged).toBe(false);
expect(result.changedTail).toBe(false);
expect(result.changedHead).toBe(false);
const catchUp = result.sideEffects.find((e) => e.type === "catch_up");
expect(catchUp).toBeDefined();
expect(catchUp!.cursor).toEqual({
epoch: "epoch-1",
endSeq: 4,
});
});
it("drops stale timeline event", () => {
const existingCursor: TimelineCursor = {
epoch: "epoch-1",
startSeq: 1,
endSeq: 8,
};
const result = processAgentStreamEvent({
...baseStreamInput,
event: makeTimelineEvent("old"),
seq: 5,
epoch: "epoch-1",
currentCursor: existingCursor,
});
expect(result.cursorChanged).toBe(false);
expect(result.sideEffects).toContainEqual({
type: "catch_up",
cursor: { endSeq: 120 },
});
});
it("clears provisional head on terminal turn events without committing it to tail", () => {
const result = processAgentStreamEvent({
...baseStreamInput,
event: {
type: "turn_completed",
provider: "claude",
},
currentHead: [
{
kind: "thought",
id: "reasoning-1",
text: "thinking",
timestamp: new Date(1000),
status: "loading",
},
],
});
expect(result.changedHead).toBe(true);
expect(result.changedTail).toBe(false);
expect(result.changedHead).toBe(false);
expect(result.sideEffects).toEqual([]);
});
it("drops timeline event with epoch mismatch", () => {
const existingCursor: TimelineCursor = {
epoch: "epoch-1",
startSeq: 1,
endSeq: 5,
};
const result = processAgentStreamEvent({
...baseStreamInput,
event: makeTimelineEvent("wrong epoch"),
seq: 6,
epoch: "epoch-2",
currentCursor: existingCursor,
});
expect(result.cursorChanged).toBe(false);
expect(result.changedTail).toBe(false);
expect(result.changedHead).toBe(false);
expect(result.sideEffects).toEqual([]);
});
it("initializes cursor when none exists", () => {
const result = processAgentStreamEvent({
...baseStreamInput,
event: makeTimelineEvent("first"),
seq: 1,
epoch: "epoch-1",
currentCursor: undefined,
});
expect(result.cursorChanged).toBe(true);
expect(result.cursor).toEqual({
epoch: "epoch-1",
startSeq: 1,
endSeq: 1,
});
});
it("derives optimistic idle status on turn_completed for running agent", () => {
const turnCompletedEvent: AgentStreamEventPayload = {
type: "turn_completed",
provider: "claude",
};
const result = processAgentStreamEvent({
...baseStreamInput,
event: turnCompletedEvent,
currentAgent: {
status: "running",
updatedAt: new Date(1000),
lastActivityAt: new Date(1000),
},
timestamp: new Date(2000),
});
expect(result.agentChanged).toBe(true);
expect(result.agent).not.toBe(null);
expect(result.agent!.status).toBe("idle");
expect(result.agent!.updatedAt.getTime()).toBe(2000);
expect(result.agent!.lastActivityAt.getTime()).toBe(2000);
});
it("derives optimistic error status on turn_failed for running agent", () => {
const turnFailedEvent: AgentStreamEventPayload = {
type: "turn_failed",
provider: "claude",
error: "something broke",
};
const result = processAgentStreamEvent({
...baseStreamInput,
event: turnFailedEvent,
currentAgent: {
status: "running",
updatedAt: new Date(1000),
lastActivityAt: new Date(1000),
},
timestamp: new Date(2000),
});
expect(result.agentChanged).toBe(true);
expect(result.agent!.status).toBe("error");
});
it("does not change agent when status is not running", () => {
const turnCompletedEvent: AgentStreamEventPayload = {
type: "turn_completed",
provider: "claude",
};
const result = processAgentStreamEvent({
...baseStreamInput,
event: turnCompletedEvent,
currentAgent: {
status: "idle",
updatedAt: new Date(1000),
lastActivityAt: new Date(1000),
},
timestamp: new Date(2000),
});
expect(result.agentChanged).toBe(false);
expect(result.agent).toBe(null);
});
it("does not change agent when no agent is provided", () => {
const turnCompletedEvent: AgentStreamEventPayload = {
type: "turn_completed",
provider: "claude",
};
const result = processAgentStreamEvent({
...baseStreamInput,
event: turnCompletedEvent,
currentAgent: null,
timestamp: new Date(2000),
});
expect(result.agentChanged).toBe(false);
expect(result.agent).toBe(null);
});
it("preserves updatedAt when agent timestamp is newer than event", () => {
const turnCompletedEvent: AgentStreamEventPayload = {
type: "turn_completed",
provider: "claude",
};
const result = processAgentStreamEvent({
...baseStreamInput,
event: turnCompletedEvent,
currentAgent: {
status: "running",
updatedAt: new Date(5000),
lastActivityAt: new Date(5000),
},
timestamp: new Date(2000),
});
expect(result.agentChanged).toBe(true);
expect(result.agent!.updatedAt.getTime()).toBe(5000);
expect(result.agent!.lastActivityAt.getTime()).toBe(5000);
});
it("does not produce agent patch for non-terminal events", () => {
const result = processAgentStreamEvent({
...baseStreamInput,
event: makeTimelineEvent("just text"),
currentAgent: {
status: "running",
updatedAt: new Date(1000),
lastActivityAt: new Date(1000),
},
seq: 1,
epoch: "epoch-1",
timestamp: new Date(2000),
});
expect(result.agentChanged).toBe(false);
expect(result.agent).toBe(null);
expect(result.head).toEqual([]);
expect(result.tail).toEqual([]);
});
});
packages/app/src/contexts/session-stream-reducers.ts
+232 -268
View File
@@ -1,7 +1,7 @@
import type { AgentStreamEventPayload } from "@server/shared/messages";
import type { AgentLifecycleStatus } from "@server/shared/agent-lifecycle";
import type { StreamItem } from "@/types/stream";
import { applyStreamEvent, hydrateStreamState, reduceStreamUpdate } from "@/types/stream";
import { hydrateStreamState, reduceStreamUpdate } from "@/types/stream";
import {
classifySessionTimelineSeq,
type SessionTimelineSeqDecision,
@@ -12,38 +12,25 @@ import {
} from "@/contexts/session-timeline-bootstrap-policy";
import { deriveOptimisticLifecycleStatus } from "@/contexts/session-stream-lifecycle";
// ---------------------------------------------------------------------------
// Shared cursor type
// ---------------------------------------------------------------------------
export type TimelineCursor = {
epoch: string;
startSeq: number;
endSeq: number;
};
// ---------------------------------------------------------------------------
// Side-effect discriminated unions
// ---------------------------------------------------------------------------
export type TimelineReducerSideEffect =
| { type: "catch_up"; cursor: { epoch: string; endSeq: number } }
| { type: "catch_up"; cursor: { endSeq: number } }
| { type: "flush_pending_updates" };
export type AgentStreamReducerSideEffect = {
type: "catch_up";
cursor: { epoch: string; endSeq: number };
cursor: { endSeq: number };
};
// ---------------------------------------------------------------------------
// processTimelineResponse
// ---------------------------------------------------------------------------
type TimelineDirection = "tail" | "before" | "after";
type InitRequestDirection = "tail" | "after";
type TimelineResponseEntry = {
seqStart: number;
seq: number;
provider: string;
item: Record<string, unknown>;
timestamp: string;
@@ -53,10 +40,8 @@ export interface ProcessTimelineResponseInput {
payload: {
agentId: string;
direction: TimelineDirection;
reset: boolean;
epoch: string;
startCursor: { seq: number } | null;
endCursor: { seq: number } | null;
startSeq: number | null;
endSeq: number | null;
entries: TimelineResponseEntry[];
error: string | null;
};
@@ -79,205 +64,9 @@ export interface ProcessTimelineResponseOutput {
sideEffects: TimelineReducerSideEffect[];
}
export function processTimelineResponse(
input: ProcessTimelineResponseInput,
): ProcessTimelineResponseOutput {
const {
payload,
currentTail,
currentHead,
currentCursor,
isInitializing,
hasActiveInitDeferred,
initRequestDirection,
} = input;
// ------------------------------------------------------------------
// Error path: reject init and leave stream state unchanged
// ------------------------------------------------------------------
if (payload.error) {
return {
tail: currentTail,
head: currentHead,
cursor: currentCursor,
cursorChanged: false,
initResolution: hasActiveInitDeferred ? "reject" : null,
clearInitializing: isInitializing,
error: payload.error,
sideEffects: [],
};
}
// ------------------------------------------------------------------
// Convert entries to timeline units
// ------------------------------------------------------------------
const timelineUnits = payload.entries.map((entry) => ({
seq: entry.seqStart,
event: {
type: "timeline",
provider: entry.provider,
item: entry.item,
} as AgentStreamEventPayload,
timestamp: new Date(entry.timestamp),
}));
const toHydratedEvents = (
units: typeof timelineUnits,
): Array<{ event: AgentStreamEventPayload; timestamp: Date }> =>
units.map(({ event, timestamp }) => ({ event, timestamp }));
// ------------------------------------------------------------------
// Derive bootstrap policy (replace vs incremental)
// ------------------------------------------------------------------
const bootstrapPolicy = deriveBootstrapTailTimelinePolicy({
direction: payload.direction,
reset: payload.reset,
epoch: payload.epoch,
endCursor: payload.endCursor,
isInitializing,
hasActiveInitDeferred,
});
const replace = bootstrapPolicy.replace;
let nextTail = currentTail;
let nextHead = currentHead;
let nextCursor: TimelineCursor | null | undefined = currentCursor;
let cursorChanged = false;
const sideEffects: TimelineReducerSideEffect[] = [];
if (replace) {
// ----------------------------------------------------------------
// Replace path: full hydration from scratch
// ----------------------------------------------------------------
nextTail = hydrateStreamState(toHydratedEvents(timelineUnits), {
source: "canonical",
});
nextHead = [];
if (payload.startCursor && payload.endCursor) {
nextCursor = {
epoch: payload.epoch,
startSeq: payload.startCursor.seq,
endSeq: payload.endCursor.seq,
};
cursorChanged = true;
} else {
nextCursor = null;
cursorChanged = true;
}
if (bootstrapPolicy.catchUpCursor) {
sideEffects.push({
type: "catch_up",
cursor: bootstrapPolicy.catchUpCursor,
});
}
} else if (timelineUnits.length > 0) {
// ----------------------------------------------------------------
// Incremental append path
// ----------------------------------------------------------------
const acceptedUnits: typeof timelineUnits = [];
let cursor = currentCursor;
let gapCursor: { epoch: string; endSeq: number } | null = null;
for (const unit of timelineUnits) {
const decision: SessionTimelineSeqDecision = classifySessionTimelineSeq({
cursor: cursor ? { epoch: cursor.epoch, endSeq: cursor.endSeq } : null,
epoch: payload.epoch,
seq: unit.seq,
});
if (decision === "gap") {
gapCursor = cursor ? { epoch: cursor.epoch, endSeq: cursor.endSeq } : null;
break;
}
if (decision === "drop_stale" || decision === "drop_epoch") {
continue;
}
acceptedUnits.push(unit);
if (decision === "init") {
cursor = {
epoch: payload.epoch,
startSeq: unit.seq,
endSeq: unit.seq,
};
continue;
}
if (!cursor) {
continue;
}
cursor = {
...cursor,
endSeq: unit.seq,
};
}
if (acceptedUnits.length > 0) {
nextTail = acceptedUnits.reduce<StreamItem[]>(
(state, { event, timestamp }) =>
reduceStreamUpdate(state, event, timestamp, {
source: "canonical",
}),
currentTail,
);
}
if (
cursor &&
(!currentCursor ||
currentCursor.epoch !== cursor.epoch ||
currentCursor.startSeq !== cursor.startSeq ||
currentCursor.endSeq !== cursor.endSeq)
) {
nextCursor = cursor;
cursorChanged = true;
}
if (gapCursor) {
sideEffects.push({ type: "catch_up", cursor: gapCursor });
}
}
// ------------------------------------------------------------------
// Flush pending agent updates side effect
// ------------------------------------------------------------------
sideEffects.push({ type: "flush_pending_updates" });
// ------------------------------------------------------------------
// Init resolution
// ------------------------------------------------------------------
const shouldResolveDeferredInit = shouldResolveTimelineInit({
hasActiveInitDeferred,
isInitializing,
initRequestDirection,
responseDirection: payload.direction,
reset: payload.reset,
});
const clearInitializing = shouldResolveDeferredInit || (isInitializing && !hasActiveInitDeferred);
const initResolution: "resolve" | "reject" | null = shouldResolveDeferredInit ? "resolve" : null;
return {
tail: nextTail,
head: nextHead,
cursor: nextCursor,
cursorChanged,
initResolution,
clearInitializing,
error: null,
sideEffects,
};
}
// ---------------------------------------------------------------------------
// processAgentStreamEvent
// ---------------------------------------------------------------------------
export interface ProcessAgentStreamEventInput {
event: AgentStreamEventPayload;
seq: number | undefined;
epoch: string | undefined;
currentTail: StreamItem[];
currentHead: StreamItem[];
currentCursor: TimelineCursor | undefined;
@@ -307,75 +96,250 @@ export interface ProcessAgentStreamEventOutput {
sideEffects: AgentStreamReducerSideEffect[];
}
function cursorsEqual(
left: TimelineCursor | null | undefined,
right: TimelineCursor | null | undefined,
): boolean {
if (!left || !right) {
return left === right;
}
return left.startSeq === right.startSeq && left.endSeq === right.endSeq;
}
function removeSupersededProvisionalItems(
head: StreamItem[],
event: AgentStreamEventPayload,
): StreamItem[] {
if (head.length === 0 || event.type !== "timeline") {
return head;
}
let nextHead = head;
if (event.item.type === "assistant_message") {
nextHead = head.filter((item) => item.kind !== "assistant_message");
} else if (event.item.type === "tool_call") {
const committedToolCall = event.item;
nextHead = head.filter(
(item) =>
item.kind !== "tool_call" ||
item.payload.source !== "agent" ||
item.payload.data.callId !== committedToolCall.callId,
);
}
return nextHead.length === head.length ? head : nextHead;
}
export function processTimelineResponse(
input: ProcessTimelineResponseInput,
): ProcessTimelineResponseOutput {
const {
payload,
currentTail,
currentHead,
currentCursor,
isInitializing,
hasActiveInitDeferred,
initRequestDirection,
} = input;
if (payload.error) {
return {
tail: currentTail,
head: currentHead,
cursor: currentCursor,
cursorChanged: false,
initResolution: hasActiveInitDeferred ? "reject" : null,
clearInitializing: isInitializing,
error: payload.error,
sideEffects: [],
};
}
const timelineUnits = payload.entries.map((entry) => ({
seq: entry.seq,
event: {
type: "timeline",
provider: entry.provider,
item: entry.item,
} as AgentStreamEventPayload,
timestamp: new Date(entry.timestamp),
}));
const bootstrapPolicy = deriveBootstrapTailTimelinePolicy({
direction: payload.direction,
endSeq: payload.endSeq,
isInitializing,
hasActiveInitDeferred,
});
let nextTail = currentTail;
let nextHead = currentHead;
let nextCursor: TimelineCursor | null | undefined = currentCursor;
let cursorChanged = false;
const sideEffects: TimelineReducerSideEffect[] = [];
if (bootstrapPolicy.replace) {
nextTail = hydrateStreamState(
timelineUnits.map(({ event, timestamp }) => ({ event, timestamp })),
{ source: "canonical" },
);
nextHead = [];
nextCursor =
typeof payload.startSeq === "number" && typeof payload.endSeq === "number"
? {
startSeq: payload.startSeq,
endSeq: payload.endSeq,
}
: null;
cursorChanged = !cursorsEqual(currentCursor, nextCursor);
if (bootstrapPolicy.catchUpCursor) {
sideEffects.push({
type: "catch_up",
cursor: bootstrapPolicy.catchUpCursor,
});
}
} else if (payload.direction === "before") {
const prepended = hydrateStreamState(
timelineUnits.map(({ event, timestamp }) => ({ event, timestamp })),
{ source: "canonical" },
);
nextTail = prepended.length > 0 ? [...prepended, ...currentTail] : currentTail;
const derivedCursor =
typeof payload.startSeq === "number"
? {
startSeq: payload.startSeq,
endSeq: currentCursor?.endSeq ?? payload.endSeq ?? payload.startSeq,
}
: currentCursor;
nextCursor = derivedCursor;
cursorChanged = !cursorsEqual(currentCursor, derivedCursor);
} else if (timelineUnits.length > 0) {
const acceptedUnits: typeof timelineUnits = [];
let cursor = currentCursor;
let gapCursor: { endSeq: number } | null = null;
for (const unit of timelineUnits) {
const decision: SessionTimelineSeqDecision = classifySessionTimelineSeq({
cursor: cursor ? { endSeq: cursor.endSeq } : null,
seq: unit.seq,
});
if (decision === "gap") {
gapCursor = cursor ? { endSeq: cursor.endSeq } : null;
break;
}
if (decision === "drop_stale") {
continue;
}
acceptedUnits.push(unit);
cursor =
decision === "init"
? { startSeq: unit.seq, endSeq: unit.seq }
: { ...(cursor ?? { startSeq: unit.seq, endSeq: unit.seq }), endSeq: unit.seq };
nextHead = removeSupersededProvisionalItems(nextHead, unit.event);
}
if (acceptedUnits.length > 0) {
nextTail = acceptedUnits.reduce<StreamItem[]>(
(state, { event, timestamp }) =>
reduceStreamUpdate(state, event, timestamp, {
source: "canonical",
}),
currentTail,
);
}
if (cursor && !cursorsEqual(currentCursor, cursor)) {
nextCursor = cursor;
cursorChanged = true;
}
if (gapCursor) {
sideEffects.push({ type: "catch_up", cursor: gapCursor });
}
}
sideEffects.push({ type: "flush_pending_updates" });
const shouldResolveDeferredInit = shouldResolveTimelineInit({
hasActiveInitDeferred,
isInitializing,
initRequestDirection,
responseDirection: payload.direction,
});
const clearInitializing = shouldResolveDeferredInit || (isInitializing && !hasActiveInitDeferred);
return {
tail: nextTail,
head: nextHead,
cursor: nextCursor,
cursorChanged,
initResolution: shouldResolveDeferredInit ? "resolve" : null,
clearInitializing,
error: null,
sideEffects,
};
}
export function processAgentStreamEvent(
input: ProcessAgentStreamEventInput,
): ProcessAgentStreamEventOutput {
const { event, seq, epoch, currentTail, currentHead, currentCursor, currentAgent, timestamp } =
input;
const { event, seq, currentTail, currentHead, currentCursor, currentAgent, timestamp } = input;
let shouldApplyStreamEvent = true;
let nextTail = currentTail;
let nextHead = currentHead;
let changedTail = false;
let changedHead = false;
let nextTimelineCursor: TimelineCursor | null = null;
let cursorChanged = false;
const sideEffects: AgentStreamReducerSideEffect[] = [];
// ------------------------------------------------------------------
// Timeline sequencing gate
// ------------------------------------------------------------------
if (event.type === "timeline" && typeof seq === "number" && typeof epoch === "string") {
if (event.type === "timeline" && typeof seq === "number") {
const decision = classifySessionTimelineSeq({
cursor: currentCursor ? { epoch: currentCursor.epoch, endSeq: currentCursor.endSeq } : null,
epoch,
cursor: currentCursor ? { endSeq: currentCursor.endSeq } : null,
seq,
});
if (decision === "init") {
nextTimelineCursor = { epoch, startSeq: seq, endSeq: seq };
cursorChanged = true;
} else if (decision === "accept") {
nextTimelineCursor = {
...(currentCursor ?? { epoch, startSeq: seq, endSeq: seq }),
epoch,
endSeq: seq,
};
cursorChanged = true;
} else if (decision === "gap") {
shouldApplyStreamEvent = false;
if (decision === "gap") {
if (currentCursor) {
sideEffects.push({
type: "catch_up",
cursor: {
epoch: currentCursor.epoch,
endSeq: currentCursor.endSeq,
},
cursor: { endSeq: currentCursor.endSeq },
});
}
} else {
// drop_stale or drop_epoch
shouldApplyStreamEvent = false;
} else if (decision !== "drop_stale") {
nextTail = reduceStreamUpdate(currentTail, event, timestamp, {
source: "canonical",
});
changedTail = nextTail !== currentTail;
nextHead = removeSupersededProvisionalItems(currentHead, event);
changedHead = nextHead !== currentHead;
nextTimelineCursor =
decision === "init"
? { startSeq: seq, endSeq: seq }
: { ...(currentCursor ?? { startSeq: seq, endSeq: seq }), endSeq: seq };
cursorChanged = !cursorsEqual(currentCursor, nextTimelineCursor);
}
} else if (event.type === "timeline") {
nextHead = reduceStreamUpdate(currentHead, event, timestamp, {
source: "live",
});
changedHead = nextHead !== currentHead;
} else if (
(event.type === "turn_completed" ||
event.type === "turn_canceled" ||
event.type === "turn_failed") &&
currentHead.length > 0
) {
nextHead = [];
changedHead = true;
}
// ------------------------------------------------------------------
// Apply stream event to tail/head
// ------------------------------------------------------------------
const { tail, head, changedTail, changedHead } = shouldApplyStreamEvent
? applyStreamEvent({
tail: currentTail,
head: currentHead,
event,
timestamp,
source: "live",
})
: {
tail: currentTail,
head: currentHead,
changedTail: false,
changedHead: false,
};
// ------------------------------------------------------------------
// Optimistic lifecycle status
// ------------------------------------------------------------------
let agentPatch: AgentPatch | null = null;
let agentChanged = false;
@@ -402,8 +366,8 @@ export function processAgentStreamEvent(
}
return {
tail,
head,
tail: nextTail,
head: nextHead,
changedTail,
changedHead,
cursor: nextTimelineCursor,
packages/app/src/contexts/session-timeline-bootstrap-policy.test.ts
+31 -24
View File
@@ -2,26 +2,42 @@ import { describe, expect, it } from "vitest";
import { classifySessionTimelineSeq } from "./session-timeline-seq-gate";
import {
deriveBootstrapTailTimelinePolicy,
deriveInitialTimelineRequest,
shouldResolveTimelineInit,
} from "./session-timeline-bootstrap-policy";
describe("deriveBootstrapTailTimelinePolicy", () => {
it("always replaces on explicit reset without catch-up cursor", () => {
const policy = deriveBootstrapTailTimelinePolicy({
direction: "after",
reset: true,
epoch: "epoch-1",
endCursor: { seq: 200 },
isInitializing: false,
hasActiveInitDeferred: false,
describe("deriveInitialTimelineRequest", () => {
it("uses tail bootstrap when history has not synced yet", () => {
expect(
deriveInitialTimelineRequest({
cursor: { seq: 42 },
hasAuthoritativeHistory: false,
initialTimelineLimit: 200,
}),
).toEqual({
direction: "tail",
limit: 200,
});
expect(policy.replace).toBe(true);
expect(policy.catchUpCursor).toBeNull();
});
it("uses catch-up after the committed cursor once history is synced", () => {
expect(
deriveInitialTimelineRequest({
cursor: { seq: 42 },
hasAuthoritativeHistory: true,
initialTimelineLimit: 200,
}),
).toEqual({
direction: "after",
cursor: { seq: 42 },
limit: 0,
});
});
});
describe("deriveBootstrapTailTimelinePolicy", () => {
it("forces baseline replace and canonical catch-up for init tail race", () => {
const advancedCursor = { epoch: "epoch-1", endSeq: 205 };
const advancedCursor = { endSeq: 205 };
const tailSeqStart = 101;
const tailSeqEnd = 200;
@@ -29,7 +45,6 @@ describe("deriveBootstrapTailTimelinePolicy", () => {
for (let seq = tailSeqStart; seq <= tailSeqEnd; seq += 1) {
const decision = classifySessionTimelineSeq({
cursor: advancedCursor,
epoch: "epoch-1",
seq,
});
if (decision === "accept" || decision === "init") {
@@ -40,16 +55,13 @@ describe("deriveBootstrapTailTimelinePolicy", () => {
const policy = deriveBootstrapTailTimelinePolicy({
direction: "tail",
reset: false,
epoch: "epoch-1",
endCursor: { seq: 200 },
endSeq: 200,
isInitializing: true,
hasActiveInitDeferred: true,
});
expect(policy.replace).toBe(true);
expect(policy.catchUpCursor).toEqual({
epoch: "epoch-1",
endSeq: 200,
});
});
@@ -57,9 +69,7 @@ describe("deriveBootstrapTailTimelinePolicy", () => {
it("does not replace non-bootstrap, non-reset responses", () => {
const policy = deriveBootstrapTailTimelinePolicy({
direction: "tail",
reset: false,
epoch: "epoch-1",
endCursor: { seq: 200 },
endSeq: 200,
isInitializing: false,
hasActiveInitDeferred: false,
});
@@ -77,7 +87,6 @@ describe("shouldResolveTimelineInit", () => {
isInitializing: true,
initRequestDirection: "tail",
responseDirection: "tail",
reset: false,
}),
).toBe(true);
});
@@ -89,7 +98,6 @@ describe("shouldResolveTimelineInit", () => {
isInitializing: true,
initRequestDirection: "tail",
responseDirection: "after",
reset: false,
}),
).toBe(false);
});
@@ -101,7 +109,6 @@ describe("shouldResolveTimelineInit", () => {
isInitializing: true,
initRequestDirection: "after",
responseDirection: "after",
reset: false,
}),
).toBe(true);
});
packages/app/src/contexts/session-timeline-bootstrap-policy.ts
+6 -23
View File
@@ -6,7 +6,6 @@ type BootstrapTailCursor = {
} | null;
type InitialTimelineCursor = {
epoch: string;
seq: number;
} | null;
@@ -20,48 +19,37 @@ export function deriveInitialTimelineRequest({
initialTimelineLimit: number;
}): {
direction: "tail" | "after";
cursor?: { epoch: string; seq: number };
cursor?: { seq: number };
limit: number;
projection: "canonical";
} {
if (!hasAuthoritativeHistory || !cursor) {
return {
direction: "tail",
limit: initialTimelineLimit,
projection: "canonical",
};
}
return {
direction: "after",
cursor: { epoch: cursor.epoch, seq: cursor.seq },
cursor: { seq: cursor.seq },
limit: 0,
projection: "canonical",
};
}
export function deriveBootstrapTailTimelinePolicy({
direction,
reset,
epoch,
endCursor,
endSeq,
isInitializing,
hasActiveInitDeferred,
}: {
direction: TimelineDirection;
reset: boolean;
epoch: string;
endCursor: BootstrapTailCursor;
endSeq: number | null;
isInitializing: boolean;
hasActiveInitDeferred: boolean;
}): {
replace: boolean;
catchUpCursor: { epoch: string; endSeq: number } | null;
catchUpCursor: { endSeq: number } | null;
} {
if (reset) {
return { replace: true, catchUpCursor: null };
}
const isBootstrapTailInit = direction === "tail" && isInitializing && hasActiveInitDeferred;
if (!isBootstrapTailInit) {
return { replace: false, catchUpCursor: null };
@@ -69,7 +57,7 @@ export function deriveBootstrapTailTimelinePolicy({
return {
replace: true,
catchUpCursor: endCursor ? { epoch, endSeq: endCursor.seq } : null,
catchUpCursor: typeof endSeq === "number" ? { endSeq } : null,
};
}
@@ -78,19 +66,14 @@ export function shouldResolveTimelineInit({
isInitializing,
initRequestDirection,
responseDirection,
reset,
}: {
hasActiveInitDeferred: boolean;
isInitializing: boolean;
initRequestDirection: InitRequestDirection;
responseDirection: TimelineDirection;
reset: boolean;
}): boolean {
if (!hasActiveInitDeferred || !isInitializing) {
return false;
}
if (reset) {
return true;
}
return responseDirection === initRequestDirection;
}
packages/app/src/contexts/session-timeline-seq-gate.test.ts
+4 -19
View File
@@ -5,8 +5,7 @@ describe("classifySessionTimelineSeq", () => {
it("accepts contiguous forward seq", () => {
expect(
classifySessionTimelineSeq({
cursor: { epoch: "epoch-1", endSeq: 4 },
epoch: "epoch-1",
cursor: { endSeq: 4 },
seq: 5,
}),
).toBe("accept");
@@ -15,8 +14,7 @@ describe("classifySessionTimelineSeq", () => {
it("drops stale seq older than the current end", () => {
expect(
classifySessionTimelineSeq({
cursor: { epoch: "epoch-1", endSeq: 8 },
epoch: "epoch-1",
cursor: { endSeq: 8 },
seq: 7,
}),
).toBe("drop_stale");
@@ -25,28 +23,16 @@ describe("classifySessionTimelineSeq", () => {
it("drops duplicate replay seq equal to the current end", () => {
expect(
classifySessionTimelineSeq({
cursor: { epoch: "epoch-1", endSeq: 8 },
epoch: "epoch-1",
cursor: { endSeq: 8 },
seq: 8,
}),
).toBe("drop_stale");
});
it("drops epoch mismatch", () => {
expect(
classifySessionTimelineSeq({
cursor: { epoch: "epoch-1", endSeq: 4 },
epoch: "epoch-2",
seq: 5,
}),
).toBe("drop_epoch");
});
it("initializes when cursor is null", () => {
expect(
classifySessionTimelineSeq({
cursor: null,
epoch: "epoch-1",
seq: 1,
}),
).toBe("init");
@@ -55,8 +41,7 @@ describe("classifySessionTimelineSeq", () => {
it("classifies forward gaps", () => {
expect(
classifySessionTimelineSeq({
cursor: { epoch: "epoch-1", endSeq: 4 },
epoch: "epoch-1",
cursor: { endSeq: 4 },
seq: 9,
}),
).toBe("gap");
packages/app/src/contexts/session-timeline-seq-gate.ts
+1 -7
View File
@@ -1,28 +1,22 @@
export type SessionTimelineSeqCursor =
| {
epoch: string;
endSeq: number;
}
| null
| undefined;
export type SessionTimelineSeqDecision = "accept" | "drop_stale" | "drop_epoch" | "gap" | "init";
export type SessionTimelineSeqDecision = "accept" | "drop_stale" | "gap" | "init";
export function classifySessionTimelineSeq({
cursor,
epoch,
seq,
}: {
cursor: SessionTimelineSeqCursor;
epoch: string;
seq: number;
}): SessionTimelineSeqDecision {
if (!cursor) {
return "init";
}
if (cursor.epoch !== epoch) {
return "drop_epoch";
}
if (seq <= cursor.endSeq) {
return "drop_stale";
}
packages/app/src/hooks/use-agent-initialization.test.ts
+4 -10
View File
@@ -2,11 +2,10 @@ import { describe, expect, it } from "vitest";
import { __private__ } from "./use-agent-initialization";
describe("useAgentInitialization timeline request policy", () => {
it("uses canonical tail bootstrap when history has not synced yet", () => {
it("uses committed tail bootstrap when history has not synced yet", () => {
expect(
__private__.deriveInitialTimelineRequest({
cursor: {
epoch: "epoch-1",
seq: 42,
},
hasAuthoritativeHistory: false,
@@ -15,11 +14,10 @@ describe("useAgentInitialization timeline request policy", () => {
).toEqual({
direction: "tail",
limit: 200,
projection: "canonical",
});
});
it("uses canonical tail bootstrap when cursor is missing", () => {
it("uses committed tail bootstrap when cursor is missing", () => {
expect(
__private__.deriveInitialTimelineRequest({
cursor: null,
@@ -29,15 +27,13 @@ describe("useAgentInitialization timeline request policy", () => {
).toEqual({
direction: "tail",
limit: 200,
projection: "canonical",
});
});
it("uses canonical catch-up after the current cursor once history is synced", () => {
it("uses committed catch-up after the current cursor once history is synced", () => {
expect(
__private__.deriveInitialTimelineRequest({
cursor: {
epoch: "epoch-1",
seq: 42,
},
hasAuthoritativeHistory: true,
@@ -45,9 +41,8 @@ describe("useAgentInitialization timeline request policy", () => {
}),
).toEqual({
direction: "after",
cursor: { epoch: "epoch-1", seq: 42 },
cursor: { seq: 42 },
limit: 0,
projection: "canonical",
});
});
@@ -61,7 +56,6 @@ describe("useAgentInitialization timeline request policy", () => {
).toEqual({
direction: "tail",
limit: 0,
projection: "canonical",
});
});
packages/app/src/hooks/use-agent-initialization.ts
+1 -2
View File
@@ -60,7 +60,7 @@ export function useAgentInitialization({
const hasAuthoritativeHistory =
session?.agentAuthoritativeHistoryApplied.get(agentId) === true;
const timelineRequest = deriveInitialTimelineRequest({
cursor: cursor ? { epoch: cursor.epoch, seq: cursor.endSeq } : null,
cursor: cursor ? { seq: cursor.endSeq } : null,
hasAuthoritativeHistory,
initialTimelineLimit,
});
@@ -107,7 +107,6 @@ export function useAgentInitialization({
await client.fetchAgentTimeline(agentId, {
direction: "tail",
limit: initialTimelineLimit,
projection: "canonical",
});
} catch (error) {
setAgentInitializing(agentId, false);
packages/app/src/stores/session-store.ts
-1
View File
@@ -191,7 +191,6 @@ export type DaemonServerInfo = {
};
export interface AgentTimelineCursorState {
epoch: string;
startSeq: number;
endSeq: number;
}
packages/server/src/client/daemon-client.test.ts
+6 -24
View File
@@ -1704,24 +1704,15 @@ describe("DaemonClient", () => {
agentId: "agent_cli",
agent: null,
direction: "tail",
projection: "projected",
epoch: "epoch-1",
reset: false,
staleCursor: false,
gap: false,
window: { minSeq: 1, maxSeq: 1, nextSeq: 2 },
startCursor: { epoch: "epoch-1", seq: 1 },
endCursor: { epoch: "epoch-1", seq: 1 },
startSeq: 1,
endSeq: 1,
hasOlder: false,
hasNewer: false,
entries: [
{
timestamp: "2026-02-08T20:20:00.000Z",
provider: "codex",
seqStart: 1,
seqEnd: 1,
sourceSeqRanges: [{ startSeq: 1, endSeq: 1 }],
collapsed: [],
seq: 1,
item: {
type: "tool_call",
callId: "call_cli_snapshot",
@@ -1798,24 +1789,15 @@ describe("DaemonClient", () => {
agentId: "agent_cli",
agent: null,
direction: "tail",
projection: "projected",
epoch: "epoch-1",
reset: false,
staleCursor: false,
gap: false,
window: { minSeq: 1, maxSeq: 1, nextSeq: 2 },
startCursor: { epoch: "epoch-1", seq: 1 },
endCursor: { epoch: "epoch-1", seq: 1 },
startSeq: 1,
endSeq: 1,
hasOlder: false,
hasNewer: false,
entries: [
{
timestamp: "2026-02-08T20:20:00.000Z",
provider: "codex",
seqStart: 1,
seqEnd: 1,
sourceSeqRanges: [{ startSeq: 1, endSeq: 1 }],
collapsed: [],
seq: 1,
item: {
type: "tool_call",
callId: "call_cli_invalid",
packages/server/src/client/daemon-client.ts
+3 -6
View File
@@ -130,7 +130,6 @@ export type DaemonEvent =
event: AgentStreamEventPayload;
timestamp: string;
seq?: number;
epoch?: string;
}
| { type: "status"; payload: { status: string } & Record<string, unknown> }
| { type: "agent_deleted"; agentId: string }
@@ -244,13 +243,13 @@ type KillTerminalPayload = KillTerminalResponse["payload"];
export type FetchAgentTimelinePayload = FetchAgentTimelineResponseMessage["payload"];
export type FetchAgentTimelineDirection = FetchAgentTimelinePayload["direction"];
export type FetchAgentTimelineProjection = FetchAgentTimelinePayload["projection"];
export type FetchAgentTimelineCursor = NonNullable<FetchAgentTimelinePayload["startCursor"]>;
export type FetchAgentTimelineCursor = NonNullable<
Extract<SessionInboundMessage, { type: "fetch_agent_timeline_request" }>["cursor"]
>;
export type FetchAgentTimelineOptions = {
direction?: FetchAgentTimelineDirection;
cursor?: FetchAgentTimelineCursor;
limit?: number;
projection?: FetchAgentTimelineProjection;
requestId?: string;
};
@@ -1396,7 +1395,6 @@ export class DaemonClient {
...(options.direction ? { direction: options.direction } : {}),
...(options.cursor ? { cursor: options.cursor } : {}),
...(typeof options.limit === "number" ? { limit: options.limit } : {}),
...(options.projection ? { projection: options.projection } : {}),
});
const payload = await this.sendRequest({
@@ -3100,7 +3098,6 @@ export class DaemonClient {
event: msg.payload.event,
timestamp: msg.payload.timestamp,
...(typeof msg.payload.seq === "number" ? { seq: msg.payload.seq } : {}),
...(typeof msg.payload.epoch === "string" ? { epoch: msg.payload.epoch } : {}),
};
case "status":
return { type: "status", payload: msg.payload };
packages/server/src/server/agent/agent-manager.test.ts
+361 -56
View File
@@ -197,6 +197,119 @@ class TestAgentSession implements AgentSession {
async close(): Promise<void> {}
}
class StreamingAssistantSession implements AgentSession {
readonly provider = "codex" as const;
readonly capabilities = TEST_CAPABILITIES;
readonly id = randomUUID();
private subscribers = new Set<(event: AgentStreamEvent) => void>();
private turnIdCounter = 0;
constructor(private readonly config: AgentSessionConfig) {}
async run(): Promise<AgentRunResult> {
return {
sessionId: this.id,
finalText: "",
timeline: [],
};
}
async startTurn(): Promise<{ turnId: string }> {
const turnId = `turn-${++this.turnIdCounter}`;
setTimeout(() => {
this.pushEvent({ type: "turn_started", provider: this.provider, turnId });
this.pushEvent({
type: "timeline",
provider: this.provider,
turnId,
item: { type: "assistant_message", text: "final " },
});
this.pushEvent({
type: "timeline",
provider: this.provider,
turnId,
item: { type: "assistant_message", text: "reply" },
});
this.pushEvent({ type: "turn_completed", provider: this.provider, turnId });
}, 0);
return { turnId };
}
subscribe(callback: (event: AgentStreamEvent) => void): () => void {
this.subscribers.add(callback);
return () => {
this.subscribers.delete(callback);
};
}
private pushEvent(event: AgentStreamEvent): void {
for (const callback of this.subscribers) {
callback(event);
}
}
async *streamHistory(): AsyncGenerator<AgentStreamEvent> {}
async getRuntimeInfo() {
return {
provider: this.provider,
sessionId: this.id,
model: this.config.model ?? null,
modeId: this.config.modeId ?? null,
};
}
async getAvailableModes() {
return [];
}
async getCurrentMode() {
return null;
}
async setMode(): Promise<void> {}
getPendingPermissions() {
return [];
}
async respondToPermission(): Promise<void> {}
describePersistence() {
return {
provider: this.provider,
sessionId: this.id,
};
}
async interrupt(): Promise<void> {}
async close(): Promise<void> {}
}
class StreamingAssistantClient implements AgentClient {
readonly provider = "codex" as const;
readonly capabilities = TEST_CAPABILITIES;
async isAvailable(): Promise<boolean> {
return true;
}
async createSession(config: AgentSessionConfig): Promise<AgentSession> {
return new StreamingAssistantSession(config);
}
async resumeSession(
_handle: AgentPersistenceHandle,
config?: Partial<AgentSessionConfig>,
): Promise<AgentSession> {
return new StreamingAssistantSession({
provider: "codex",
cwd: config?.cwd ?? process.cwd(),
});
}
}
describe("AgentManager", () => {
const logger = createTestLogger();
@@ -784,7 +897,7 @@ describe("AgentManager", () => {
}
});
test("fetchTimeline returns full timeline with reset when cursor epoch is stale", async () => {
test("fetchTimeline returns committed rows after a known seq without reset metadata", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-stale-"));
const storagePath = join(workdir, "agents");
const storage = new AgentStorage(storagePath, logger);
@@ -802,49 +915,51 @@ describe("AgentManager", () => {
cwd: workdir,
});
await manager.appendTimelineItem(snapshot.id, {
type: "assistant_message",
text: "one",
});
await manager.appendTimelineItem(snapshot.id, {
type: "assistant_message",
text: "two",
});
await manager.appendTimelineItem(snapshot.id, {
type: "assistant_message",
text: "three",
});
for (let seq = 1; seq <= 120; seq += 1) {
await manager.appendTimelineItem(snapshot.id, {
type: "assistant_message",
text: `committed row ${seq}`,
});
}
const baseline = manager.fetchTimeline(snapshot.id, {
direction: "tail",
limit: 2,
limit: 0,
});
expect(baseline.rows).toHaveLength(120);
await manager.emitLiveTimelineItem(snapshot.id, {
type: "assistant_message",
text: "partial reply",
});
await manager.appendTimelineItem(snapshot.id, {
type: "assistant_message",
text: "finalized reply",
});
expect(baseline.rows).toHaveLength(2);
const result = manager.fetchTimeline(snapshot.id, {
direction: "after",
cursor: {
epoch: "stale-epoch",
seq: baseline.rows[baseline.rows.length - 1]!.seq,
seq: 120,
},
limit: 1,
limit: 0,
});
expect(result.reset).toBe(true);
expect(result.staleCursor).toBe(true);
expect(result.gap).toBe(false);
expect(result.rows).toHaveLength(3);
expect(result.rows[0]?.seq).toBe(1);
expect(result.rows[result.rows.length - 1]?.seq).toBe(3);
expect(result.rows).toHaveLength(1);
expect(result.rows[0]?.seq).toBe(121);
expect(result.rows[0]?.item).toEqual({
type: "assistant_message",
text: "finalized reply",
});
});
test("emits live timeline updates without recording canonical timeline rows", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-live-timeline-"));
test("streams assistant chunks provisionally and commits one finalized assistant row", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-provisional-timeline-"));
const storagePath = join(workdir, "agents");
const storage = new AgentStorage(storagePath, logger);
const manager = new AgentManager({
clients: {
codex: new TestAgentClient(),
codex: new StreamingAssistantClient(),
},
registry: storage,
logger,
@@ -858,9 +973,9 @@ describe("AgentManager", () => {
const streamEvents: Array<{
seq?: number;
epoch?: string;
eventType?: string;
itemType?: string;
text?: string;
}> = [];
manager.subscribe(
(event) => {
@@ -869,37 +984,61 @@ describe("AgentManager", () => {
}
streamEvents.push({
seq: event.seq,
epoch: event.epoch,
eventType: event.event.type,
itemType: event.event.type === "timeline" ? event.event.item.type : undefined,
text:
event.event.type === "timeline" && event.event.item.type === "assistant_message"
? event.event.item.text
: undefined,
});
},
{ agentId: snapshot.id, replayState: false },
);
await manager.emitLiveTimelineItem(snapshot.id, {
type: "assistant_message",
text: "live-only update",
});
const stream = manager.streamAgent(snapshot.id, "hello");
while (true) {
const next = await stream.next();
if (next.done) {
break;
}
}
expect(streamEvents).toHaveLength(1);
expect(streamEvents[0]).toMatchObject({
const assistantTimelineEvents = streamEvents.filter((event) => event.itemType === "assistant_message");
expect(assistantTimelineEvents).toHaveLength(3);
expect(assistantTimelineEvents[0]).toMatchObject({
eventType: "timeline",
itemType: "assistant_message",
text: "final ",
});
expect(assistantTimelineEvents[0]?.seq).toBeUndefined();
expect(assistantTimelineEvents[1]).toMatchObject({
text: "reply",
});
expect(assistantTimelineEvents[1]?.seq).toBeUndefined();
expect(assistantTimelineEvents[2]).toMatchObject({
text: "final reply",
seq: 1,
});
expect(streamEvents[0]?.seq).toBeUndefined();
expect(streamEvents[0]?.epoch).toBeUndefined();
expect(manager.getTimeline(snapshot.id)).toEqual([]);
expect(manager.getTimeline(snapshot.id)).toEqual([
{
type: "assistant_message",
text: "final reply",
},
]);
const fetched = manager.fetchTimeline(snapshot.id, {
direction: "tail",
limit: 0,
});
expect(fetched.rows).toEqual([]);
expect(fetched.rows).toHaveLength(1);
expect(fetched.rows[0]?.item).toEqual({
type: "assistant_message",
text: "final reply",
});
});
test("fetchTimeline returns full timeline with reset when cursor seq falls behind retention window", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-gap-"));
test("fetchTimeline supports older-history pagination with before seq", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-before-"));
const storagePath = join(workdir, "agents");
const storage = new AgentStorage(storagePath, logger);
const manager = new AgentManager({
@@ -908,7 +1047,6 @@ describe("AgentManager", () => {
},
registry: storage,
logger,
maxTimelineItems: 2,
idFactory: () => "00000000-0000-4000-8000-000000000119",
});
@@ -933,32 +1071,27 @@ describe("AgentManager", () => {
type: "assistant_message",
text: "fourth",
});
const fresh = manager.fetchTimeline(snapshot.id, {
direction: "tail",
limit: 0,
await manager.appendTimelineItem(snapshot.id, {
type: "assistant_message",
text: "fifth",
});
expect(fresh.window.minSeq).toBe(3);
expect(fresh.window.maxSeq).toBe(4);
const result = manager.fetchTimeline(snapshot.id, {
direction: "after",
direction: "before",
cursor: {
epoch: fresh.epoch,
seq: 1,
seq: 5,
},
limit: 10,
limit: 2,
});
expect(result.reset).toBe(true);
expect(result.staleCursor).toBe(false);
expect(result.gap).toBe(true);
expect(result.rows).toHaveLength(2);
expect(result.rows[0]?.seq).toBe(3);
expect(result.rows[1]?.seq).toBe(4);
expect(result.hasOlder).toBe(true);
expect(result.hasNewer).toBe(true);
});
test("does not trim timeline by default", async () => {
test("does not trim committed history", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-unbounded-"));
const storagePath = join(workdir, "agents");
const storage = new AgentStorage(storagePath, logger);
@@ -998,6 +1131,178 @@ describe("AgentManager", () => {
expect(fetched.window.maxSeq).toBe(3);
});
test("hydrateTimeline canonicalizes tool-interleaved assistant replay into the committed turn shape", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-history-canonical-assistant-"));
const storagePath = join(workdir, "agents");
const storage = new AgentStorage(storagePath, logger);
class ChunkedAssistantHistorySession extends TestAgentSession {
constructor(config: AgentSessionConfig) {
super(config);
}
async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
yield {
type: "timeline",
provider: this.provider,
item: { type: "assistant_message", text: "chunk one " },
};
yield {
type: "timeline",
provider: this.provider,
item: { type: "assistant_message", text: "chunk two" },
};
yield {
type: "timeline",
provider: this.provider,
item: { type: "reasoning", text: "internal" },
};
yield {
type: "timeline",
provider: this.provider,
item: {
type: "tool_call",
callId: "call-history-1",
name: "shell",
status: "completed",
detail: {
type: "shell",
command: "echo hi",
output: "hi\n",
exitCode: 0,
},
error: null,
},
};
yield {
type: "timeline",
provider: this.provider,
item: { type: "assistant_message", text: "final answer" },
};
}
}
class ChunkedAssistantHistoryClient implements AgentClient {
readonly provider = "codex" as const;
readonly capabilities = TEST_CAPABILITIES;
async isAvailable(): Promise<boolean> {
return true;
}
async createSession(config: AgentSessionConfig): Promise<AgentSession> {
return new ChunkedAssistantHistorySession(config);
}
async resumeSession(): Promise<AgentSession> {
throw new Error("Not used in this test");
}
}
const manager = new AgentManager({
clients: {
codex: new ChunkedAssistantHistoryClient(),
},
registry: storage,
logger,
idFactory: () => "00000000-0000-4000-8000-000000000121",
});
const snapshot = await manager.createAgent({
provider: "codex",
cwd: workdir,
});
await manager.hydrateTimelineFromProvider(snapshot.id);
expect(manager.getTimeline(snapshot.id)).toEqual([
{
type: "tool_call",
callId: "call-history-1",
name: "shell",
status: "completed",
detail: {
type: "shell",
command: "echo hi",
output: "hi\n",
exitCode: 0,
},
error: null,
},
{ type: "assistant_message", text: "chunk one chunk twofinal answer" },
]);
});
test("hydrateTimeline canonicalizes reasoning-interleaved assistant replay into one committed assistant row", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-history-reasoning-interleave-"));
const storagePath = join(workdir, "agents");
const storage = new AgentStorage(storagePath, logger);
class ReasoningInterleavedHistorySession extends TestAgentSession {
constructor(config: AgentSessionConfig) {
super(config);
}
async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
yield {
type: "timeline",
provider: this.provider,
item: { type: "assistant_message", text: "before reasoning " },
};
yield {
type: "timeline",
provider: this.provider,
item: { type: "reasoning", text: "internal step" },
};
yield {
type: "timeline",
provider: this.provider,
item: { type: "assistant_message", text: "after reasoning" },
};
}
}
class ReasoningInterleavedHistoryClient implements AgentClient {
readonly provider = "codex" as const;
readonly capabilities = TEST_CAPABILITIES;
async isAvailable(): Promise<boolean> {
return true;
}
async createSession(config: AgentSessionConfig): Promise<AgentSession> {
return new ReasoningInterleavedHistorySession(config);
}
async resumeSession(): Promise<AgentSession> {
throw new Error("Not used in this test");
}
}
const manager = new AgentManager({
clients: {
codex: new ReasoningInterleavedHistoryClient(),
},
registry: storage,
logger,
idFactory: () => "00000000-0000-4000-8000-000000000122",
});
const snapshot = await manager.createAgent({
provider: "codex",
cwd: workdir,
});
await manager.hydrateTimelineFromProvider(snapshot.id);
expect(manager.getTimeline(snapshot.id)).toEqual([
{
type: "assistant_message",
text: "before reasoning after reasoning",
},
]);
});
test("createAgent fails when generated agent ID is not a UUID", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-"));
const storagePath = join(workdir, "agents");
packages/server/src/server/agent/agent-manager.ts
+95 -89
View File
@@ -42,7 +42,6 @@ export type AgentManagerEvent =
agentId: string;
event: AgentStreamEvent;
seq?: number;
epoch?: string;
};
export type AgentSubscriber = (event: AgentManagerEvent) => void;
@@ -70,7 +69,6 @@ export type ProviderAvailability = {
export type AgentManagerOptions = {
clients?: Partial<Record<AgentProvider, AgentClient>>;
maxTimelineItems?: number;
idFactory?: () => string;
registry?: AgentStorage;
onAgentAttention?: AgentAttentionCallback;
@@ -99,7 +97,6 @@ export type AgentTimelineRow = {
};
export type AgentTimelineCursor = {
epoch: string;
seq: number;
};
@@ -123,11 +120,7 @@ export type AgentTimelineWindow = {
};
export type AgentTimelineFetchResult = {
epoch: string;
direction: AgentTimelineFetchDirection;
reset: boolean;
staleCursor: boolean;
gap: boolean;
window: AgentTimelineWindow;
hasOlder: boolean;
hasNewer: boolean;
@@ -173,8 +166,8 @@ type ManagedAgentBase = {
pendingReplacement: boolean;
timeline: AgentTimelineItem[];
timelineRows: AgentTimelineRow[];
timelineEpoch: string;
timelineNextSeq: number;
provisionalAssistantText: string | null;
persistence: AgentPersistenceHandle | null;
historyPrimed: boolean;
lastUserMessageAt: Date | null;
@@ -308,7 +301,6 @@ export class AgentManager {
private readonly agents = new Map<string, ActiveManagedAgent>();
private readonly pendingForegroundRuns = new Map<string, PendingForegroundRun>();
private readonly subscribers = new Set<SubscriptionRecord>();
private readonly maxTimelineItems: number | null;
private readonly idFactory: () => string;
private readonly registry?: AgentStorage;
private readonly previousStatuses = new Map<string, AgentLifecycleStatus>();
@@ -317,13 +309,6 @@ export class AgentManager {
private logger: Logger;
constructor(options: AgentManagerOptions) {
const maxTimelineItems = options?.maxTimelineItems;
this.maxTimelineItems =
typeof maxTimelineItems === "number" &&
Number.isFinite(maxTimelineItems) &&
maxTimelineItems >= 0
? Math.floor(maxTimelineItems)
: null;
this.idFactory = options?.idFactory ?? (() => randomUUID());
this.registry = options?.registry;
this.onAgentAttention = options?.onAgentAttention;
@@ -525,7 +510,7 @@ export class AgentManager {
fetchTimeline(id: string, options?: AgentTimelineFetchOptions): AgentTimelineFetchResult {
const agent = this.requireAgent(id);
const { rows, epoch, nextSeq, minSeq, maxSeq } = this.ensureTimelineState(agent);
const { rows, nextSeq, minSeq, maxSeq } = this.ensureTimelineState(agent);
const direction = options?.direction ?? "tail";
const requestedLimit = options?.limit;
const limit =
@@ -536,44 +521,12 @@ export class AgentManager {
const window: AgentTimelineWindow = { minSeq, maxSeq, nextSeq };
if (cursor && cursor.epoch !== epoch) {
return {
epoch,
direction,
reset: true,
staleCursor: true,
gap: false,
window,
hasOlder: false,
hasNewer: false,
rows: rows.map((row) => ({ ...row })),
};
}
const selectAll = limit === 0;
const cloneRows = (items: AgentTimelineRow[]) => items.map((row) => ({ ...row }));
if (direction === "after" && cursor && rows.length > 0 && cursor.seq < minSeq - 1) {
return {
epoch,
direction,
reset: true,
staleCursor: false,
gap: true,
window,
hasOlder: false,
hasNewer: false,
rows: cloneRows(rows),
};
}
if (rows.length === 0) {
return {
epoch,
direction,
reset: false,
staleCursor: false,
gap: false,
window,
hasOlder: false,
hasNewer: false,
@@ -585,11 +538,7 @@ export class AgentManager {
const selected = selectAll || limit >= rows.length ? rows : rows.slice(rows.length - limit);
const hasOlder = selected.length > 0 && selected[0]!.seq > minSeq;
return {
epoch,
direction,
reset: false,
staleCursor: false,
gap: false,
window,
hasOlder,
hasNewer: false,
@@ -602,11 +551,7 @@ export class AgentManager {
const startIdx = rows.findIndex((row) => row.seq > baseSeq);
if (startIdx < 0) {
return {
epoch,
direction,
reset: false,
staleCursor: false,
gap: false,
window,
hasOlder: baseSeq >= minSeq,
hasNewer: false,
@@ -617,11 +562,7 @@ export class AgentManager {
const selected = selectAll ? rows.slice(startIdx) : rows.slice(startIdx, startIdx + limit);
const lastSelected = selected[selected.length - 1];
return {
epoch,
direction,
reset: false,
staleCursor: false,
gap: false,
window,
hasOlder: selected[0]!.seq > minSeq,
hasNewer: Boolean(lastSelected && lastSelected.seq < maxSeq),
@@ -640,11 +581,7 @@ export class AgentManager {
const hasOlder = selected.length > 0 && selected[0]!.seq > minSeq;
const hasNewer = endExclusive >= 0;
return {
epoch,
direction,
reset: false,
staleCursor: false,
gap: false,
window,
hasOlder,
hasNewer,
@@ -722,8 +659,8 @@ export class AgentManager {
const timelineState = this.ensureTimelineState(existing);
const preservedTimeline = [...existing.timeline];
const preservedTimelineRows = timelineState.rows.map((row) => ({ ...row }));
const preservedTimelineEpoch = timelineState.epoch;
const preservedTimelineNextSeq = timelineState.nextSeq;
const preservedProvisionalAssistantText = existing.provisionalAssistantText;
const preservedHistoryPrimed = existing.historyPrimed;
const preservedLastUsage = existing.lastUsage;
const preservedLastError = existing.lastError;
@@ -768,8 +705,8 @@ export class AgentManager {
lastUserMessageAt: existing.lastUserMessageAt,
timeline: preservedTimeline,
timelineRows: preservedTimelineRows,
timelineEpoch: preservedTimelineEpoch,
timelineNextSeq: preservedTimelineNextSeq,
provisionalAssistantText: preservedProvisionalAssistantText,
historyPrimed: preservedHistoryPrimed,
lastUsage: preservedLastUsage,
lastError: preservedLastError,
@@ -964,7 +901,6 @@ export class AgentManager {
},
{
seq: row.seq,
epoch: this.ensureTimelineState(agent).epoch,
},
);
if (options?.emitState !== false) {
@@ -985,7 +921,6 @@ export class AgentManager {
},
{
seq: row.seq,
epoch: this.ensureTimelineState(agent).epoch,
},
);
await this.persistSnapshot(agent);
@@ -1640,8 +1575,8 @@ export class AgentManager {
labels?: Record<string, string>;
timeline?: AgentTimelineItem[];
timelineRows?: AgentTimelineRow[];
timelineEpoch?: string;
timelineNextSeq?: number;
provisionalAssistantText?: string | null;
historyPrimed?: boolean;
lastUsage?: AgentUsage;
lastError?: string;
@@ -1689,8 +1624,8 @@ export class AgentManager {
unsubscribeSession: null,
timeline: initialTimeline,
timelineRows: initialTimelineRows,
timelineEpoch: options?.timelineEpoch ?? randomUUID(),
timelineNextSeq: derivedNextSeq,
provisionalAssistantText: options?.provisionalAssistantText ?? null,
persistence: attachPersistenceCwd(session.describePersistence(), config.cwd),
historyPrimed: options?.historyPrimed ?? false,
lastUserMessageAt: options?.lastUserMessageAt ?? null,
@@ -1841,7 +1776,6 @@ export class AgentManager {
private ensureTimelineState(agent: ManagedAgent): {
rows: AgentTimelineRow[];
epoch: string;
nextSeq: number;
minSeq: number;
maxSeq: number;
@@ -1853,7 +1787,6 @@ export class AgentManager {
return {
rows: agent.timelineRows,
epoch: agent.timelineEpoch,
nextSeq: agent.timelineNextSeq,
minSeq,
maxSeq,
@@ -1939,14 +1872,63 @@ export class AgentManager {
return [[messageId, row.item.text]];
}),
);
const pendingTurnItems: AgentTimelineItem[] = [];
let bufferedAssistantText = "";
const flushPendingTurn = () => {
for (const item of pendingTurnItems) {
this.recordTimeline(agent, item);
}
pendingTurnItems.length = 0;
if (bufferedAssistantText) {
this.recordTimeline(agent, {
type: "assistant_message",
text: bufferedAssistantText,
});
bufferedAssistantText = "";
}
};
try {
for await (const event of agent.session.streamHistory()) {
this.handleStreamEvent(agent, event, {
fromHistory: true,
canonicalUserMessagesById:
canonicalUserMessagesById.size > 0 ? canonicalUserMessagesById : undefined,
});
if (event.type !== "timeline") {
if (
event.type === "turn_completed" ||
event.type === "turn_failed" ||
event.type === "turn_canceled"
) {
flushPendingTurn();
}
continue;
}
if (event.item.type === "user_message") {
flushPendingTurn();
const eventMessageId = normalizeMessageId(event.item.messageId);
if (eventMessageId) {
const canonicalText = canonicalUserMessagesById.get(eventMessageId);
if (canonicalText === event.item.text) {
continue;
}
}
this.recordTimeline(agent, event.item);
continue;
}
if (event.item.type === "assistant_message") {
bufferedAssistantText += event.item.text;
continue;
}
if (event.item.type === "reasoning") {
continue;
}
if (event.item.type === "tool_call" && event.item.status === "running") {
continue;
}
pendingTurnItems.push(event.item);
}
flushPendingTurn();
} catch {
// ignore history failures
}
@@ -2019,6 +2001,16 @@ export class AgentManager {
}
}
}
if (event.item.type === "assistant_message") {
agent.provisionalAssistantText = `${agent.provisionalAssistantText ?? ""}${event.item.text}`;
break;
}
if (event.item.type === "reasoning") {
break;
}
if (event.item.type === "tool_call" && event.item.status === "running") {
break;
}
timelineRow = this.recordTimeline(agent, event.item);
if (!options?.fromHistory && event.item.type === "user_message") {
agent.lastUserMessageAt = new Date();
@@ -2035,6 +2027,27 @@ export class AgentManager {
},
"handleStreamEvent: turn_completed",
);
if (agent.provisionalAssistantText) {
const item: AgentTimelineItem = {
type: "assistant_message",
text: agent.provisionalAssistantText,
};
timelineRow = this.recordTimeline(agent, item);
if (!options?.fromHistory) {
this.dispatchStream(
agent.id,
{
type: "timeline",
item,
provider: event.provider,
},
{
seq: timelineRow.seq,
},
);
}
agent.provisionalAssistantText = null;
}
agent.lastUsage = event.usage;
agent.lastError = undefined;
// For autonomous turns (not foreground), transition to idle
@@ -2058,6 +2071,7 @@ export class AgentManager {
},
"handleStreamEvent: turn_failed",
);
agent.provisionalAssistantText = null;
// For autonomous turns, set error state directly
if (!isForegroundEvent) {
agent.lifecycle = "error";
@@ -2094,6 +2108,7 @@ export class AgentManager {
},
"handleStreamEvent: turn_canceled",
);
agent.provisionalAssistantText = null;
// For autonomous turns, transition to idle
// unless a replacement is pending (avoid idle flash during replace)
if (!isForegroundEvent && !agent.pendingReplacement) {
@@ -2125,6 +2140,7 @@ export class AgentManager {
},
"handleStreamEvent: turn_started",
);
agent.provisionalAssistantText = null;
// For autonomous turn_started (no foreground match), set running
if (!isForegroundEvent) {
(agent as ActiveManagedAgent).lifecycle = "running";
@@ -2161,7 +2177,6 @@ export class AgentManager {
timelineRow
? {
seq: timelineRow.seq,
epoch: this.ensureTimelineState(agent).epoch,
}
: undefined,
);
@@ -2203,7 +2218,6 @@ export class AgentManager {
},
{
seq: row.seq,
epoch: this.ensureTimelineState(agent).epoch,
},
);
}
@@ -2234,14 +2248,6 @@ export class AgentManager {
agent.timelineNextSeq = timelineState.nextSeq + 1;
agent.timeline.push(item);
timelineState.rows.push(row);
if (
typeof this.maxTimelineItems === "number" &&
agent.timeline.length > this.maxTimelineItems
) {
const removeCount = agent.timeline.length - this.maxTimelineItems;
agent.timeline.splice(0, removeCount);
timelineState.rows.splice(0, removeCount);
}
return row;
}
@@ -2337,7 +2343,7 @@ export class AgentManager {
private dispatchStream(
agentId: string,
event: AgentStreamEvent,
metadata?: { seq?: number; epoch?: string },
metadata?: { seq?: number },
): void {
this.dispatch({ type: "agent_stream", agentId, event, ...metadata });
}
packages/server/src/server/daemon-client.e2e.test.ts
-2
View File
@@ -529,7 +529,6 @@ describe("daemon client E2E", () => {
const timelineResult = await ctx.client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 1,
projection: "projected",
});
expect(timelineResult.agentId).toBe(agent.id);
@@ -756,7 +755,6 @@ describe("daemon client E2E", () => {
const timeline = await ctx.client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "projected",
});
expect(timeline.entries.length).toBeGreaterThan(0);
packages/server/src/server/daemon-e2e/agent-operations.e2e.test.ts
-2
View File
@@ -71,7 +71,6 @@ describe("daemon E2E", () => {
await ctx.client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 200,
projection: "projected",
});
const refreshedResult = await ctx.client.fetchAgent(agent.id);
@@ -89,7 +88,6 @@ describe("daemon E2E", () => {
await ctx.client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 200,
projection: "projected",
});
const clearResult = await ctx.client.fetchAgent(agent.id);
packages/server/src/server/daemon-e2e/claude-autonomous-wake-simple.real.e2e.test.ts
-2
View File
@@ -60,7 +60,6 @@ describe("daemon E2E (real claude) - autonomous wake simple", () => {
const timelineAtIdle = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
const idleAssistantText = timelineAtIdle.entries
.filter(
@@ -91,7 +90,6 @@ describe("daemon E2E (real claude) - autonomous wake simple", () => {
const finalTimeline = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
const finalAssistantText = finalTimeline.entries
.filter(
packages/server/src/server/daemon-e2e/claude-autonomous-wake.real.e2e.test.ts
-9
View File
@@ -390,7 +390,6 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () =
const timelineAtIdle = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
await client.waitForAgentUpsert(
@@ -405,7 +404,6 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () =
const timelineAfterWake = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
expect(timelineAfterWake.entries.length).toBeGreaterThanOrEqual(
timelineAtIdle.entries.length,
@@ -417,7 +415,6 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () =
const nextTimeline = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
sawTimelineGrowth = nextTimeline.entries.length > timelineAtIdle.entries.length;
}
@@ -605,7 +602,6 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () =
const timelineBeforeWake = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
const summarized = timelineBeforeWake.entries.map(summarizeTimelineEntry);
// Required by reproduction request: log timeline at idle edge before autonomous wake.
@@ -735,7 +731,6 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () =
const timelineAtWake = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
// eslint-disable-next-line no-console
@@ -751,7 +746,6 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () =
const timelineAfterWake = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
expect(timelineAfterWake.entries.length).toBeGreaterThanOrEqual(
timelineAtWake.entries.length,
@@ -916,7 +910,6 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () =
const timelineAtWake = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
// eslint-disable-next-line no-console
@@ -932,7 +925,6 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () =
const timelineAfterWake = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
expect(timelineAfterWake.entries.length).toBeGreaterThanOrEqual(
timelineAtWake.entries.length,
@@ -1066,7 +1058,6 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () =
const timeline = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
assistantTexts = timeline.entries
.filter(
packages/server/src/server/daemon-e2e/persistence.e2e.test.ts
+5 -2
View File
@@ -114,11 +114,14 @@ describe("daemon E2E - persistence", () => {
const timeline = await ctx.client.fetchAgentTimeline(agentId, {
direction: "tail",
limit: 0,
projection: "canonical",
});
const timelineItems = timeline.entries.map((entry) => entry.item);
expect(timelineItems.length).toBeGreaterThan(0);
expect(timelineItems.some((item) => item.type === "assistant_message")).toBe(true);
const assistantMessages = timelineItems.filter(
(item): item is Extract<(typeof timelineItems)[number], { type: "assistant_message" }> =>
item.type === "assistant_message",
);
expect(assistantMessages).toEqual([{ type: "assistant_message", text: "timeline test" }]);
} finally {
await ctx.cleanup();
cleaned = true;
packages/server/src/server/daemon-e2e/rewind-user-message-dedupe-claude.real.e2e.test.ts
-1
View File
@@ -48,7 +48,6 @@ describe("daemon E2E (real claude) - rewind user message dedupe", () => {
const timeline = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
const rewindUserMessages = timeline.entries.filter(
packages/server/src/server/daemon-e2e/timeline-reconnect-contract.e2e.test.ts
+206
View File
@@ -0,0 +1,206 @@
import { afterEach, beforeEach, describe, expect, test } from "vitest";
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import { createDaemonTestContext, type DaemonTestContext, DaemonClient } from "../test-utils/index.js";
import { createMessageCollector } from "../test-utils/message-collector.js";
import type { SessionOutboundMessage } from "../messages.js";
function tmpCwd(): string {
return mkdtempSync(path.join(tmpdir(), "daemon-e2e-"));
}
async function waitFor(
predicate: () => boolean,
timeoutMs = 5_000,
intervalMs = 10,
): Promise<void> {
const startedAt = Date.now();
while (!predicate()) {
if (Date.now() - startedAt > timeoutMs) {
throw new Error(`Timed out after ${timeoutMs}ms waiting for condition`);
}
await new Promise((resolve) => setTimeout(resolve, intervalMs));
}
}
function isSeqLessAssistantTimeline(
message: SessionOutboundMessage,
agentId: string,
text?: string,
): boolean {
return (
message.type === "agent_stream" &&
message.payload.agentId === agentId &&
message.payload.event.type === "timeline" &&
message.payload.event.item.type === "assistant_message" &&
message.payload.seq === undefined &&
(text === undefined || message.payload.event.item.text === text)
);
}
describe("daemon E2E - timeline reconnect contract", () => {
let ctx: DaemonTestContext;
beforeEach(async () => {
ctx = await createDaemonTestContext();
});
afterEach(async () => {
await ctx.cleanup();
}, 60_000);
test("reconnect catches up committed rows without replaying a provisional seed", async () => {
const cwd = tmpCwd();
const primaryCollector = createMessageCollector(ctx.client);
try {
const agent = await ctx.client.createAgent({
provider: "codex",
cwd,
title: "Reconnect Contract Test",
modeId: "full-access",
});
for (let seq = 1; seq <= 120; seq += 1) {
await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, {
type: "assistant_message",
text: `committed row ${seq}`,
});
}
primaryCollector.clear();
await ctx.daemon.daemon.agentManager.emitLiveTimelineItem(agent.id, {
type: "assistant_message",
text: "partial before disconnect",
});
await waitFor(() =>
primaryCollector.messages.some((message) =>
isSeqLessAssistantTimeline(message, agent.id, "partial before disconnect"),
),
);
await ctx.client.close();
await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, {
type: "assistant_message",
text: "finalized while disconnected",
});
const reconnectClient = new DaemonClient({
url: `ws://127.0.0.1:${ctx.daemon.port}/ws`,
});
await reconnectClient.connect();
const reconnectCollector = createMessageCollector(reconnectClient);
try {
await reconnectClient.fetchAgents({
subscribe: { subscriptionId: "timeline-reconnect-a" },
});
expect(
reconnectCollector.messages.some((message) =>
isSeqLessAssistantTimeline(message, agent.id),
),
).toBe(false);
const catchUp = await reconnectClient.fetchAgentTimeline(agent.id, {
direction: "after",
cursor: { seq: 120 },
limit: 0,
});
expect(catchUp.entries).toHaveLength(1);
expect(catchUp.entries[0]?.seq).toBe(121);
expect(catchUp.entries[0]?.item).toEqual({
type: "assistant_message",
text: "finalized while disconnected",
});
} finally {
reconnectCollector.unsubscribe();
await reconnectClient.close();
}
} finally {
primaryCollector.unsubscribe();
rmSync(cwd, { recursive: true, force: true });
}
}, 30_000);
test("reconnect with no new committed rows resumes from future live provisional updates only", async () => {
const cwd = tmpCwd();
const primaryCollector = createMessageCollector(ctx.client);
try {
const agent = await ctx.client.createAgent({
provider: "codex",
cwd,
title: "Reconnect No Seed Test",
modeId: "full-access",
});
for (let seq = 1; seq <= 120; seq += 1) {
await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, {
type: "assistant_message",
text: `committed row ${seq}`,
});
}
primaryCollector.clear();
await ctx.daemon.daemon.agentManager.emitLiveTimelineItem(agent.id, {
type: "assistant_message",
text: "partial before disconnect",
});
await waitFor(() =>
primaryCollector.messages.some((message) =>
isSeqLessAssistantTimeline(message, agent.id, "partial before disconnect"),
),
);
await ctx.client.close();
const reconnectClient = new DaemonClient({
url: `ws://127.0.0.1:${ctx.daemon.port}/ws`,
});
await reconnectClient.connect();
const reconnectCollector = createMessageCollector(reconnectClient);
try {
await reconnectClient.fetchAgents({
subscribe: { subscriptionId: "timeline-reconnect-b" },
});
expect(
reconnectCollector.messages.some((message) =>
isSeqLessAssistantTimeline(message, agent.id),
),
).toBe(false);
const catchUp = await reconnectClient.fetchAgentTimeline(agent.id, {
direction: "after",
cursor: { seq: 120 },
limit: 0,
});
expect(catchUp.entries).toHaveLength(0);
reconnectCollector.clear();
await ctx.daemon.daemon.agentManager.emitLiveTimelineItem(agent.id, {
type: "assistant_message",
text: "fresh live after reconnect",
});
await waitFor(() =>
reconnectCollector.messages.some((message) =>
isSeqLessAssistantTimeline(message, agent.id, "fresh live after reconnect"),
),
);
} finally {
reconnectCollector.unsubscribe();
await reconnectClient.close();
}
} finally {
primaryCollector.unsubscribe();
rmSync(cwd, { recursive: true, force: true });
}
}, 30_000);
});
packages/server/src/server/daemon-e2e/timeline-window.e2e.test.ts
+4 -7
View File
@@ -20,7 +20,7 @@ describe("daemon E2E - timeline window", () => {
await ctx.cleanup();
}, 60_000);
test("canonical tail limit keeps assistant chunks intact at the window boundary", async () => {
test("canonical tail limit returns one finalized committed assistant row at the window boundary", async () => {
const cwd = tmpCwd();
try {
const agent = await ctx.client.createAgent({
@@ -38,16 +38,14 @@ describe("daemon E2E - timeline window", () => {
const timeline = await ctx.client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 1,
projection: "canonical",
});
const assistantTexts = timeline.entries
.filter((entry) => entry.item.type === "assistant_message")
.map((entry) => entry.item.text);
expect(assistantTexts).toHaveLength(2);
expect(assistantTexts.join("")).toBe(expected);
expect(timeline.startCursor?.seq).toBeLessThan(timeline.endCursor?.seq ?? 0);
expect(assistantTexts).toEqual([expected]);
expect(timeline.startSeq).toBe(timeline.endSeq);
} finally {
rmSync(cwd, { recursive: true, force: true });
}
@@ -73,7 +71,6 @@ describe("daemon E2E - timeline window", () => {
const timeline = await ctx.client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 1,
projection: "canonical",
});
const assistantTexts = timeline.entries
@@ -82,7 +79,7 @@ describe("daemon E2E - timeline window", () => {
expect(assistantTexts.join("")).toBe(expected);
expect(timeline.hasOlder).toBe(true);
expect(timeline.startCursor?.seq).toBeGreaterThan(1);
expect(timeline.startSeq).toBeGreaterThan(1);
} finally {
rmSync(cwd, { recursive: true, force: true });
}
packages/server/src/server/daemon-e2e/ui-action-stress.real.e2e.test.ts
-1
View File
@@ -289,7 +289,6 @@ async function resolveLatestAssistantMessage(
const timeline = await client.fetchAgentTimeline(agentId, {
direction: "tail",
limit: 300,
projection: "canonical",
});
for (let idx = timeline.entries.length - 1; idx >= 0; idx -= 1) {
const entry = timeline.entries[idx];
packages/server/src/server/session.ts
+16 -109
View File
@@ -78,11 +78,6 @@ import {
appendTimelineItemIfAgentKnown,
emitLiveTimelineItemIfAgentKnown,
} from "./agent/timeline-append.js";
import {
projectTimelineRows,
selectTimelineWindowByProjectedLimit,
type TimelineProjectionMode,
} from "./agent/timeline-projection.js";
import {
DEFAULT_STRUCTURED_GENERATION_PROVIDERS,
StructuredAgentFallbackError,
@@ -992,7 +987,6 @@ export class Session {
event: serializedEvent,
timestamp: new Date().toISOString(),
...(typeof event.seq === "number" ? { seq: event.seq } : {}),
...(typeof event.epoch === "string" ? { epoch: event.epoch } : {}),
} as const;
this.emit({
@@ -6409,17 +6403,10 @@ export class Session {
msg: Extract<SessionInboundMessage, { type: "fetch_agent_timeline_request" }>,
): Promise<void> {
const direction: AgentTimelineFetchDirection = msg.direction ?? (msg.cursor ? "after" : "tail");
const projection: TimelineProjectionMode = msg.projection ?? "projected";
const requestedLimit = msg.limit;
const limit = requestedLimit ?? (direction === "after" ? 0 : undefined);
const shouldLimitByProjectedWindow =
projection === "canonical" &&
direction === "tail" &&
typeof requestedLimit === "number" &&
requestedLimit > 0;
const cursor: AgentTimelineCursor | undefined = msg.cursor
? {
epoch: msg.cursor.epoch,
seq: msg.cursor.seq,
}
: undefined;
@@ -6427,87 +6414,19 @@ export class Session {
try {
const snapshot = await this.ensureAgentLoaded(msg.agentId);
const agentPayload = await this.buildAgentPayload(snapshot);
let timeline = this.agentManager.fetchTimeline(msg.agentId, {
const timeline = this.agentManager.fetchTimeline(msg.agentId, {
direction,
cursor,
limit:
shouldLimitByProjectedWindow && typeof requestedLimit === "number"
? Math.max(1, Math.floor(requestedLimit))
: limit,
limit,
});
let hasOlder = timeline.hasOlder;
let hasNewer = timeline.hasNewer;
let startCursor: { epoch: string; seq: number } | null = null;
let endCursor: { epoch: string; seq: number } | null = null;
let entries: ReturnType<typeof projectTimelineRows>;
if (shouldLimitByProjectedWindow) {
const projectedLimit = Math.max(1, Math.floor(requestedLimit));
let fetchLimit = projectedLimit;
let projectedWindow = selectTimelineWindowByProjectedLimit({
rows: timeline.rows,
provider: snapshot.provider,
direction,
limit: projectedLimit,
collapseToolLifecycle: false,
});
while (timeline.hasOlder) {
const needsMoreProjectedEntries =
projectedWindow.projectedEntries.length < projectedLimit;
const firstLoadedRow = timeline.rows[0];
const firstSelectedRow = projectedWindow.selectedRows[0];
const startsAtLoadedBoundary =
firstLoadedRow != null &&
firstSelectedRow != null &&
firstSelectedRow.seq === firstLoadedRow.seq;
const boundaryIsAssistantChunk =
startsAtLoadedBoundary && firstLoadedRow.item.type === "assistant_message";
if (!needsMoreProjectedEntries && !boundaryIsAssistantChunk) {
break;
}
const maxRows = Math.max(0, timeline.window.maxSeq - timeline.window.minSeq + 1);
const nextFetchLimit = Math.min(maxRows, fetchLimit * 2);
if (nextFetchLimit <= fetchLimit) {
break;
}
fetchLimit = nextFetchLimit;
timeline = this.agentManager.fetchTimeline(msg.agentId, {
direction,
cursor,
limit: fetchLimit,
});
projectedWindow = selectTimelineWindowByProjectedLimit({
rows: timeline.rows,
provider: snapshot.provider,
direction,
limit: projectedLimit,
collapseToolLifecycle: false,
});
}
const selectedRows = projectedWindow.selectedRows;
entries = projectTimelineRows(selectedRows, snapshot.provider, projection);
if (projectedWindow.minSeq !== null && projectedWindow.maxSeq !== null) {
startCursor = { epoch: timeline.epoch, seq: projectedWindow.minSeq };
endCursor = { epoch: timeline.epoch, seq: projectedWindow.maxSeq };
hasOlder = projectedWindow.minSeq > timeline.window.minSeq;
hasNewer = false;
}
} else {
const firstRow = timeline.rows[0];
const lastRow = timeline.rows[timeline.rows.length - 1];
startCursor = firstRow ? { epoch: timeline.epoch, seq: firstRow.seq } : null;
endCursor = lastRow ? { epoch: timeline.epoch, seq: lastRow.seq } : null;
entries = projectTimelineRows(timeline.rows, snapshot.provider, projection);
}
const firstRow = timeline.rows[0];
const lastRow = timeline.rows[timeline.rows.length - 1];
const entries = timeline.rows.map((row) => ({
provider: snapshot.provider,
item: row.item,
timestamp: row.timestamp,
seq: row.seq,
}));
this.emit({
type: "fetch_agent_timeline_response",
@@ -6516,16 +6435,10 @@ export class Session {
agentId: msg.agentId,
agent: agentPayload,
direction,
projection,
epoch: timeline.epoch,
reset: timeline.reset,
staleCursor: timeline.staleCursor,
gap: timeline.gap,
window: timeline.window,
startCursor,
endCursor,
hasOlder,
hasNewer,
startSeq: firstRow?.seq ?? null,
endSeq: lastRow?.seq ?? null,
hasOlder: timeline.hasOlder,
hasNewer: timeline.hasNewer,
entries,
error: null,
},
@@ -6542,14 +6455,8 @@ export class Session {
agentId: msg.agentId,
agent: null,
direction,
projection,
epoch: "",
reset: false,
staleCursor: false,
gap: false,
window: { minSeq: 0, maxSeq: 0, nextSeq: 0 },
startCursor: null,
endCursor: null,
startSeq: null,
endSeq: null,
hasOlder: false,
hasNewer: false,
entries: [],
packages/server/src/shared/messages.stream-parsing.test.ts
+71 -12
View File
@@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest";
import {
AgentStreamMessageSchema,
FetchAgentTimelineRequestMessageSchema,
FetchAgentTimelineResponseMessageSchema,
SessionInboundMessageSchema,
SessionOutboundMessageSchema,
@@ -17,14 +18,8 @@ describe("shared messages stream parsing", () => {
agentId: "agent_live",
agent: null,
direction: "tail",
projection: "projected",
epoch: "epoch-1",
reset: false,
staleCursor: false,
gap: false,
window: { minSeq: 1, maxSeq: 2, nextSeq: 3 },
startCursor: { epoch: "epoch-1", seq: 1 },
endCursor: { epoch: "epoch-1", seq: 2 },
startSeq: 1,
endSeq: 2,
hasOlder: false,
hasNewer: false,
entries: [
@@ -32,10 +27,7 @@ describe("shared messages stream parsing", () => {
provider: "codex",
item: { type: "assistant_message", text: "hello" },
timestamp: "2026-02-08T20:10:00.000Z",
seqStart: 1,
seqEnd: 2,
sourceSeqRanges: [{ startSeq: 1, endSeq: 2 }],
collapsed: ["assistant_merge"],
seq: 2,
},
],
error: null,
@@ -46,6 +38,51 @@ describe("shared messages stream parsing", () => {
expect(parsed.payload.entries[0]?.item.type).toBe("assistant_message");
});
it("rejects removed fetch timeline request baggage at the parser boundary", () => {
const parsed = FetchAgentTimelineRequestMessageSchema.safeParse({
type: "fetch_agent_timeline_request",
agentId: "agent_live",
requestId: "req-legacy",
direction: "after",
cursor: {
seq: 12,
epoch: "legacy-epoch",
},
projection: "canonical",
});
expect(parsed.success).toBe(false);
});
it("rejects removed fetch timeline response baggage at the parser boundary", () => {
const parsed = FetchAgentTimelineResponseMessageSchema.safeParse({
type: "fetch_agent_timeline_response",
payload: {
requestId: "req-1",
agentId: "agent_live",
agent: null,
direction: "tail",
startSeq: 1,
endSeq: 2,
hasOlder: false,
hasNewer: false,
reset: false,
startCursor: { seq: 1 },
entries: [
{
provider: "codex",
item: { type: "assistant_message", text: "hello" },
timestamp: "2026-02-08T20:10:00.000Z",
seq: 2,
},
],
error: null,
},
});
expect(parsed.success).toBe(false);
});
it("parses explicit shutdown and restart lifecycle request payloads as distinct message types", () => {
const shutdownParsed = SessionInboundMessageSchema.safeParse({
type: "shutdown_server_request",
@@ -70,6 +107,7 @@ describe("shared messages stream parsing", () => {
payload: {
agentId: "agent_live",
timestamp: "2026-02-08T20:10:00.000Z",
seq: 12,
event: {
type: "timeline",
provider: "claude",
@@ -97,6 +135,27 @@ describe("shared messages stream parsing", () => {
}
});
it("rejects removed agent_stream baggage at the parser boundary", () => {
const parsed = AgentStreamMessageSchema.safeParse({
type: "agent_stream",
payload: {
agentId: "agent_live",
timestamp: "2026-02-08T20:10:00.000Z",
epoch: "legacy-epoch",
event: {
type: "timeline",
provider: "claude",
item: {
type: "assistant_message",
text: "hello",
},
},
},
});
expect(parsed.success).toBe(false);
});
it("parses representative sub_agent tool_call event", () => {
const parsed = AgentStreamMessageSchema.parse({
type: "agent_stream",
packages/server/src/shared/messages.ts
+57 -65
View File
@@ -722,22 +722,23 @@ export const ShutdownServerRequestMessageSchema = z.object({
requestId: z.string(),
});
export const AgentTimelineCursorSchema = z.object({
epoch: z.string(),
seq: z.number().int().nonnegative(),
});
export const AgentTimelineCursorSchema = z
.object({
seq: z.number().int().nonnegative(),
})
.strict();
export const FetchAgentTimelineRequestMessageSchema = z.object({
type: z.literal("fetch_agent_timeline_request"),
agentId: z.string(),
requestId: z.string(),
direction: z.enum(["tail", "before", "after"]).optional(),
cursor: AgentTimelineCursorSchema.optional(),
// 0 means "all matching rows for this query window".
limit: z.number().int().nonnegative().optional(),
// Default should be projected for app timeline loading.
projection: z.enum(["projected", "canonical"]).optional(),
});
export const FetchAgentTimelineRequestMessageSchema = z
.object({
type: z.literal("fetch_agent_timeline_request"),
agentId: z.string(),
requestId: z.string(),
direction: z.enum(["tail", "before", "after"]).optional(),
cursor: AgentTimelineCursorSchema.optional(),
// 0 means "all matching rows for this query window".
limit: z.number().int().nonnegative().optional(),
})
.strict();
export const SetAgentModeRequestMessageSchema = z.object({
type: z.literal("set_agent_mode_request"),
@@ -1545,17 +1546,20 @@ export const AgentUpdateMessageSchema = z.object({
]),
});
export const AgentStreamMessageSchema = z.object({
type: z.literal("agent_stream"),
payload: z.object({
agentId: z.string(),
event: AgentStreamEventPayloadSchema,
timestamp: z.string(),
// Present for timeline events. Maps 1:1 to canonical in-memory timeline rows.
seq: z.number().int().nonnegative().optional(),
epoch: z.string().optional(),
}),
});
export const AgentStreamMessageSchema = z
.object({
type: z.literal("agent_stream"),
payload: z
.object({
agentId: z.string(),
event: AgentStreamEventPayloadSchema,
timestamp: z.string(),
// Present only for committed timeline events.
seq: z.number().int().nonnegative().optional(),
})
.strict(),
})
.strict();
export const AgentStatusMessageSchema = z.object({
type: z.literal("agent_status"),
@@ -1649,46 +1653,34 @@ export const FetchAgentResponseMessageSchema = z.object({
}),
});
const AgentTimelineSeqRangeSchema = z.object({
startSeq: z.number().int().nonnegative(),
endSeq: z.number().int().nonnegative(),
});
export const AgentTimelineEntryPayloadSchema = z
.object({
provider: AgentProviderSchema,
item: AgentTimelineItemPayloadSchema,
timestamp: z.string(),
seq: z.number().int().nonnegative(),
})
.strict();
export const AgentTimelineEntryPayloadSchema = z.object({
provider: AgentProviderSchema,
item: AgentTimelineItemPayloadSchema,
timestamp: z.string(),
seqStart: z.number().int().nonnegative(),
seqEnd: z.number().int().nonnegative(),
sourceSeqRanges: z.array(AgentTimelineSeqRangeSchema),
collapsed: z.array(z.enum(["assistant_merge", "tool_lifecycle"])),
});
export const FetchAgentTimelineResponseMessageSchema = z.object({
type: z.literal("fetch_agent_timeline_response"),
payload: z.object({
requestId: z.string(),
agentId: z.string(),
agent: AgentSnapshotPayloadSchema.nullable(),
direction: z.enum(["tail", "before", "after"]),
projection: z.enum(["projected", "canonical"]),
epoch: z.string(),
reset: z.boolean(),
staleCursor: z.boolean(),
gap: z.boolean(),
window: z.object({
minSeq: z.number().int().nonnegative(),
maxSeq: z.number().int().nonnegative(),
nextSeq: z.number().int().nonnegative(),
}),
startCursor: AgentTimelineCursorSchema.nullable(),
endCursor: AgentTimelineCursorSchema.nullable(),
hasOlder: z.boolean(),
hasNewer: z.boolean(),
entries: z.array(AgentTimelineEntryPayloadSchema),
error: z.string().nullable(),
}),
});
export const FetchAgentTimelineResponseMessageSchema = z
.object({
type: z.literal("fetch_agent_timeline_response"),
payload: z
.object({
requestId: z.string(),
agentId: z.string(),
agent: AgentSnapshotPayloadSchema.nullable(),
direction: z.enum(["tail", "before", "after"]),
startSeq: z.number().int().nonnegative().nullable(),
endSeq: z.number().int().nonnegative().nullable(),
hasOlder: z.boolean(),
hasNewer: z.boolean(),
entries: z.array(AgentTimelineEntryPayloadSchema),
error: z.string().nullable(),
})
.strict(),
})
.strict();
export const SendAgentMessageResponseMessageSchema = z.object({
type: z.literal("send_agent_message_response"),