Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(agent): simplify inbox-driven turn admission

Browse Source
This commit is contained in:
_Kerman
2026-08-02 00:27:37 +08:00
parent d38c8bfaf3
commit dbdf270af0
104 changed files with 550 additions and 511 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts
View File

@@ -64,7 +64,7 @@ function appendClosedTurn(session: Session): void {
content: [{ type: 'text', text: 'hello' }],
source: { kind: 'user' },
}), { surfaceOp: 'append' })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
}
// Run the shared backend contract against the real JSONL backend.
@@ -224,7 +224,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
}),
}, surfaceOp: 'append', sourceEventSeqs: [2, 3] },
{ type: 'step/end', seq: 5, time: 6, data: { turn: 1, step: 1 } },
{ type: 'turn/end', seq: 6, time: 7, data: { turn: 1, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 6, time: 7, data: { turn: 1, step: 1, reason: { kind: 'completed' } } },
]
await ctx.sessionPersistence.create(m)
await ctx.sessionPersistence.append(m.id, log)
@@ -344,7 +344,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
JSON.stringify(toHeaderLine(m)),
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } }),
JSON.stringify({ type: 'request/header-delta', seq: 1, time: 2, data: { config: { model: 'legacy' } } }),
JSON.stringify({ type: 'turn/end', seq: 2, time: 3, data: { turn: 1, reason: { kind: 'completed' } } }),
JSON.stringify({ type: 'turn/end', seq: 2, time: 3, data: { turn: 1, step: 0, reason: { kind: 'completed' } } }),
'',
].join('\n'))
@@ -419,7 +419,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
// The next append continues at seq 10 (the balanced length).
const turn3 = [
{ type: 'turn/start', seq: 10, time: 11, data: { turn: 3 } },
{ type: 'turn/end', seq: 11, time: 12, data: { turn: 3, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 11, time: 12, data: { turn: 3, step: 0, reason: { kind: 'completed' } } },
] as SessionEvent[]
await ctx.sessionPersistence.append(m.id, turn3)
const reloaded = await ctx.sessionPersistence.load(m.id)
@@ -438,7 +438,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
await ctx.sessionPersistence.load(m.id)
await ctx.sessionPersistence.append(m.id, [
{ type: 'turn/start', seq: 6, time: 9, data: { turn: 2 } },
{ type: 'turn/end', seq: 7, time: 10, data: { turn: 2, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 7, time: 10, data: { turn: 2, step: 0, reason: { kind: 'completed' } } },
] as SessionEvent[])
const after = await readFile(rawLogPath(root, undefined, m.id), 'utf8')
// the committed prefix is byte-for-byte intact at the head of the file
@@ -466,7 +466,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
const turn2 = [
{ type: 'turn/start', seq: 6, time: 9, data: { turn: 2 } },
{ type: 'turn/end', seq: 7, time: 10, data: { turn: 2, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 7, time: 10, data: { turn: 2, step: 0, reason: { kind: 'completed' } } },
] as SessionEvent[]
// The append rejects, but the partial bytes are truncated back: the file is
// its pre-append size and the cursor is unchanged.
@@ -529,7 +529,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
mutableHeader(loaded.meta).cwd = '/evil'
await ctx.sessionPersistence.append(m.id, [
{ type: 'turn/start', seq: 6, time: 9, data: { turn: 2 } },
{ type: 'turn/end', seq: 7, time: 10, data: { turn: 2, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 7, time: 10, data: { turn: 2, step: 0, reason: { kind: 'completed' } } },
] as SessionEvent[])
// The append landed in the ORIGINAL /proj log, not beside an /evil path.
const reloaded = await ctx.sessionPersistence.load(m.id)
@@ -606,8 +606,8 @@ describe('SessionPersistenceJsonl: write path (session/event → flush)', () =>
b.append('user/message', createUserMessage({
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' } })
a.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
b.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
await ctx.sessions.flush(a)
await ctx.sessions.flush(b)
@@ -694,7 +694,7 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
JSON.stringify({ type: 'session', version: 0, id: 'g2', createdAt: 1, delegationDepth: 0 }),
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } }),
JSON.stringify({ type: 'step/start', seq: 2, time: 2, data: { turn: 1, step: 1 } }), // gap: missing seq 1
JSON.stringify({ type: 'turn/end', seq: 3, time: 3, data: { turn: 1, reason: { kind: 'completed' } } }),
JSON.stringify({ type: 'turn/end', seq: 3, time: 3, data: { turn: 1, step: 1, reason: { kind: 'completed' } } }),
].join('\n') + '\n'
// A turn/end exists, so the prefix up to it is committed — but it has a hole.
// Truncating it would silently drop committed data → unloadable.
@@ -705,7 +705,7 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
const log = [
JSON.stringify({ type: 'session', version: 0, id: 'c', createdAt: 1, delegationDepth: 0 }),
'{not json', // corrupt, sits in the committed region (a turn/end follows)
JSON.stringify({ type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'completed' } } }),
JSON.stringify({ type: 'turn/end', seq: 1, time: 2, data: { turn: 1, step: 0, reason: { kind: 'completed' } } }),
].join('\n') + '\n'
expect(() => scanLog(Buffer.from(log))).toThrow(/unparsable committed event/)
})
@@ -733,7 +733,7 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
const log = [
JSON.stringify({ type: 'session', version: 0, id: 't', createdAt: 1, delegationDepth: 0 }),
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } }),
JSON.stringify({ type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'completed' } } }),
JSON.stringify({ type: 'turn/end', seq: 1, time: 2, data: { turn: 1, step: 0, reason: { kind: 'completed' } } }),
JSON.stringify({ type: 'step/start', seq: 9, time: 3, data: { turn: 2, step: 1 } }), // gap in uncommitted tail
].join('\n') + '\n'
const { events } = scanLog(Buffer.from(log))
@@ -777,7 +777,7 @@ describe('SessionPersistenceJsonl: default packed chunk rows', () => {
}),
}, surfaceOp: 'append', sourceEventSeqs: [2, 3, 4, 5, 6] },
{ type: 'step/end', seq: 8, time: 9, data: { turn: 1, step: 1 } },
{ type: 'turn/end', seq: 9, time: 10, data: { turn: 1, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 9, time: 10, data: { turn: 1, step: 1, reason: { kind: 'completed' } } },
]
}
@@ -855,7 +855,7 @@ describe('SessionPersistenceJsonl: default packed chunk rows', () => {
JSON.stringify({ type: 'session', version: 0, id: 'rows', createdAt: 1, delegationDepth: 0 }),
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } }),
JSON.stringify({ type: 'text-chunks', seq0: 1, time0: 2, data: { turn: 1, step: 1, index: 0, dt: [1, 1], texts: ['a', 'b', 'c'] } }),
JSON.stringify({ type: 'turn/end', seq: 4, time: 5, data: { turn: 1, reason: { kind: 'completed' } } }),
JSON.stringify({ type: 'turn/end', seq: 4, time: 5, data: { turn: 1, step: 1, reason: { kind: 'completed' } } }),
].join('\n') + '\n'
const { events } = scanLog(Buffer.from(logText))
expect(events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4])
@@ -867,7 +867,7 @@ describe('SessionPersistenceJsonl: default packed chunk rows', () => {
JSON.stringify({ type: 'session', version: 0, id: 'bad-row', createdAt: 1, delegationDepth: 0 }),
// dt arity mismatch — row validation throws, so the line is a committed hole.
JSON.stringify({ type: 'text-chunks', seq0: 0, time0: 1, data: { turn: 1, step: 1, index: 0, dt: [], texts: ['a', 'b'] } }),
JSON.stringify({ type: 'turn/end', seq: 2, time: 3, data: { turn: 1, reason: { kind: 'completed' } } }),
JSON.stringify({ type: 'turn/end', seq: 2, time: 3, data: { turn: 1, step: 1, reason: { kind: 'completed' } } }),
].join('\n') + '\n'
expect(() => scanLog(Buffer.from(logText))).toThrow(/unparsable committed event/)
})
@@ -1154,7 +1154,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
const firstFiber = await ctx.plugin(Object.assign((inner: Context) => {
const a = inner.sessions.create(SessionId('bound'), { meta: { cwd: '/a' } })
a.append('turn/start', { turn: 1 })
a.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
a.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
}, { inject: ['sessions'] }))
for (const s of ctx.sessions.list()) await ctx.sessions.flush(s)
await firstFiber.dispose()
@@ -1232,7 +1232,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
await ctx2.sessionPersistence.append(m.id, [
{ type: 'turn/start', seq: 6, time: 9, data: { turn: 2 } },
{ type: 'turn/end', seq: 7, time: 10, data: { turn: 2, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 7, time: 10, data: { turn: 2, step: 0, reason: { kind: 'completed' } } },
] as SessionEvent[])
const loaded = await ctx2.sessionPersistence.load(m.id)
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
@@ -1282,7 +1282,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
session.append('user/message', createUserMessage({
content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' },
}), { surfaceOp: 'append' })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
// Make the durable materialize fail on the next flush.
const backend = ctx2.sessionPersistence as unknown as { materialize: (...args: unknown[]) => Promise<void> }
const origMat = backend.materialize.bind(backend)

6
packages/session-persistence/session-persistence-jsonl/tests/zstd.spec.ts
View File

@@ -286,7 +286,7 @@ describe('SessionPersistenceJsonl: default Zstandard encoding', () => {
const before = await readFile(path)
const secondTurn = [
{ type: 'turn/start', seq: 6, time: 7, data: { turn: 2 } },
{ type: 'turn/end', seq: 7, time: 8, data: { turn: 2, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 7, time: 8, data: { turn: 2, step: 0, reason: { kind: 'completed' } } },
] as SessionEvent[]
await ctx.sessionPersistence.append(header.id, secondTurn)
@@ -428,7 +428,7 @@ describe('SessionPersistenceJsonl: default Zstandard encoding', () => {
const path = logPath(root, header.cwd, header.id, 'zstd')
const secondTurn = [
{ type: 'turn/start', seq: 6, time: 7, data: { turn: 2 } },
{ type: 'turn/end', seq: 7, time: 8, data: { turn: 2, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 7, time: 8, data: { turn: 2, step: 0, reason: { kind: 'completed' } } },
] as SessionEvent[]
const frame = await compressZstdFrame(secondTurn.map(e => JSON.stringify(e)).join('\n') + '\n')
await appendFile(path, frame.subarray(0, -1))
@@ -476,7 +476,7 @@ describe('SessionPersistenceJsonl: default Zstandard encoding', () => {
})
const secondTurn = [
{ type: 'turn/start', seq: 6, time: 7, data: { turn: 2 } },
{ type: 'turn/end', seq: 7, time: 8, data: { turn: 2, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 7, time: 8, data: { turn: 2, step: 0, reason: { kind: 'completed' } } },
] as SessionEvent[]
await expect(ctx.sessionPersistence.append(header.id, secondTurn)).rejects.toThrow(/simulated Zstandard fsync failure/)
expect(await readFile(path)).toEqual(before)

18
packages/session-persistence/session-persistence-sqlite/tests/sqlite.spec.ts
View File

@@ -135,7 +135,7 @@ describe('scanRows', () => {
const gapped: SessionEvent[] = [
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } },
{ type: 'step/start', seq: 2, time: 2, data: { turn: 1, step: 1 } }, // seq 1 missing
{ type: 'turn/end', seq: 3, time: 3, data: { turn: 1, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 3, time: 3, data: { turn: 1, step: 1, reason: { kind: 'completed' } } },
]
expect(() => scanRows(rows(gapped))).toThrow(/seq gap in committed region/)
})
@@ -143,7 +143,7 @@ describe('scanRows', () => {
it('throws on an unparsable row inside the committed region', () => {
const withCorruptCommitted: EventRow[] = [
{ seq: 0, type: 'turn/start', time: 1, data: '{not json', source_event_seqs: null, surface_op: null }, // corrupt, sits before a turn/end
{ seq: 1, type: 'turn/end', time: 2, data: JSON.stringify({ turn: 1, reason: { kind: 'completed' } }), source_event_seqs: null, surface_op: null },
{ seq: 1, type: 'turn/end', time: 2, data: JSON.stringify({ turn: 1, step: 0, reason: { kind: 'completed' } }), source_event_seqs: null, surface_op: null },
]
expect(() => scanRows(withCorruptCommitted)).toThrow(/unparsable committed event/)
})
@@ -185,7 +185,7 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
const insert = db.prepare('INSERT INTO events (session_id, seq, type, time, data) VALUES (?, ?, ?, ?, ?)')
insert.run(m.id, 0, 'turn/start', 1, JSON.stringify({ turn: 1 }))
insert.run(m.id, 1, 'request/header-delta', 2, JSON.stringify({ config: { model: 'legacy' } }))
insert.run(m.id, 2, 'turn/end', 3, JSON.stringify({ turn: 1, reason: { kind: 'completed' } }))
insert.run(m.id, 2, 'turn/end', 3, JSON.stringify({ turn: 1, step: 0, reason: { kind: 'completed' } }))
db.close()
const mounted = await backend(path)
@@ -252,7 +252,7 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
// length (seq 10) and a reload round-trips identically.
await ctx2.sessionPersistence.append(m.id, [
{ type: 'turn/start', seq: 10, time: 9, data: { turn: 3 } },
{ type: 'turn/end', seq: 11, time: 10, data: { turn: 3, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 11, time: 10, data: { turn: 3, step: 0, reason: { kind: 'completed' } } },
])
const reloaded = await ctx2.sessionPersistence.load(m.id)
expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11])
@@ -478,7 +478,7 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
// load physically deleted the corrupt tail row, so a fresh append continues.
await b2.ctx.sessionPersistence.append(m.id, [
{ type: 'turn/start', seq: 6, time: 8, data: { turn: 2 } },
{ type: 'turn/end', seq: 7, time: 9, data: { turn: 2, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 7, time: 9, data: { turn: 2, step: 0, reason: { kind: 'completed' } } },
])
const reloaded = await b2.ctx.sessionPersistence.load(m.id)
expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
@@ -704,7 +704,7 @@ describe('SessionPersistenceSqlite: edge cases', () => {
await b2.ctx.sessionPersistence.load(m.id) // cursor 6 in b2
const turn2: SessionEvent[] = [
{ type: 'turn/start', seq: 6, time: 7, data: { turn: 2 } },
{ type: 'turn/end', seq: 7, time: 8, data: { turn: 2, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 7, time: 8, data: { turn: 2, step: 0, reason: { kind: 'completed' } } },
]
// b1 commits seq 6..7 first.
await b1.ctx.sessionPersistence.append(m.id, turn2)
@@ -799,7 +799,7 @@ describe('surface field round-trip', () => {
data: JSON.stringify({ content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }),
source_event_seqs: null, surface_op: '{"op":"replace","start":0,"end":0}' },
{ seq: 1, type: 'turn/end', time: 2,
data: JSON.stringify({ turn: 1, reason: { kind: 'completed' } }),
data: JSON.stringify({ turn: 1, step: 0, reason: { kind: 'completed' } }),
source_event_seqs: null, surface_op: null },
]
const { preserved } = scanRows(rows)
@@ -831,7 +831,7 @@ describe('surface field round-trip', () => {
}),
}, { surfaceOp: 'append', sourceEventSeqs: [2] })
session.append('step/end', { turn: 1, step: 1 })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 1, step: 1, reason: { kind: 'completed' } })
await ctx.sessions.flush(session)
const loaded = await ctx.sessionPersistence.load(SessionId('roundtrip-surface'))
expect(loaded.events).toHaveLength(6)
@@ -857,7 +857,7 @@ describe('surface field round-trip', () => {
source: { kind: 'user' },
}),
}, { surfaceOp: 'append' })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
await ctx.sessions.flush(session)
const loaded = await ctx.sessionPersistence.load(SessionId('surface-noseq'))
expect((loaded.events[1]! as SurfaceEvent).surfaceOp).toBe('append')

4
packages/session-persistence/session-persistence/tests/contract.ts
View File

@@ -53,7 +53,7 @@ export function oneTurnLog(): SessionEvent[] {
}),
}, surfaceOp: 'append' },
{ type: 'step/end', seq: 4, time: 5, data: { turn: 1, step: 1 } },
{ type: 'turn/end', seq: 5, time: 6, data: { turn: 1, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 5, time: 6, data: { turn: 1, step: 1, reason: { kind: 'completed' } } },
]
}
@@ -158,7 +158,7 @@ export function runPersistenceContract(name: string, make: () => Promise<Contrac
// the balanced length (seq 10), and a reload round-trips identically.
await persistence.append(m.id, [
{ type: 'turn/start', seq: 10, time: 9, data: { turn: 3 } },
{ type: 'turn/end', seq: 11, time: 10, data: { turn: 3, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 11, time: 10, data: { turn: 3, step: 0, reason: { kind: 'completed' } } },
])
const reloaded = await persistence.load(m.id)
expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11])

36
packages/session-persistence/session-persistence/tests/coordinator-contract.ts
View File

@@ -115,7 +115,7 @@ function legacyMessageLog(): SessionEvent[] {
surfaceOp: { op: 'replace', start: 5, end: 5 },
},
{ type: 'step/end', seq: 8, time: 9, data: { turn: 1, step: 1 } },
{ type: 'turn/end', seq: 9, time: 10, data: { turn: 1, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 9, time: 10, data: { turn: 1, step: 1, reason: { kind: 'completed' } } },
] as unknown as SessionEvent[]
}
@@ -215,7 +215,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
const live = ctx.sessions.create(id, { seed: [start], meta: header })
await expect(loading).rejects.toThrow(/live turn is open/)
live.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
live.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
await ctx.sessions.flush(live)
const loaded = await ctx.sessionPersistence.load(id)
// The constructor's end-seed event persisted between the stored
@@ -303,7 +303,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
expect(() => {
;(ev.data as { content: { type: 'text'; text: string }[] }).content[0]!.text = 'HACKED'
}).toThrow(TypeError)
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 1, step: 1, reason: { kind: 'completed' } })
await ctx.sessions.flush(session)
const loaded = await ctx.sessionPersistence.load(SessionId('mutate'))
@@ -511,7 +511,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
const s2 = second.ctx.sessions.create(SessionId('resumed'), { seed: loaded.events, meta: { cwd: WORK } })
await second.ctx.sessions.flush(s2) // let onCreated adopt
s2.append('turn/start', { turn: 2 })
s2.append('turn/end', { turn: 2, reason: { kind: 'completed' } })
s2.append('turn/end', { turn: 2, step: 0, reason: { kind: 'completed' } })
await second.ctx.sessions.flush(s2)
const reloaded = await second.ctx.sessionPersistence.load(SessionId('resumed'))
@@ -536,7 +536,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
session.append('user/message', createUserMessage({
content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' },
}), { surfaceOp: 'append' })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
const fiber = await fix.mount(ctx)
try {
@@ -560,7 +560,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
session.append('user/message', createUserMessage({
content: [{ type: 'text', text: 'buffered' }], source: { kind: 'user' },
}), { surfaceOp: 'append' })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
// No explicit flush — dispose must drain.
await fiber.dispose()
@@ -588,7 +588,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
session.append('user/message', createUserMessage({
content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' },
}), { surfaceOp: 'append' })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
await ctx.sessions.flush(session)
// Hot-reload: dispose instance 1, mount instance 2 over the same storage while the
@@ -600,7 +600,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
session.append('user/message', createUserMessage({
content: [{ type: 'text', text: 'again' }], source: { kind: 'user' },
}), { surfaceOp: 'append' })
session.append('turn/end', { turn: 2, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 2, step: 0, reason: { kind: 'completed' } })
await expect(ctx.sessions.flush(session)).resolves.not.toThrow()
const loaded = await ctx.sessionPersistence.load(SessionId('hmr-adopt'))
@@ -620,7 +620,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
// Instance 1 flushes turn 1.
const backend1 = await fix.mount(ctx)
session.append('turn/start', { turn: 1 })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
await ctx.sessions.flush(session)
// Append turn 2 to the LIVE session, then dispose instance 1 WITHOUT
@@ -628,7 +628,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
// backend never buffered it via session/event.
await backend1.dispose()
session.append('turn/start', { turn: 2 })
session.append('turn/end', { turn: 2, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 2, step: 0, reason: { kind: 'completed' } })
// Instance 2 adopts the stored prefix (turn 1) and MUST also persist the
// live suffix (turn 2) carried in the session's events.
@@ -661,7 +661,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
// The live session is still the authority: it appends the REAL step/turn
// end. Adoption must truncate the torn tail but NOT synthesize closers.
session.append('step/end', { turn: 1, step: 1 })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 1, step: 1, reason: { kind: 'completed' } })
await ctx.sessions.flush(session)
const loaded = await ctx.sessionPersistence.load(SessionId('hmr-open'))
@@ -721,7 +721,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
}, { inject: ['sessions'] }))
await expect(ctx.sessions.flush(reuse)).resolves.toBeUndefined()
reuse.append('turn/start', { turn: 1 })
reuse.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
reuse.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
await ctx.sessions.flush(reuse)
const loaded = await ctx.sessionPersistence.load(SessionId('abandoned'))
expect(loaded.events.map(e => e.seq)).toEqual([0, 1])
@@ -742,7 +742,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
await ctx.sessions.flush(first)
// Append a turn but do NOT flush — events sit in the write-behind buffer.
first.append('turn/start', { turn: 1 })
first.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
first.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
await firstFiber.dispose()
// Disposal is an observe-only notification. Poll storage rather than
@@ -772,7 +772,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
session.append('user/message', createUserMessage({
content: [{ type: 'text', text: 'x' }], source: { kind: 'user' },
}), { surfaceOp: 'append' })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
await ctx.sessions.flush(session)
// Re-emit session/created for the SAME live session (idempotent initFor).
ctx.emit(scopeTarget(session, undefined), 'session/created', session)
@@ -848,7 +848,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
cont = inner.sessions.create(SessionId('claim'), { seed: [
...events,
{ type: 'turn/start', seq: 6, time: 7, data: { turn: 2 } },
{ type: 'turn/end', seq: 7, time: 8, data: { turn: 2, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 7, time: 8, data: { turn: 2, step: 0, reason: { kind: 'completed' } } },
], meta: { cwd: WORK, createdAt: 2000 } })
}, { inject: ['sessions'] }))
await ctx.sessions.flush(cont)
@@ -940,7 +940,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
try {
await second.ctx.sessionPersistence.append(SessionId('adopt-append'), [
{ type: 'turn/start', seq: 6, time: 7, data: { turn: 2 } },
{ type: 'turn/end', seq: 7, time: 8, data: { turn: 2, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 7, time: 8, data: { turn: 2, step: 0, reason: { kind: 'completed' } } },
])
const loaded = await second.ctx.sessionPersistence.load(SessionId('adopt-append'))
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
@@ -1043,7 +1043,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
session.append('user/message', createUserMessage({
content: [{ type: 'text', text: 'q' }], source: { kind: 'user' },
}), { surfaceOp: 'append' })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
await ctx.sessions.flush(session)
const loaded = await ctx.sessionPersistence.load(SessionId('flush-nostate'))
expect(loaded.events).toHaveLength(3)
@@ -1100,7 +1100,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
// (seq 10) and a reload round-trips identically.
await second.ctx.sessionPersistence.append(SessionId('torn'), [
{ type: 'turn/start', seq: 10, time: 9, data: { turn: 3 } },
{ type: 'turn/end', seq: 11, time: 10, data: { turn: 3, reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 11, time: 10, data: { turn: 3, step: 0, reason: { kind: 'completed' } } },
])
const reloaded = await second.ctx.sessionPersistence.load(SessionId('torn'))
expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11])

18
packages/session-persistence/session-persistence/tests/persistence.spec.ts
View File

@@ -240,7 +240,7 @@ describe('PersistenceCoordinator eager writes', () => {
session.append('turn/start', { turn: 1 })
await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
appendGate.resolve(true)
await vi.waitFor(() => {
@@ -273,7 +273,7 @@ describe('PersistenceCoordinator eager writes', () => {
const session = ctx.sessions.create(SessionId('eager-flush-retry'))
await ctx.sessions.flush(session)
session.append('turn/start', { turn: 1 })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
const barriers = [ctx.sessions.flush(session), ctx.sessions.flush(session)]
@@ -535,7 +535,7 @@ describe('PersistenceCoordinator observation cancellation', () => {
session = inner.sessions.create(id)
}, { inject: ['sessions'] }))
session.append('turn/start', { turn: 1 })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
// Dispose the session so retirement starts; its append is gated, so the
// retirement promise stays pending in the coordinator.
await sessionFiber.dispose()
@@ -678,7 +678,7 @@ describe('PersistenceCoordinator retirement', () => {
await ctx.sessions.flush(first)
backend.beforeAppend = async () => { await appendGate.promise }
first.append('turn/start', { turn: 1 })
first.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
first.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
await firstFiber.dispose()
@@ -718,7 +718,7 @@ describe('PersistenceCoordinator retirement', () => {
await ctx.sessions.flush(first)
backend.beforeAppend = async () => { await appendGate.promise }
first.append('turn/start', { turn: 1 })
first.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
first.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
await firstFiber.dispose()
const baselineLoads = backend.loadAttempts
@@ -782,7 +782,7 @@ describe('PersistenceCoordinator retirement', () => {
type: 'turn/end',
seq: 1,
time: 2,
data: { turn: 1, reason: { kind: 'completed' } },
data: { turn: 1, step: 0, reason: { kind: 'completed' } },
}])
await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
@@ -825,7 +825,7 @@ describe('PersistenceCoordinator retirement', () => {
session = inner.sessions.create(SessionId('retry-retirement'))
}, { inject: ['sessions'] }))
session.append('turn/start', { turn: 1 })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
await sessionFiber.dispose()
await vi.waitFor(() => {
@@ -869,7 +869,7 @@ describe('PersistenceCoordinator retirement', () => {
session = inner.sessions.create(SessionId('inflight-retirement'))
}, { inject: ['sessions'] }))
session.append('turn/start', { turn: 1 })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
await sessionFiber.dispose()
await vi.waitFor(() => {
expect(backend.appendAttempts).toBe(1)
@@ -1054,7 +1054,7 @@ describe('SessionPersistence service registration', () => {
session = inner.sessions.create(SessionId(`disposed-${index}`))
}, { inject: ['sessions'] }))
session.append('turn/start', { turn: 1 })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/end', { turn: 1, step: 0, reason: { kind: 'completed' } })
await ctx.sessions.flush(session)
await sessionFiber.dispose()
}