From 6089e226bc2423229eea942ae7e253cc847799f8 Mon Sep 17 00:00:00 2001 From: Hypatia May Date: Tue, 23 Jun 2026 13:05:59 +0800 Subject: [PATCH] refactor(session): make surface the sole derivation path, drop legacy fallback --- docs/cordis-catalog/events-and-services.md | 2 +- .../2026-06-18-session-surface.md | 2 +- packages/core/session/README.md | 6 +- packages/core/session/src/index.ts | 59 +++++++++---------- packages/core/session/src/surface.ts | 14 ----- packages/core/session/src/types.ts | 12 ++-- packages/core/session/tests/session.spec.ts | 32 +++++----- packages/core/session/tests/surface.spec.ts | 47 +-------------- .../tests/jsonl.spec.ts | 8 +-- .../tests/coordinator-contract.ts | 14 ++--- .../invariants/tests/invariants.spec.ts | 36 +++++------ 11 files changed, 87 insertions(+), 145 deletions(-) diff --git a/docs/cordis-catalog/events-and-services.md b/docs/cordis-catalog/events-and-services.md index 1ee86d6a5e..f8c6b12c0d 100644 --- a/docs/cordis-catalog/events-and-services.md +++ b/docs/cordis-catalog/events-and-services.md @@ -390,7 +390,7 @@ get(id: SessionId): Session | undefined list(): Session[] ``` -Source: [`packages/core/session/src/index.ts:303`](../../packages/core/session/src/index.ts) +Source: [`packages/core/session/src/index.ts:300`](../../packages/core/session/src/index.ts) ### `ctx.systemPrompt` — `SystemPrompt` diff --git a/docs/rfc/implemented/architecture/2026-06-18-session-surface.md b/docs/rfc/implemented/architecture/2026-06-18-session-surface.md index 644d526373..ae2475c786 100644 --- a/docs/rfc/implemented/architecture/2026-06-18-session-surface.md +++ b/docs/rfc/implemented/architecture/2026-06-18-session-surface.md @@ -53,7 +53,7 @@ The dev-mode invariants plugin validates: `sourceEventSeqs` references (non-empt ## Consequences -- **`packages/core/session`**: New `surface.ts` (`SurfaceManager`), new types (`SurfaceOp`, `SurfaceAppendOpts`), new fields on `SessionEvent`, modified `append()` (third optional `SurfaceAppendOpts` param), refactored `deriveMessages()` (surface path + legacy fallback), surface-aware `repair.ts`. +- **`packages/core/session`**: New `surface.ts` (`SurfaceManager`), new types (`SurfaceOp`, `SurfaceIntent`), new fields on `SessionEvent`, modified `append()` (third required `SurfaceIntent` param), refactored `deriveMessages()` (surface path + legacy fallback), surface-aware `repair.ts`. - **`packages/core/agent-loop`**: All surface-capable appends pass surface opts. Chunk seqs are collected for `assistant/message` provenance; `tool/call` seqs are captured for `tool/result` provenance. - **`packages/session-persistence/session-persistence-sqlite`**: Two new nullable TEXT columns (`source_event_seqs`, `surface_op`) on the `events` table; `SCHEMA_VERSION` bumped (bump-and-reject, no migration). - **`packages/support/invariants`**: Surface-related validation rules. diff --git a/packages/core/session/README.md b/packages/core/session/README.md index 206022e306..08323eff0c 100644 --- a/packages/core/session/README.md +++ b/packages/core/session/README.md @@ -34,8 +34,8 @@ Creates and holds event-sourced `Session` instances. Persistence is intentionall Plain class (not a Cordis Service). Create via `ctx.sessions.create()`. -- `session.append(type, data, opts?): SessionEvent` — synchronous, never blocks on I/O. **Throws** if `data` is not losslessly JSON-serializable (BigInt, function, symbol, undefined, non-finite number, circular ref, or an exotic object like Map/Set/Date) — the event log is the durable source of truth, so this invariant is enforced at the source (exported as `isJsonValue` for backends to reuse on their replay/fork entry points). An optional third parameter `opts: SurfaceAppendOpts` carries surface metadata: `surfaceOp` controls how the event enters the surface linked list, and `sourceEventSeqs` records provenance (the seq numbers of events this one derives from). -- `session.deriveMessages(): Message[]` — derive the LLM message history. If any event in the log carries `surfaceOp`, derivation walks the surface linked list (skipping non-surface events). Otherwise, falls back to a linear scan of the raw log (legacy sessions without surface markers). +- `session.append(type, data, opts?): SessionEvent` — synchronous, never blocks on I/O. **Throws** if `data` is not losslessly JSON-serializable (BigInt, function, symbol, undefined, non-finite number, circular ref, or an exotic object like Map/Set/Date) — the event log is the durable source of truth, so this invariant is enforced at the source (exported as `isJsonValue` for backends to reuse on their replay/fork entry points). A third parameter `opts: SurfaceIntent` carries surface metadata: `surfaceOp` controls how the event enters the surface linked list, and `sourceEventSeqs` records provenance (the seq numbers of events this one derives from). It is **required** for the five `SurfaceEventType` events (every message-producing event must declare how it joins the surface) and rejected by the compiler for non-surface types. +- `session.deriveMessages(): Message[]` — derive the LLM message history by walking the surface linked list (skipping non-surface events like chunks and boundaries; a `replace` shadows the nodes it covers). The surface is the single source of derived history — there is no raw-log fallback. - `session.surface: SurfaceManager` — the derived surface, lazily rebuilt from `surfaceOp` markers in the log. Processes only new events (delta) on each access — the log is append-only, so prior events never change. - `session.events`, `session.seq`, `session.id` - `session.header: SessionHeader` — immutable creation metadata (`version`, `id`, `createdAt`, optional `cwd`/`parentSession`). Kept out of the event log (a storage concern, not replayable state); a minimal header (stamped with the current `SESSION_FORMAT_VERSION`) is synthesized for bare `Session` construction. @@ -43,7 +43,7 @@ Plain class (not a Cordis Service). Create via `ctx.sessions.create()`. ### Surface types - `SurfaceOp` — how a surface node entered the linked list: `'append'` (normal tail append) or `{ op: 'replace', start, end }` (replace nodes from `start` through `end` inclusive — both must be valid surface node seqs; `start === end` replaces a single node). Used by compaction to shadow old nodes without deleting them. -- `SurfaceAppendOpts` — `{ surfaceOp?: SurfaceOp; sourceEventSeqs?: number[] }`, the optional third parameter to `session.append()`. +- `SurfaceIntent` — `{ surfaceOp: SurfaceOp; sourceEventSeqs?: number[] }`, the required third parameter to `session.append()` for surface-eligible types. - `SurfaceNode` — `{ seq: number; prev: number | null; next: number | null }`, one node in the surface linked list. ### Session event vocabulary (`types.ts`) diff --git a/packages/core/session/src/index.ts b/packages/core/session/src/index.ts index 23e65cebda..5df4db58e9 100644 --- a/packages/core/session/src/index.ts +++ b/packages/core/session/src/index.ts @@ -10,7 +10,7 @@ import { Context, Service } from 'cordis' import { isAbsolute } from 'node:path' import type { ContentBlock, Message, MessageSource } from '@deepseek-ai/dsh-llm' import { SESSION_FORMAT_VERSION, SessionId } from './types.ts' -import type { CreateSessionOptions, SessionEvent, SessionEventMap, SessionEventType, SessionHeader, SurfaceAppendOpts, SurfaceEventType } from './types.ts' +import type { CreateSessionOptions, SessionEvent, SessionEventMap, SessionEventType, SessionHeader, SurfaceIntent, SurfaceEventType } from './types.ts' import { isJsonValue } from './json.ts' import { SurfaceManager } from './surface.ts' @@ -149,11 +149,13 @@ export class Session { * * @param type - The event type (key of {@link SessionEventMap}). * @param data - The event payload; must be JSON-serializable. - * @param opts - Optional surface metadata: `surfaceOp` controls how the - * event enters the surface linked list; `sourceEventSeqs` records - * provenance (the seq numbers of events this one derives from). Only - * accepted for {@link SurfaceEventType} events — the compiler rejects - * surface opts for non-surface types like `turn/start` or `assistant/chunk`. + * @param opts - Surface metadata: `surfaceOp` controls how the event enters + * the surface linked list; `sourceEventSeqs` records provenance (the seq + * numbers of events this one derives from). REQUIRED for + * {@link SurfaceEventType} events (every message-producing event must + * declare how it joins the surface, the sole source of derived history) and + * rejected by the compiler for non-surface types like `turn/start` or + * `assistant/chunk`. * @throws if `data` is not losslessly JSON-serializable (BigInt, function, * symbol, undefined, non-finite number, circular ref, or an exotic object * like Map/Set/Date). The event log is the durable source of truth, so this @@ -165,7 +167,7 @@ export class Session { append( type: T, data: SessionEventMap[T], - ...opts: T extends SurfaceEventType ? [opts?: SurfaceAppendOpts] : [] + ...opts: T extends SurfaceEventType ? [opts: SurfaceIntent] : [] ): SessionEvent { if (!isJsonValue(data)) { throw new Error(`session event "${type}" carries non-JSON-serializable data`) @@ -183,7 +185,7 @@ export class Session { // Surface metadata is snapshot separately: sourceEventSeqs (number[] — // primitives, so array spread is a complete copy) and surfaceOp (a string // primitive, or cloned if it's a replace object). - const surfaceOpts: SurfaceAppendOpts | undefined = opts[0] + const surfaceOpts: SurfaceIntent | undefined = opts[0] // Build the event shape with conditional surface fields via spreading. // The result is cast through `unknown` because the conditional spreads // produce an intersection type that the assignability checker can't @@ -206,9 +208,12 @@ export class Session { } /** - * Derive the LLM message history from the session surface (when surface - * markers exist) or from a linear scan of the raw event log (legacy sessions - * without surface markers). + * Derive the LLM message history by walking the session surface — the linked + * list of message-producing events maintained by `surfaceOp` markers. The + * surface is the single source of derived history: every message-producing + * append records its `surfaceOp`, so a raw event with no marker (a chunk, a + * turn boundary) is correctly absent, and a compaction `replace` deletes the + * shadowed nodes from the derivation. * * - `user/message` → user message * - `assistant/message` → assistant message (chunks are skipped — they are @@ -229,33 +234,24 @@ export class Session { * negligible next to a model call. */ deriveMessages(): Message[] { - if (this.surface.hasSurface) { - const messages: Message[] = [] - for (const node of this.surface.nodes) { - // Surface nodes are built from this.log — node.seq is always a valid - // index by construction. The non-null assertion expresses that invariant. - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - const msg = this._deriveOneMessage(this.log[node.seq]!) - // A surface node is one of the five message-producing types, but an - // empty-content assistant/message (a max-tokens step that hosts only - // usage) derives to null and must not enter the transcript. - if (msg) messages.push(msg) - } - return messages - } - // Legacy path: linear scan for sessions without surface markers. const messages: Message[] = [] - for (const event of this.log) { - const msg = this._deriveOneMessage(event) + for (const node of this.surface.nodes) { + // Surface nodes are built from this.log — node.seq is always a valid + // index by construction. The non-null assertion expresses that invariant. + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + const msg = this._deriveOneMessage(this.log[node.seq]!) + // A surface node is one of the five message-producing types, but an + // empty-content assistant/message (a max-tokens step that hosts only + // usage) derives to null and must not enter the transcript. if (msg) messages.push(msg) } return messages } /** - * Derive a single LLM message from one event, or null if the event type - * does not produce a message. Extracted so both the surface path and the - * legacy linear-scan path share the same derivation rules. + * Derive a single LLM message from one surface event, or null if it produces + * no message (an empty-content assistant/message that exists only to host + * usage). */ private _deriveOneMessage(event: SessionEvent): Message | null { // Intentionally non-exhaustive: only message-producing events derive @@ -288,6 +284,7 @@ export class Session { const { content, source } = event.data return { role: 'user', content: renderTagged('steering', structuredClone(content), source) } } + /* v8 ignore next 2 -- unreachable: only surface nodes (the 5 message-producing types) reach here */ default: return null } diff --git a/packages/core/session/src/surface.ts b/packages/core/session/src/surface.ts index eaa8baeb4a..7ed1743af3 100644 --- a/packages/core/session/src/surface.ts +++ b/packages/core/session/src/surface.ts @@ -79,20 +79,6 @@ export class SurfaceManager { return this._nodes } - /** Whether any event in the log carries `surfaceOp` markers. */ - get hasSurface(): boolean { - if (this._nodes.length > 0) return true - // Never processed anything — scan the whole log. - if (this._lastProcessedSeq === -1) return this.log.some(e => isSurfaceEvent(e)) - // Processed up to _lastProcessedSeq without finding surface nodes; check - // only new events. - for (let i = this._lastProcessedSeq + 1; i < this.log.length; i++) { - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - if (isSurfaceEvent(this.log[i]!)) return true - } - return false - } - /** * Process events from `_lastProcessedSeq + 1` through the end of the log, * folding new surface markers into the existing linked list. diff --git a/packages/core/session/src/types.ts b/packages/core/session/src/types.ts index 1fcd150d37..44e8e35fdf 100644 --- a/packages/core/session/src/types.ts +++ b/packages/core/session/src/types.ts @@ -222,17 +222,19 @@ export type SurfaceOp = | { op: 'replace'; start: number; end: number } /** - * Optional surface metadata passed to {@link Session.append}. + * Surface metadata passed to {@link Session.append}. * `surfaceOp` controls how the event enters the surface linked list; * `sourceEventSeqs` records the seq numbers of events that are provenance * sources of this one (e.g. the `assistant/chunk` seqs behind an * `assistant/message`, or the shadowed nodes behind a compaction replacement). * - * Only accepted for {@link SurfaceEventType} events — non-surface event types - * (`turn/start`, `assistant/chunk`, `error`, …) cannot carry surface metadata. + * Required for {@link SurfaceEventType} events — every message-producing event + * MUST declare how it enters the surface, because the surface is the sole + * source of derived history. Non-surface event types (`turn/start`, + * `assistant/chunk`, `error`, …) cannot carry surface metadata. */ -export interface SurfaceAppendOpts { - surfaceOp?: SurfaceOp +export interface SurfaceIntent { + surfaceOp: SurfaceOp sourceEventSeqs?: number[] } diff --git a/packages/core/session/tests/session.spec.ts b/packages/core/session/tests/session.spec.ts index f095bdfb40..46f96742cc 100644 --- a/packages/core/session/tests/session.spec.ts +++ b/packages/core/session/tests/session.spec.ts @@ -7,7 +7,7 @@ describe('Session', () => { it('derives message history from the event log', () => { const session = new Session(SessionId('s1')) session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) - session.append('user/message', { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } }) + session.append('user/message', { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) session.append('assistant/chunk', { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'hi' } }) session.append('assistant/message', { turn: 1, step: 1, @@ -15,8 +15,8 @@ describe('Session', () => { { type: 'text', text: 'let me check' }, { type: 'tool-call', id: CallId('c1'), name: 'echo', arguments: '{}' }, ], - }) - session.append('tool/result', { turn: 1, step: 1, callId: CallId('c1'), content: [{ type: 'text', text: 'ok' }], isError: false }) + }, { surfaceOp: 'append' }) + session.append('tool/result', { turn: 1, step: 1, callId: CallId('c1'), content: [{ type: 'text', text: 'ok' }], isError: false }, { surfaceOp: 'append' }) session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) const messages = session.deriveMessages() @@ -44,12 +44,12 @@ describe('Session', () => { session.append('context/message', { content: [{ type: 'text', text: 'file changed: a.ts' }], source: { kind: 'plugin', plugin: 'watcher' }, - }) + }, { surfaceOp: 'append' }) session.append('steering/message', { turn: 1, content: [{ type: 'text', text: 'focus on tests' }], source: { kind: 'user' }, - }) + }, { surfaceOp: 'append' }) const [contextMessage, steeringMessage] = session.deriveMessages() expect(contextMessage!.role).toBe('user') @@ -60,8 +60,8 @@ describe('Session', () => { it('replays identically from a seeded event log', () => { const original = new Session(SessionId('s3')) - original.append('user/message', { content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } }) - original.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'a' }] }) + original.append('user/message', { content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) + original.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'a' }] }, { surfaceOp: 'append' }) const replayed = new Session(SessionId('s3-replay'), [...original.events]) expect(replayed.deriveMessages()).toEqual(original.deriveMessages()) @@ -70,11 +70,11 @@ describe('Session', () => { it('isolates the log from mutation through a derived message (append-only contract)', () => { const session = new Session(SessionId('s4')) - session.append('user/message', { content: [{ type: 'text', text: 'original' }], source: { kind: 'user' } }) + session.append('user/message', { content: [{ type: 'text', text: 'original' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) session.append('tool/result', { turn: 1, step: 1, callId: CallId('c1'), content: [{ type: 'text', text: 'tool out' }], isError: false, - }) + }, { surfaceOp: 'append' }) const before = structuredClone(session.events) // A request middleware / adapter mutates the messages it was handed. @@ -95,7 +95,7 @@ describe('Session', () => { it('rejects non-JSON-serializable event data at the source (incl. sparse arrays)', () => { const session = new Session(SessionId('s5')) - const bad = (extra: unknown) => () => session.append('user/message', { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' }, extra } as never) + const bad = (extra: unknown) => () => session.append('user/message', { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' }, extra } as never, { surfaceOp: 'append' }) expect(bad(1n)).toThrow(/non-JSON-serializable/) expect(bad(() => 0)).toThrow(/non-JSON-serializable/) expect(bad(Symbol('s'))).toThrow(/non-JSON-serializable/) @@ -122,7 +122,7 @@ describe('Session', () => { it('accepts dense arrays and nested plain objects', () => { const session = new Session(SessionId('s6')) - expect(() => session.append('user/message', { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' }, extra: [1, 2, [3, { a: null, b: true }]] } as never)).not.toThrow() + expect(() => session.append('user/message', { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' }, extra: [1, 2, [3, { a: null, b: true }]] } as never, { surfaceOp: 'append' })).not.toThrow() expect(session.events).toHaveLength(1) }) @@ -174,7 +174,7 @@ describe('Session', () => { it('snapshots append data: mutating the passed object after append does not affect session.events', () => { const session = new Session(SessionId('append-snapshot')) const data = { content: [{ type: 'text' as const, text: 'original' }], source: { kind: 'user' as const } } - const event = session.append('user/message', data) + const event = session.append('user/message', data, { surfaceOp: 'append' }) // Mutate the caller's object after append returns. A shared reference would // make session.events diverge from the value that passed validation. data.content[0]!.text = 'HACKED' @@ -201,7 +201,7 @@ describe('SessionStore', () => { const session = ctx.sessions.create() expect(created).toEqual([session]) - session.append('user/message', { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' } }) + session.append('user/message', { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) expect(events).toHaveLength(1) expect(events[0]![0]).toBe(session) expect(events[0]![1].type).toBe('user/message') @@ -216,7 +216,7 @@ describe('SessionStore', () => { const a = ctx.sessions.create(SessionId('fixed')) expect(() => ctx.sessions.create(SessionId('fixed'))).toThrow('already exists') - a.append('user/message', { content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } }) + a.append('user/message', { content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) const forked = ctx.sessions.create(SessionId('fork'), { seed: [...a.events] }) expect(forked.deriveMessages()).toEqual(a.deriveMessages()) }) @@ -309,7 +309,7 @@ describe('SessionStore', () => { await fiber.dispose() expect(ctx.sessions.get(SessionId('scoped'))).toBeUndefined() - session.append('user/message', { content: [{ type: 'text', text: 'late' }], source: { kind: 'user' } }) + session.append('user/message', { content: [{ type: 'text', text: 'late' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) expect(observed).toBe(0) }) @@ -332,7 +332,7 @@ describe('SessionStore', () => { ctx.on('session/event', (_session, event) => void events.push(event)) const session = ctx.sessions.create(SessionId('fixed')) expect(ctx.sessions.get(SessionId('fixed'))).toBe(session) - session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }) + session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) expect(events).toHaveLength(1) }) }) diff --git a/packages/core/session/tests/surface.spec.ts b/packages/core/session/tests/surface.spec.ts index b4e1022494..a7f4c13dad 100644 --- a/packages/core/session/tests/surface.spec.ts +++ b/packages/core/session/tests/surface.spec.ts @@ -28,38 +28,6 @@ describe('SurfaceManager', () => { expect(nodes[1]!.next).toBeNull() }) - it('hasSurface returns false when no events have surfaceOp', () => { - const s = new Session(SessionId('nosurface')) - s.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) - s.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }) - s.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) - expect(s.surface.hasSurface).toBe(false) - }) - - it('hasSurface returns true when any event has surfaceOp', () => { - const s = surfaceSession() - expect(s.surface.hasSurface).toBe(true) - }) - - it('hasSurface detects surface markers that arrive after initial processing', () => { - // Start with no surface markers. Access nodes first to set _lastProcessedSeq - // (via delta processing), keeping _nodes empty. Then append a mix of non-surface - // and surface events, and verify hasSurface detects via the delta-only check. - const s = new Session(SessionId('late')) - s.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) - s.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }) - // Access nodes to trigger processing: sets _lastProcessedSeq = 1, _nodes = []. - expect(s.surface.nodes.length).toBe(0) - // Append non-surface events first (exercises the loop-continue branch), then - // a surface event (exercises the return-true branch). - s.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) - s.append('turn/start', { turn: 2, trigger: { kind: 'continuation' } }) - s.append('assistant/message', { turn: 2, step: 1, content: [] }, { surfaceOp: 'append' }) - // hasSurface checks only new seqs [2, 3, 4]; skips 2 and 3 (non-surface), - // finds surfaceOp on seq 4 and returns true. - expect(s.surface.hasSurface).toBe(true) - }) - it('invalidate resets to full rebuild', () => { const s = surfaceSession() expect(s.surface.nodes.length).toBe(2) @@ -76,7 +44,6 @@ describe('SurfaceManager', () => { s.append('step/end', { turn: 1, step: 1 }) s.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) expect(s.surface.nodes.length).toBe(0) - expect(s.surface.hasSurface).toBe(false) // deriveMessages returns empty array expect(s.deriveMessages()).toEqual([]) }) @@ -235,16 +202,6 @@ describe('deriveMessages with surface', () => { expect(messages[1]!.content[0]).toMatchObject({ type: 'text', text: 'hi' }) }) - it('falls back to linear scan when no surface markers exist', () => { - const s = new Session(SessionId('legacy')) - s.append('user/message', { content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } }) - s.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'a' }] }) - const messages = s.deriveMessages() - expect(messages).toHaveLength(2) - expect(messages[0]!.role).toBe('user') - expect(messages[1]!.role).toBe('assistant') - }) - it('surface path skips non-surface events (chunks, boundaries)', () => { const s = new Session(SessionId('filter')) s.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) @@ -308,9 +265,9 @@ describe('Session.append surface opts', () => { expect(s.deriveMessages()).toHaveLength(0) }) - it('append without surface opts produces an event without surface fields', () => { + it('a non-surface event carries no surface fields', () => { const s = new Session(SessionId('noopts')) - s.append('user/message', { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' } }) + s.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) expect((s.events[0] as SessionEvent).sourceEventSeqs).toBeUndefined() expect((s.events[0] as SessionEvent).surfaceOp).toBeUndefined() }) diff --git a/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts b/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts index 1df70f9c9a..7df0fd5180 100644 --- a/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts +++ b/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts @@ -276,8 +276,8 @@ describe('SessionPersistenceJsonl: write path (session/event → flush)', () => const a = ctx.sessions.create(SessionId('sa')) const b = ctx.sessions.create(SessionId('sb')) - a.append('user/message', { content: [{ type: 'text', text: 'A' }], source: { kind: 'user' } }) - b.append('user/message', { content: [{ type: 'text', text: 'B' }], source: { kind: 'user' } }) + a.append('user/message', { content: [{ type: 'text', text: 'A' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) + b.append('user/message', { content: [{ type: 'text', text: 'B' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) a.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) b.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) await ctx.parallel('session/flush', a) @@ -647,7 +647,7 @@ describe('SessionPersistenceJsonl: edge cases', () => { await ctx2.plugin(SessionPersistenceJsonl, { root }) const session = ctx2.sessions.create(SessionId('flush-fail')) // A full turn lands in the write-behind buffer. - session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }) + session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) // Make the durable materialize fail on the next flush. const backend = ctx2.sessionPersistence as unknown as { materialize: (...args: unknown[]) => Promise } @@ -695,7 +695,7 @@ describe('SessionPersistenceJsonl: edge cases', () => { // can never diverge from the live log. The throw surfaces at the caller's // append site, not asynchronously in a backend flush. expect(() => { - session.append('user/message', { content: [{ type: 'text', text: 'bad' }], source: { kind: 'user' }, bad: 1n } as never) + session.append('user/message', { content: [{ type: 'text', text: 'bad' }], source: { kind: 'user' }, bad: 1n } as never, { surfaceOp: 'append' }) }).toThrow(/non-JSON-serializable/) // The bad event was rejected, so the log stayed empty. expect(session.events.length).toBe(0) diff --git a/packages/session-persistence/session-persistence/tests/coordinator-contract.ts b/packages/session-persistence/session-persistence/tests/coordinator-contract.ts index 431d02b4cb..71cd296fb0 100644 --- a/packages/session-persistence/session-persistence/tests/coordinator-contract.ts +++ b/packages/session-persistence/session-persistence/tests/coordinator-contract.ts @@ -128,7 +128,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise< const { ctx, fiber } = await freshCtx(fix) try { const session = ctx.sessions.create(SessionId('mutate'), { meta: { cwd: WORK } }) - const ev = session.append('user/message', { content: [{ type: 'text', text: 'original' }], source: { kind: 'user' } }) + const ev = session.append('user/message', { content: [{ type: 'text', text: 'original' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) // Mutate the live event object AFTER it was buffered by session/event. ;(ev.data as { content: { type: 'text'; text: string }[] }).content[0]!.text = 'HACKED' session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) @@ -232,7 +232,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise< await ctx.plugin(SessionStore) // A session exists BEFORE the persistence plugin is applied. const session = ctx.sessions.create(SessionId('pre-existing'), { meta: { cwd: WORK } }) - session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }) + session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) const fiber = await fix.mount(ctx) @@ -253,7 +253,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise< await ctx.plugin(SessionStore) const fiber = await fix.mount(ctx) const session = await liveSessionInFiber(ctx, 'drain', WORK) - session.append('user/message', { content: [{ type: 'text', text: 'buffered' }], source: { kind: 'user' } }) + session.append('user/message', { content: [{ type: 'text', text: 'buffered' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) // No explicit flush — dispose must drain. await fiber.dispose() @@ -279,7 +279,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise< // Backend instance 1 materializes the session. const backend1 = await fix.mount(ctx) session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) - session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }) + session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) await ctx.parallel('session/flush', session) @@ -290,7 +290,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise< await backend1.dispose() await fix.mount(ctx) session.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }) - session.append('user/message', { content: [{ type: 'text', text: 'again' }], source: { kind: 'user' } }) + session.append('user/message', { content: [{ type: 'text', text: 'again' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) session.append('turn/end', { turn: 2, reason: { kind: 'completed' } }) await expect(ctx.parallel('session/flush', session)).resolves.not.toThrow() @@ -452,7 +452,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise< const { ctx, fiber } = await freshCtx(fix) try { const session = ctx.sessions.create(SessionId('idem'), { meta: { cwd: WORK } }) - session.append('user/message', { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' } }) + session.append('user/message', { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) await ctx.parallel('session/flush', session) // Re-emit session/created for the SAME live session (idempotent initFor). @@ -704,7 +704,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise< // async onCreated init has necessarily set state (exercises the // state-undefined cursor path). const session = ctx.sessions.create(SessionId('flush-nostate'), { meta: { cwd: WORK } }) - session.append('user/message', { content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } }) + session.append('user/message', { content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) await ctx.parallel('session/flush', session) const loaded = await ctx.sessionPersistence.load(SessionId('flush-nostate')) diff --git a/packages/support/invariants/tests/invariants.spec.ts b/packages/support/invariants/tests/invariants.spec.ts index f66b6a8929..2eaae73169 100644 --- a/packages/support/invariants/tests/invariants.spec.ts +++ b/packages/support/invariants/tests/invariants.spec.ts @@ -25,12 +25,12 @@ describe('session-log invariants', () => { const session = ctx.sessions.create() expect(() => { session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) - session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }) + session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) session.append('step/start', { turn: 1, step: 1 }) session.append('assistant/chunk', { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'h' } }) - session.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'tool-call', id: CallId('c1'), name: 'echo', arguments: '{}' }] }) + session.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'tool-call', id: CallId('c1'), name: 'echo', arguments: '{}' }] }, { surfaceOp: 'append' }) session.append('tool/call', { turn: 1, step: 1, callId: CallId('c1'), name: 'echo', arguments: '{}' }) - session.append('tool/result', { turn: 1, step: 1, callId: CallId('c1'), content: [{ type: 'text', text: 'ok' }], isError: false }) + session.append('tool/result', { turn: 1, step: 1, callId: CallId('c1'), content: [{ type: 'text', text: 'ok' }], isError: false }, { surfaceOp: 'append' }) session.append('step/end', { turn: 1, step: 1 }) session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) }).not.toThrow() @@ -89,9 +89,9 @@ describe('session-log invariants', () => { const { ctx } = await setup({ freeze: false }) const session = ctx.sessions.create() // No turn open: every message-bearing event must be turn-enclosed (the turn-enclosure RFC). - expect(() => session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } })) + expect(() => session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })) .toThrow(/outside any open turn/) - expect(() => session.append('context/message', { content: [{ type: 'text', text: 'ctx' }], source: { kind: 'user' } })) + expect(() => session.append('context/message', { content: [{ type: 'text', text: 'ctx' }], source: { kind: 'user' } }, { surfaceOp: 'append' })) .toThrow(/outside any open turn/) }) @@ -100,7 +100,7 @@ describe('session-log invariants', () => { const session = ctx.sessions.create() // steering/message is turn-scoped: outside a turn it would land past the // commit boundary and be dropped on resume (the turn-enclosure RFC). - expect(() => session.append('steering/message', { turn: 1, content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })) + expect(() => session.append('steering/message', { turn: 1, content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }, { surfaceOp: 'append' })) .toThrow(/outside any open turn/) // A PLUGIN-added (merge-extensible) event type is caught by the default too. // Cast through `any`: 'compaction/marker' is not in SessionEventType (it's @@ -115,7 +115,7 @@ describe('session-log invariants', () => { const { ctx } = await setup({ freeze: false }) const session = ctx.sessions.create() session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) - expect(() => session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } })) + expect(() => session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })) .not.toThrow() }) @@ -124,7 +124,7 @@ describe('session-log invariants', () => { const session = ctx.sessions.create() session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) session.append('step/start', { turn: 1, step: 1 }) - expect(() => session.append('tool/result', { turn: 1, step: 1, callId: CallId('ghost'), content: [], isError: false })) + expect(() => session.append('tool/result', { turn: 1, step: 1, callId: CallId('ghost'), content: [], isError: false }, { surfaceOp: 'append' })) .toThrow(/no prior tool\/call/) }) @@ -136,7 +136,7 @@ describe('session-log invariants', () => { session.append('step/start', { turn: 1, step: 1 }) session.append('assistant/message', { turn: 1, step: 1, content: [ { type: 'tool-call', id: CallId('crashed'), name: 'bash', arguments: '{}' }, - ] }) + ] }, { surfaceOp: 'append' }) session.append('tool/result', { turn: 1, step: 1, @@ -144,7 +144,7 @@ describe('session-log invariants', () => { content: [{ type: 'text', text: 'interrupted' }], isError: true, error: { name: 'InterruptedError', code: 'interrupted' }, - }) + }, { surfaceOp: 'append' }) session.append('step/end', { turn: 1, step: 1 }) session.append('turn/end', { turn: 1, reason: { kind: 'interrupted' } }) }).not.toThrow() @@ -190,10 +190,10 @@ describe('session-log invariants', () => { expect(() => { session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) session.append('step/start', { turn: 1, step: 1 }) - session.append('assistant/message', { turn: 1, step: 1, content: [] }) + session.append('assistant/message', { turn: 1, step: 1, content: [] }, { surfaceOp: 'append' }) session.append('step/end', { turn: 1, step: 1 }) session.append('step/start', { turn: 1, step: 2 }) - session.append('assistant/message', { turn: 1, step: 2, content: [] }) + session.append('assistant/message', { turn: 1, step: 2, content: [] }, { surfaceOp: 'append' }) session.append('step/end', { turn: 1, step: 2 }) session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) session.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }) @@ -246,7 +246,7 @@ describe('session-log invariants', () => { // step ends with the call unresolved — pendingCalls is cleared. session.append('step/end', { turn: 1, step: 1 }) session.append('step/start', { turn: 1, step: 2 }) - expect(() => session.append('tool/result', { turn: 1, step: 2, callId: CallId('c1'), content: [], isError: false })) + expect(() => session.append('tool/result', { turn: 1, step: 2, callId: CallId('c1'), content: [], isError: false }, { surfaceOp: 'append' })) .toThrow(/no prior tool\/call in this step/) }) @@ -255,7 +255,7 @@ describe('session-log invariants', () => { const session = ctx.sessions.create() session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) session.append('step/start', { turn: 1, step: 1 }) - expect(() => session.append('assistant/message', { turn: 1, step: 2, content: [] })) + expect(() => session.append('assistant/message', { turn: 1, step: 2, content: [] }, { surfaceOp: 'append' })) .toThrow(/open is turn 1\/step 1/) }) }) @@ -287,7 +287,7 @@ describe('dev-freeze', () => { const { ctx } = await setup() // freeze defaults true const session = ctx.sessions.create() session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) - const event = session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }) + const event = session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) expect(Object.isFrozen(event)).toBe(true) expect(Object.isFrozen(event.data)).toBe(true) expect(Object.isFrozen(event.data.content)).toBe(true) @@ -298,7 +298,7 @@ describe('dev-freeze', () => { const { ctx } = await setup({ freeze: false }) const session = ctx.sessions.create() session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) - const event = session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }) + const event = session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) expect(Object.isFrozen(event)).toBe(false) }) @@ -324,7 +324,7 @@ describe('dev-freeze', () => { // the caller's input — read the event back and assert on its data. const innerContent: { type: 'text'; text: string }[] = [{ type: 'text', text: 'inner' }] const block = Object.freeze({ type: 'tool-result' as const, toolCallId: CallId('c1'), content: innerContent, isError: false }) - const event = session.append('user/message', { content: [block], source: { kind: 'user' } }) + const event = session.append('user/message', { content: [block], source: { kind: 'user' } }, { surfaceOp: 'append' }) const logged = event.data.content[0] as { content: { type: 'text'; text: string }[] } expect(Object.isFrozen(logged.content)).toBe(true) expect(Object.isFrozen(logged.content[0])).toBe(true) @@ -424,7 +424,7 @@ describe('HMR safety', () => { const spy = vi.fn() ctx.on('session/event', spy) const session = ctx.sessions.create() - session.append('user/message', { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' } }) + session.append('user/message', { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) // our own spy fires, proving events still flow — but the plugin's frozen. expect(spy).toHaveBeenCalledOnce() expect(Object.isFrozen(session.events[0])).toBe(false)