Merge branch 'feat/acp-1-max-tokens-turn-end' into feat/acp-2-bridge
# Conflicts: # AGENTS.md # docs/cookbook/extension-cookbook.md # yarn.lock
This commit is contained in:
@@ -11,6 +11,6 @@ Naming notes:
|
||||
- Files `src/index.ts` export the service default + all public types
|
||||
- `src/types.ts` contain only types — no runtime code
|
||||
- Tests live at package level under `tests/`, not `src/__tests__/`
|
||||
- A package's README and module/JSDoc comments are part of the change: when you alter behavior (config keys, defaults, error codes, wire fields), update them in the same commit. CI runs `yarn doc-sync`, which typechecks fenced `ts` blocks in `packages/*/README.md` and verifies the event-taxonomy table — but it does NOT cover this file or prose drift (config keys, defaults, error codes), so those stay on the author.
|
||||
- A package's README and module/JSDoc comments are part of the change: when you alter behavior (config keys, defaults, error codes, wire fields), update them in the same commit. CI runs `pnpm run doc-sync`, which typechecks fenced `ts` blocks in `packages/*/README.md` and verifies the event-taxonomy table — but it does NOT cover this file or prose drift (config keys, defaults, error codes), so those stay on the author.
|
||||
|
||||
Read the per-package README.md for package-specific details: service API, events, extension points, TODOs.
|
||||
|
||||
@@ -51,14 +51,14 @@ The JSON-RPC frames go on stdout, so this plugin MUST run in an example that loa
|
||||
|
||||
## Running
|
||||
|
||||
`yarn demo:acp` boots `examples/acp-agent` (needs `DEEPSEEK_API_KEY`). Point an ACP client at it; for Zed, add to `agent_servers`:
|
||||
`pnpm run demo:acp` boots `examples/acp-agent` (needs `DEEPSEEK_API_KEY`). Point an ACP client at it; for Zed, add to `agent_servers`:
|
||||
|
||||
```json
|
||||
{
|
||||
"agent_servers": {
|
||||
"DeepSeek Harness": {
|
||||
"command": "yarn",
|
||||
"args": ["demo:acp"]
|
||||
"command": "pnpm",
|
||||
"args": ["run", "demo:acp"]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -32,14 +32,14 @@
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-agent": "^0.0.1",
|
||||
"@deepseek-ai/dsh-agent-loop": "^0.0.1",
|
||||
"@deepseek-ai/dsh-llm": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session-persistence": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session-persistence-jsonl": "^0.0.1",
|
||||
"@deepseek-ai/dsh-system-prompt": "^0.0.1",
|
||||
"@deepseek-ai/dsh-tools": "^0.0.1",
|
||||
"@deepseek-ai/dsh-agent": "workspace:^",
|
||||
"@deepseek-ai/dsh-agent-loop": "workspace:^",
|
||||
"@deepseek-ai/dsh-llm": "workspace:^",
|
||||
"@deepseek-ai/dsh-session": "workspace:^",
|
||||
"@deepseek-ai/dsh-session-persistence": "workspace:^",
|
||||
"@deepseek-ai/dsh-session-persistence-jsonl": "workspace:^",
|
||||
"@deepseek-ai/dsh-system-prompt": "workspace:^",
|
||||
"@deepseek-ai/dsh-tools": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -32,14 +32,14 @@
|
||||
"schemastery": "^3.18.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-agent": "^0.0.1",
|
||||
"@deepseek-ai/dsh-invariants": "^0.0.1",
|
||||
"@deepseek-ai/dsh-llm": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session-persistence": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session-persistence-jsonl": "^0.0.1",
|
||||
"@deepseek-ai/dsh-system-prompt": "^0.0.1",
|
||||
"@deepseek-ai/dsh-tools": "^0.0.1",
|
||||
"@deepseek-ai/dsh-agent": "workspace:^",
|
||||
"@deepseek-ai/dsh-invariants": "workspace:^",
|
||||
"@deepseek-ai/dsh-llm": "workspace:^",
|
||||
"@deepseek-ai/dsh-session": "workspace:^",
|
||||
"@deepseek-ai/dsh-session-persistence": "workspace:^",
|
||||
"@deepseek-ai/dsh-session-persistence-jsonl": "workspace:^",
|
||||
"@deepseek-ai/dsh-system-prompt": "workspace:^",
|
||||
"@deepseek-ai/dsh-tools": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -17,7 +17,7 @@ import { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { Session } from '@deepseek-ai/dsh-session'
|
||||
import type {} from '@deepseek-ai/dsh-system-prompt'
|
||||
import type {} from '@deepseek-ai/dsh-tools'
|
||||
import type {} from '@deepseek-ai/dsh-session-persistence'
|
||||
import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
|
||||
import { LoopAgent } from './agent.ts'
|
||||
|
||||
export { LoopAgent } from './agent.ts'
|
||||
@@ -32,7 +32,19 @@ declare module 'cordis' {
|
||||
|
||||
export interface Config {
|
||||
/** Agents created from configuration at startup. */
|
||||
agents: (AgentOptions & { id: string })[]
|
||||
agents: (AgentOptions & {
|
||||
id: string
|
||||
/**
|
||||
* If set, the config agent RESUMES this persisted session id instead of
|
||||
* starting a fresh `${id}-session-<uuid>`. Sourced from an env var in
|
||||
* cordis.yml (`resumeSessionId: !!js process.env.RESUME_SESSION_ID`), so a
|
||||
* demo can continue a prior conversation without code changes. Requires a
|
||||
* `dsh-session-persistence` backend; the resume is deferred until that
|
||||
* service is available (via `ctx.inject`) and the loaded session's events
|
||||
* seed the live session so history continues.
|
||||
*/
|
||||
resumeSessionId?: string
|
||||
})[]
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -53,6 +65,7 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
id: z.string().required(),
|
||||
model: z.string(),
|
||||
systemPrompt: z.string(),
|
||||
resumeSessionId: z.string(),
|
||||
})).default([]),
|
||||
})
|
||||
|
||||
@@ -61,8 +74,27 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
// Provide the agent-creation factory to the registry (effect-scoped: the
|
||||
// slot is cleared on dispose).
|
||||
ctx.effect(() => this.ctx.agents.setFactory(this), 'agentLoop.setFactory()')
|
||||
for (const { id, ...options } of config.agents) {
|
||||
this.create(id, options)
|
||||
for (const { id, resumeSessionId, ...options } of config.agents) {
|
||||
if (resumeSessionId !== undefined && resumeSessionId !== '') {
|
||||
// Resume a prior session instead of starting fresh. resume() needs
|
||||
// `ctx.sessionPersistence`, which may load AFTER this plugin (cordis.yml
|
||||
// lists the backend later). `ctx.inject(['sessionPersistence'], cb)`
|
||||
// runs `cb` with a child ctx once the service exists; the child reads
|
||||
// the persistence and hands it to resumeWith (which uses this.ctx — the
|
||||
// parent — for sessions/registry, all in AgentLoop's static inject). A
|
||||
// failed resume is contained + logged: startup must not crash.
|
||||
ctx.effect(() => {
|
||||
const fiber = this.ctx.inject(['sessionPersistence'], (childCtx: Context) => {
|
||||
void this.resumeWith(childCtx.sessionPersistence, { agentId: id, resumeSessionId, agentOptions: options })
|
||||
.catch((error: unknown) => {
|
||||
this.ctx.logger.warn(`agent "${id}": config-driven resume of "${resumeSessionId}" failed: ${String(error)}`)
|
||||
})
|
||||
})
|
||||
return () => void fiber.dispose()
|
||||
}, `agentLoop.resume(${id})`)
|
||||
} else {
|
||||
this.create(id, options)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -120,7 +152,6 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
* by the time this runs the service exists.
|
||||
*/
|
||||
async resume(options: ResumeAgentOptions): Promise<Agent> {
|
||||
this.assertAgentIdFree(options.agentId)
|
||||
const persistence = this.ctx.sessionPersistence
|
||||
// `sessionPersistence` is declaration-merged onto Context as non-optional,
|
||||
// but the service is only present when a backend plugin is loaded — and
|
||||
@@ -130,6 +161,20 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
if (persistence === undefined) {
|
||||
throw new Error('cannot resume: session persistence is not configured (load a dsh-session-persistence backend)')
|
||||
}
|
||||
return this.resumeWith(persistence, options)
|
||||
}
|
||||
|
||||
/**
|
||||
* Resume against an EXPLICIT persistence handle. Factored out of {@link resume}
|
||||
* so the config-driven path can pass the handle it obtained from a
|
||||
* `ctx.inject(['sessionPersistence'], …)` child context: `this.ctx` (the
|
||||
* service's own fiber) did not inject `sessionPersistence`, so reading it
|
||||
* there from inside the inject child trips the cordis inject guard. The
|
||||
* sessions store + registry are still read through `this.ctx` (both are in
|
||||
* AgentLoop's static inject, so they resolve fine).
|
||||
*/
|
||||
private async resumeWith(persistence: SessionPersistence, options: ResumeAgentOptions): Promise<Agent> {
|
||||
this.assertAgentIdFree(options.agentId)
|
||||
const { meta, events } = await persistence.load(SessionId(options.resumeSessionId))
|
||||
// Re-check the agent id AFTER the await: the pre-load check above can go
|
||||
// stale while load() is pending (a concurrent resume/create may register the
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { afterEach, describe, expect, it } from 'vitest'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
@@ -62,4 +62,76 @@ describe('config-driven session id', () => {
|
||||
await waitForIdle(ctx2, a2)
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('config-driven resumeSessionId continues a persisted session (env-var resume)', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'dsh-cfg-resume-'))
|
||||
dirs.push(root)
|
||||
|
||||
// Run 1: a programmatically-created agent on a KNOWN session id persists a
|
||||
// completed turn, so run 2 has a concrete id to resume.
|
||||
const ctx1 = new Context()
|
||||
await ctx1.plugin(LlmService)
|
||||
await ctx1.plugin(SessionStore)
|
||||
await ctx1.plugin(SystemPrompt)
|
||||
await ctx1.plugin(ToolRegistry)
|
||||
await ctx1.plugin(AgentRegistry)
|
||||
await ctx1.plugin(AgentLoop, { agents: [] })
|
||||
await ctx1.plugin(SessionPersistenceJsonl, { root })
|
||||
ctx1.llm.registerAdapter(['mock'], new MockAdapter([textResponse('first')]))
|
||||
const a1 = ctx1.agents.create({ agentId: 'main', sessionId: 'sticky-1' }) as LoopAgent
|
||||
a1.send([{ type: 'text', text: 'remember me' }], { source: { kind: 'user' } })
|
||||
await waitForIdle(ctx1, a1)
|
||||
await ctx1.fiber.dispose()
|
||||
|
||||
// Run 2: a CONFIG agent with resumeSessionId continues that session. The
|
||||
// resume is deferred until sessionPersistence loads (ctx.inject), so wait
|
||||
// for the agent to appear, then assert it is on the resumed id with history.
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(LlmService)
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SystemPrompt)
|
||||
await ctx2.plugin(ToolRegistry)
|
||||
await ctx2.plugin(AgentRegistry)
|
||||
await ctx2.plugin(AgentLoop, { agents: [{ id: 'main', model: 'mock', systemPrompt: '', resumeSessionId: 'sticky-1' }] })
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root })
|
||||
ctx2.llm.registerAdapter(['mock'], new MockAdapter([textResponse('second')]))
|
||||
|
||||
// The deferred resume runs on a microtask after the backend is available.
|
||||
let resumed: LoopAgent | undefined
|
||||
for (let i = 0; i < 50 && !resumed; i++) {
|
||||
await new Promise(r => setTimeout(r, 5))
|
||||
resumed = ctx2.agents.get('main') as LoopAgent | undefined
|
||||
}
|
||||
expect(resumed).toBeDefined()
|
||||
// The live session id IS the resumed id (NOT a fresh ${id}-session-<uuid>),
|
||||
// and the prior turn's user message is in the derived history.
|
||||
expect(resumed!.session.id).toBe('sticky-1')
|
||||
const derived = resumed!.session.deriveMessages()
|
||||
expect(JSON.stringify(derived)).toContain('remember me')
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('config-driven resume of a missing session is contained: logs a warning, no agent, no crash', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'dsh-cfg-resume-miss-'))
|
||||
dirs.push(root)
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SystemPrompt)
|
||||
await ctx.plugin(ToolRegistry)
|
||||
await ctx.plugin(AgentRegistry)
|
||||
await ctx.plugin(AgentLoop, { agents: [{ id: 'main', model: 'mock', systemPrompt: '', resumeSessionId: 'does-not-exist' }] })
|
||||
const warn = vi.spyOn((ctx.agentLoop as unknown as { ctx: { logger: { warn: (...a: unknown[]) => void } } }).ctx.logger, 'warn')
|
||||
.mockImplementation(() => undefined)
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root })
|
||||
ctx.llm.registerAdapter(['mock'], new MockAdapter([textResponse('x')]))
|
||||
|
||||
// The deferred resume fails (no such session on disk). It must be contained:
|
||||
// a warning is logged, no 'main' agent is registered, and the app stays up.
|
||||
await new Promise(r => setTimeout(r, 200))
|
||||
expect(ctx.agents.get('main')).toBeUndefined()
|
||||
expect(warn).toHaveBeenCalledWith(expect.stringContaining('config-driven resume of "does-not-exist" failed'))
|
||||
warn.mockRestore()
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
})
|
||||
|
||||
@@ -25,8 +25,8 @@
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-llm": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session": "^0.0.1",
|
||||
"@deepseek-ai/dsh-llm": "workspace:^",
|
||||
"@deepseek-ai/dsh-session": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -27,7 +27,7 @@
|
||||
"schemastery": "^3.18.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-bash": "^0.0.1",
|
||||
"@deepseek-ai/dsh-bash": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -26,9 +26,9 @@
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-agent": "^0.0.1",
|
||||
"@deepseek-ai/dsh-llm": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session": "^0.0.1",
|
||||
"@deepseek-ai/dsh-agent": "workspace:^",
|
||||
"@deepseek-ai/dsh-llm": "workspace:^",
|
||||
"@deepseek-ai/dsh-session": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -43,4 +43,4 @@ Non-2xx responses throw `LlmError` with stable codes: `AUTH` (401/403), `RATE_LI
|
||||
|
||||
## Testing
|
||||
|
||||
Unit suites run against a local `node:http` mock SSE server (no network). Real-API coverage lives in `tests/adapter.e2e.ts` (`yarn test:e2e`, key-gated): V4 Flash + V4 Pro across thinking enabled/disabled and both official effort levels, including the thinking+tools round trip with reasoning passback.
|
||||
Unit suites run against a local `node:http` mock SSE server (no network). Real-API coverage lives in `tests/adapter.e2e.ts` (`pnpm run test:e2e`, key-gated): V4 Flash + V4 Pro across thinking enabled/disabled and both official effort levels, including the thinking+tools round trip with reasoning passback.
|
||||
|
||||
@@ -27,7 +27,7 @@
|
||||
"schemastery": "^3.18.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-llm": "^0.0.1",
|
||||
"@deepseek-ai/dsh-llm": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -35,4 +35,4 @@ Same MVP contract as llm-deepseek: `prefill` throws `UNSUPPORTED`, images are no
|
||||
|
||||
## Testing
|
||||
|
||||
Unit suites run against a local `node:http` mock SSE server (pi-ai's openai SDK happily talks to any base URL). Real-API coverage in `tests/adapter.e2e.ts` (`yarn test:e2e`, key-gated): V4 Flash + V4 Pro across all exposed reasoning levels (off/high/xhigh), the thinking+tools round trip, and a cross-adapter structural-equivalence check against llm-deepseek.
|
||||
Unit suites run against a local `node:http` mock SSE server (pi-ai's openai SDK happily talks to any base URL). Real-API coverage in `tests/adapter.e2e.ts` (`pnpm run test:e2e`, key-gated): V4 Flash + V4 Pro across all exposed reasoning levels (off/high/xhigh), the thinking+tools round trip, and a cross-adapter structural-equivalence check against llm-deepseek.
|
||||
|
||||
@@ -28,8 +28,8 @@
|
||||
"schemastery": "^3.18.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-llm": "^0.0.1",
|
||||
"@deepseek-ai/dsh-llm-deepseek": "^0.0.1",
|
||||
"@deepseek-ai/dsh-llm": "workspace:^",
|
||||
"@deepseek-ai/dsh-llm-deepseek": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -24,7 +24,7 @@ The JSONL durable session-persistence backend — a concrete `SessionPersistence
|
||||
|
||||
- **Lazy materialization.** `create(meta)` writes nothing; the `.jsonl` (header + first batch) is written atomically (temp-write + `fsync` + rename) on the first `append`. A created-but-never-appended session leaves nothing on disk and is absent from `has`/`list`.
|
||||
- **Append-only.** Committed events (at or below a flushed `turn/end`) are never rewritten. Subsequent appends are line appends at EOF + `fsync`.
|
||||
- **Truncation-repair.** `load` returns events only up to the last complete `turn/end` and records the byte offset of any never-committed crash tail; the first post-load `append` `ftruncate`s to that offset (+ `fsync`) before writing, atomically discarding only the uncommitted tail.
|
||||
- **Crash recovery — close, don't truncate.** A crash can leave a log whose final turn never closed (real events after the last `turn/end`). `load` PRESERVES those events (a turn can be huge — they are real work) and closes the orphaned turn by durably appending synthetic boundary events (a `step/end` if a step was open, then `turn/end {kind:'interrupted'}`), returning a balanced log. Only a never-fully-written **torn tail fragment** (a final line with no newline / unparseable) is `ftruncate`d away before the closers are written. See ADR 0018.
|
||||
- **Contiguous-seq.** `load` rejects a mid-log parse error or `seq` gap (unloadable); `append` rejects a batch whose first `seq` does not continue the stored log, and rejects non-JSON-serializable `event.data` naming the offending event type.
|
||||
- **Format version.** Only v1 is supported; `load` rejects an unknown version. A future format change requires a version bump + migration.
|
||||
|
||||
|
||||
@@ -28,8 +28,8 @@
|
||||
"schemastery": "^3.18.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-session": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session-persistence": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session": "workspace:^",
|
||||
"@deepseek-ai/dsh-session-persistence": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -120,16 +120,23 @@ export function eventLine(event: SessionEvent): string {
|
||||
}
|
||||
|
||||
/**
|
||||
* Compute the byte offset of the END of the last complete `turn/end` line in a
|
||||
* JSONL log buffer (the header line is index 0). Returns the offset to which a
|
||||
* crash tail should be truncated, and the contiguous events up to and including
|
||||
* that `turn/end`. A parse error or a `seq` gap in the MIDDLE (at or before the
|
||||
* last `turn/end`) makes the session unloadable and throws; trailing garbage
|
||||
* AFTER the last `turn/end` is the tolerated crash tail and is excluded.
|
||||
* Parse a JSONL log buffer into its preserved event prefix (the header is line
|
||||
* 0). Returns the longest prefix of complete, seq-contiguous events plus the
|
||||
* byte offset of the end of the last preserved line (`committedBytes`).
|
||||
*
|
||||
* A crash can leave a durable log whose final turn never closed: real,
|
||||
* fully-written events sit after the last `turn/end`. Those are PRESERVED (a
|
||||
* single turn can be huge in a long-horizon task — truncating it would destroy
|
||||
* real work); the backend closes the orphaned open turn with a synthetic
|
||||
* `turn/end {kind:'interrupted'}` on reload (ADR 0018). Only a TORN trailing
|
||||
* fragment — a final line never fully flushed (no newline, unparseable, or a
|
||||
* seq gap) — is excluded; it bounds the preserved region. A parse error or seq
|
||||
* gap AT OR BEFORE the last committed `turn/end` is committed-data corruption
|
||||
* and makes the session unloadable (throws).
|
||||
*
|
||||
* This relies on the session-log invariant that every event lives inside a turn
|
||||
* (`Session.append` enforces it): the last `turn/end` is therefore the last
|
||||
* durable boundary, and nothing committed can sit outside a completed turn.
|
||||
* (`Session.append` enforces it): only the final turn can be open, so the
|
||||
* preserved tail is at most one unclosed turn.
|
||||
*/
|
||||
export function scanLog(buffer: Buffer): { meta: SessionMeta; events: SessionEvent[]; committedBytes: number } {
|
||||
const text = buffer.toString('utf8')
|
||||
@@ -187,38 +194,46 @@ export function scanLog(buffer: Buffer): { meta: SessionMeta; events: SessionEve
|
||||
}
|
||||
})
|
||||
|
||||
// The last index (into eventEntries) that is a valid `turn/end`.
|
||||
// The last index (into eventEntries) that is a valid `turn/end` — the last
|
||||
// fully-committed boundary (the loop flushes only at turn/end).
|
||||
let lastTurnEnd = -1
|
||||
for (let i = parsed.length - 1; i >= 0; i--) {
|
||||
const p = parsed[i]
|
||||
if (p?.ok && p.event?.type === 'turn/end') { lastTurnEnd = i; break }
|
||||
}
|
||||
|
||||
// No committed turn/end anywhere: nothing is committed. The whole event
|
||||
// region is an uncommitted (first-turn) tail — committedBytes is the header.
|
||||
if (lastTurnEnd < 0) {
|
||||
const meta = metaFrom(headerLine)
|
||||
return { meta, events: [], committedBytes: headerEntry.endByte }
|
||||
}
|
||||
|
||||
// Pass 2: the committed prefix [0..lastTurnEnd] must be fully intact and
|
||||
// contiguous (line i is a parsed event with seq === i). A hole or seq gap in
|
||||
// the committed region means committed data was damaged → unloadable.
|
||||
const committed: SessionEvent[] = []
|
||||
for (let i = 0; i <= lastTurnEnd; i++) {
|
||||
// Walk the longest PREFIX of complete, seq-contiguous, parseable event lines
|
||||
// (line i is a parsed event with seq === i). This is the preservable region:
|
||||
// it includes any fully-written events of an interrupted final turn AFTER the
|
||||
// last turn/end — those are real, durably-written work and must NOT be
|
||||
// truncated (a single turn can be huge in a long-horizon task; the orphaned
|
||||
// open turn is closed with a synthetic turn/end on reload, not discarded —
|
||||
// ADR 0018). The walk stops at the first hole (unparseable line or seq gap):
|
||||
// - if that hole is AT OR BEFORE the last committed turn/end, committed data
|
||||
// was damaged → the session is unloadable (throw);
|
||||
// - if it is AFTER (or there is no committed turn/end yet), it is the
|
||||
// tolerated crash boundary — a torn final line never fully flushed — and
|
||||
// it simply bounds the preserved tail.
|
||||
const preserved: SessionEvent[] = []
|
||||
for (let i = 0; i < parsed.length; i++) {
|
||||
const p = parsed[i]
|
||||
if (!p?.ok || p.event === undefined) {
|
||||
throw new Error(`corrupt session log: unparsable committed event at line ${i + 1}`)
|
||||
if (i <= lastTurnEnd) throw new Error(`corrupt session log: unparsable committed event at line ${i + 1}`)
|
||||
break // torn tail fragment after the last turn/end — stop, tolerate
|
||||
}
|
||||
if (p.event.seq !== i) {
|
||||
throw new Error(`corrupt session log: seq gap in committed region at line ${i + 1} (expected ${i}, got ${p.event.seq})`)
|
||||
if (i <= lastTurnEnd) throw new Error(`corrupt session log: seq gap in committed region at line ${i + 1} (expected ${i}, got ${p.event.seq})`)
|
||||
break // gap after the last turn/end — torn tail, stop
|
||||
}
|
||||
committed.push(p.event)
|
||||
preserved.push(p.event)
|
||||
}
|
||||
const lastEntry = parsed[lastTurnEnd]
|
||||
/* v8 ignore next -- lastTurnEnd indexes a parsed entry by construction */
|
||||
const committedBytes = lastEntry ? lastEntry.endByte : headerEntry.endByte
|
||||
return { meta: metaFrom(headerLine), events: committed, committedBytes }
|
||||
|
||||
// committedBytes = end of the last PRESERVED line (header if none): the next
|
||||
// append truncates any torn bytes past this point before writing the
|
||||
// synthetic closers + new events.
|
||||
const lastPreserved = parsed[preserved.length - 1]
|
||||
const committedBytes = preserved.length > 0 && lastPreserved ? lastPreserved.endByte : headerEntry.endByte
|
||||
return { meta: metaFrom(headerLine), events: preserved, committedBytes }
|
||||
}
|
||||
|
||||
/** Build the load-time {@link SessionMeta} from a header line (summary overlaid later). */
|
||||
|
||||
@@ -27,7 +27,7 @@ import { open, mkdir, readFile, readdir, rename, link, rm, truncate } from 'node
|
||||
import { resolve } from 'node:path'
|
||||
import { randomBytes } from 'node:crypto'
|
||||
import { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
|
||||
import { isJsonValue } from '@deepseek-ai/dsh-session'
|
||||
import { isJsonValue, interruptedTurnClosers } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionEvent, SessionId, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session'
|
||||
import {
|
||||
encodeSegment, eventLine, logPath, parseHeaderMeta, scanLog, sessionDir, sidecarPath, toHeaderLine,
|
||||
@@ -58,11 +58,6 @@ interface SessionState {
|
||||
* new session's events to be dropped against the old cursor).
|
||||
*/
|
||||
owner?: Session
|
||||
/**
|
||||
* If a load truncation-repair is pending, the byte offset to truncate the
|
||||
* file to before the next append (discards the never-committed crash tail).
|
||||
*/
|
||||
repairTo?: number
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -238,13 +233,6 @@ export class SessionPersistenceJsonl extends SessionPersistence {
|
||||
let state = this.states.get(id)
|
||||
if (state === undefined) state = await this.adopt(id) // calls loadCore, not load
|
||||
|
||||
// Truncation-repair: on the first append after a load that found a crash
|
||||
// tail, physically discard the orphaned bytes before writing.
|
||||
if (state.repairTo !== undefined) {
|
||||
await this.repair(state, state.repairTo)
|
||||
delete state.repairTo
|
||||
}
|
||||
|
||||
// Contiguity contract: each event's seq must continue the stored log.
|
||||
for (const [i, event] of events.entries()) {
|
||||
if (event.seq !== state.cursor + i) {
|
||||
@@ -280,19 +268,42 @@ export class SessionPersistenceJsonl extends SessionPersistence {
|
||||
const summary = await this.readSidecar(id, meta.cwd)
|
||||
const fullMeta: SessionMeta = { ...meta, ...summary }
|
||||
|
||||
// Record the state so the next append repairs the crash tail (if any) and
|
||||
// continues at the committed length. The state keeps its OWN copy of the
|
||||
// meta; the value returned to the caller is a SEPARATE copy so a consumer
|
||||
// mutating `loaded.meta` (e.g. `cwd`) cannot corrupt the backend's pathing
|
||||
// metadata and send later reads/writes to the wrong log.
|
||||
const needsRepair = committedBytes < buffer.byteLength
|
||||
this.states.set(id, {
|
||||
// Crash-recovery: if the log ended mid-turn (an open turn with real,
|
||||
// preserved events but no closing turn/end), close it durably DURING load so
|
||||
// disk, the returned log, and the cursor all agree — both append routes then
|
||||
// continue with no special-casing. Synthesize the boundary events (a
|
||||
// step/end if a step was open, then a turn/end {kind:'interrupted'}); the
|
||||
// interrupted turn's real events are preserved, never truncated (a turn can
|
||||
// be huge — ADR 0018).
|
||||
const closers = interruptedTurnClosers(events)
|
||||
const balanced = [...events, ...closers]
|
||||
|
||||
// Set state BEFORE the repair writes so they can resolve the log path.
|
||||
const needsTorn = committedBytes < buffer.byteLength
|
||||
const state: SessionState = {
|
||||
meta: { ...fullMeta },
|
||||
cursor: events.length,
|
||||
materialized: true,
|
||||
...needsRepair ? { repairTo: committedBytes } : {},
|
||||
})
|
||||
return { meta: fullMeta, events }
|
||||
}
|
||||
this.states.set(id, state)
|
||||
|
||||
if (needsTorn) {
|
||||
// Discard the torn trailing fragment (a final line never fully flushed)
|
||||
// before writing the closers, so the closers land at a clean EOF.
|
||||
await this.repair(state, committedBytes)
|
||||
}
|
||||
if (closers.length > 0) {
|
||||
// Durably append the synthetic closers, then advance the cursor to the
|
||||
// balanced length. After this, disk == balanced and the next append (live
|
||||
// or direct) continues cleanly. No sidecar touch here: load is not a
|
||||
// summary-changing op (the closers carry no new title/firstPrompt), and
|
||||
// the next real append bumps `updatedAt` — keeping the summary write off
|
||||
// the recovery path avoids a second best-effort failure mode.
|
||||
await this.appendLines(state, closers)
|
||||
state.cursor = balanced.length
|
||||
}
|
||||
|
||||
return { meta: fullMeta, events: balanced }
|
||||
}
|
||||
|
||||
async list(): Promise<SessionMeta[]> {
|
||||
|
||||
@@ -107,36 +107,41 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
expect(loaded.events).toEqual(log) // chunks preserved, contiguous seqs
|
||||
})
|
||||
|
||||
it('crash tolerance: load truncates an uncommitted final turn back to the last turn/end', async () => {
|
||||
it('crash recovery: load preserves the interrupted turn and closes it with a synthetic turn/end {interrupted}', async () => {
|
||||
const m = meta('crash', '/proj')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, oneTurnLog()) // seqs 0..5, turn/end at 5
|
||||
|
||||
// Simulate a crash mid-second-turn: append raw lines that are NOT closed by
|
||||
// a turn/end (and a final partial line with no newline).
|
||||
// a turn/end (turn/start + step/start are fully written), plus a final
|
||||
// partial line with no newline (a torn fragment never fully flushed).
|
||||
const path = logPath(root, '/proj', m.id)
|
||||
const tail = [
|
||||
await writeFile(path, [
|
||||
JSON.stringify({ type: 'turn/start', seq: 6, time: 8, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } }),
|
||||
JSON.stringify({ type: 'step/start', seq: 7, time: 9, data: { turn: 2, step: 1 } }),
|
||||
'{"type":"assistant/chunk","seq":8,"ti', // truncated partial line
|
||||
].join('\n')
|
||||
await writeFile(path, tail, { flag: 'a' })
|
||||
'{"type":"assistant/chunk","seq":8,"ti', // truncated partial line (no newline)
|
||||
].join('\n'), { flag: 'a' })
|
||||
|
||||
// load returns only the committed first turn.
|
||||
// load PRESERVES the interrupted turn's real events (turn/start 6, step/start
|
||||
// 7) — a turn can be huge, so they must not be truncated — and durably closes
|
||||
// the orphaned turn with synthetic step/end (8) + turn/end {interrupted} (9).
|
||||
const loaded = await ctx.sessionPersistence.load(m.id)
|
||||
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5])
|
||||
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
|
||||
const last = loaded.events.at(-1)!
|
||||
expect(last.type === 'turn/end' && last.data.reason).toEqual({ kind: 'interrupted' })
|
||||
const stepEnd = loaded.events[8]!
|
||||
expect(stepEnd.type).toBe('step/end')
|
||||
// the torn seq-8 chunk fragment did not survive
|
||||
expect(loaded.events.some(e => e.type === 'assistant/chunk' && e.seq === 8)).toBe(false)
|
||||
|
||||
// The next append repairs the file (discarding the crash tail) and resumes
|
||||
// at seq 6.
|
||||
const turn2 = [
|
||||
{ type: 'turn/start', seq: 6, time: 10, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'turn/end', seq: 7, time: 11, data: { turn: 2, reason: { kind: 'completed' } } },
|
||||
// The next append continues at seq 10 (the balanced length).
|
||||
const turn3 = [
|
||||
{ type: 'turn/start', seq: 10, time: 11, data: { turn: 3, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'turn/end', seq: 11, time: 12, data: { turn: 3, reason: { kind: 'completed' } } },
|
||||
] as SessionEvent[]
|
||||
await ctx.sessionPersistence.append(m.id, turn2)
|
||||
await ctx.sessionPersistence.append(m.id, turn3)
|
||||
const reloaded = await ctx.sessionPersistence.load(m.id)
|
||||
expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
|
||||
// and no orphaned seq-8 chunk survived
|
||||
expect(reloaded.events.some(e => e.seq === 8)).toBe(false)
|
||||
expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11])
|
||||
})
|
||||
|
||||
it('committed events are never rewritten: only the crash tail is repaired', async () => {
|
||||
@@ -490,15 +495,17 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
|
||||
expect(() => scanLog(Buffer.from('{"type":"event"}\n'))).toThrow(/session header/)
|
||||
})
|
||||
|
||||
it('a seq gap with NO committed turn/end yields zero committed events (uncommitted tail)', () => {
|
||||
it('a seq gap after the last turn/end bounds the preserved tail (torn fragment tolerated)', () => {
|
||||
const log = [
|
||||
JSON.stringify({ type: 'session', version: 1, id: 'g', createdAt: 1 }),
|
||||
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
|
||||
JSON.stringify({ type: 'step/start', seq: 2, time: 2, data: { turn: 1, step: 1 } }), // gap: missing seq 1
|
||||
].join('\n') + '\n'
|
||||
// Nothing reached a turn/end, so nothing is committed — the whole region is
|
||||
// an uncommitted (crash) tail. Safe to load as empty, NOT a corruption.
|
||||
expect(scanLog(Buffer.from(log)).events).toEqual([])
|
||||
// No committed turn/end, so the gap is a tolerated crash boundary: scanLog
|
||||
// PRESERVES the contiguous prefix (turn/start seq 0) — real interrupted-turn
|
||||
// work, not discarded — and stops at the gap. The orphaned open turn is
|
||||
// closed by loadCore's synthetic turn/end, not here.
|
||||
expect(scanLog(Buffer.from(log)).events.map(e => e.seq)).toEqual([0])
|
||||
})
|
||||
|
||||
it('rejects a seq gap BEFORE a later committed turn/end (committed data damaged)', () => {
|
||||
@@ -522,13 +529,23 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
|
||||
expect(() => scanLog(Buffer.from(log))).toThrow(/unparsable committed event/)
|
||||
})
|
||||
|
||||
it('a corrupt line with NO committed turn/end yields zero committed events', () => {
|
||||
it('a header-only log (no event lines at all) preserves nothing — committedBytes is the header', () => {
|
||||
const log = JSON.stringify({ type: 'session', version: 1, id: 'h0', createdAt: 1 }) + '\n'
|
||||
const scanned = scanLog(Buffer.from(log))
|
||||
expect(scanned.events).toEqual([])
|
||||
// committedBytes falls back to the header line's end (no preserved events).
|
||||
expect(scanned.committedBytes).toBe(Buffer.byteLength(log, 'utf8'))
|
||||
})
|
||||
|
||||
it('a corrupt line after the last turn/end bounds the preserved tail', () => {
|
||||
const log = [
|
||||
JSON.stringify({ type: 'session', version: 1, id: 'c2', createdAt: 1 }),
|
||||
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
|
||||
'{not json', // corrupt crash fragment, no turn/end committed
|
||||
].join('\n') + '\n'
|
||||
expect(scanLog(Buffer.from(log)).events).toEqual([])
|
||||
// The contiguous prefix (turn/start seq 0) is preserved; the corrupt
|
||||
// fragment after it is the tolerated crash boundary.
|
||||
expect(scanLog(Buffer.from(log)).events.map(e => e.seq)).toEqual([0])
|
||||
})
|
||||
|
||||
it('tolerates a seq gap AFTER a turn/end (uncommitted tail)', () => {
|
||||
@@ -1066,13 +1083,19 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('a header-only log (no turn/end) loads as zero committed events', () => {
|
||||
const log = [
|
||||
JSON.stringify({ type: 'session', version: 1, id: 'open', createdAt: 1 }),
|
||||
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
|
||||
].join('\n') + '\n'
|
||||
const { events } = scanLog(Buffer.from(log))
|
||||
expect(events).toEqual([]) // nothing committed (no turn/end)
|
||||
it('a header-only log (open turn, no turn/end) preserves the open turn on load and closes it', async () => {
|
||||
// A session whose only durable content is an unclosed first turn. scanLog
|
||||
// preserves the turn/start; loadCore closes it with a synthetic
|
||||
// turn/end {interrupted} so the returned log is balanced.
|
||||
const m = meta('open-turn', '/h')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, [
|
||||
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
] as SessionEvent[])
|
||||
const { events } = await ctx.sessionPersistence.load(m.id)
|
||||
expect(events.map(e => e.type)).toEqual(['turn/start', 'turn/end'])
|
||||
const end = events[1]!
|
||||
expect(end.type === 'turn/end' && end.data.reason).toEqual({ kind: 'interrupted' })
|
||||
})
|
||||
|
||||
it('initFor is idempotent: a re-seeded existing session is not re-initialized', async () => {
|
||||
|
||||
@@ -1,18 +1,20 @@
|
||||
# @deepseek-ai/dsh-session-persistence-sqlite
|
||||
|
||||
A SQLite durable session-persistence backend — a second `SessionPersistence` implementation ([ADR 0016](../../docs/adr/0016-session-persistence.md)), built to validate that the abstract seam and the shared `runPersistenceContract` suite are genuinely backend-agnostic. It satisfies the SAME contract as `dsh-session-persistence-jsonl` (append-only, contiguous-seq, lazy materialization, crash-tail-on-load), expressed over `node:sqlite` rows instead of file bytes.
|
||||
A SQLite durable session-persistence backend — a second `SessionPersistence` implementation ([ADR 0018](../../docs/adr/0018-session-persistence.md)), built to validate that the abstract seam and the shared `runPersistenceContract` suite are genuinely backend-agnostic. It satisfies the SAME contract as `dsh-session-persistence-jsonl` (append-only, contiguous-seq, lazy materialization, interrupted-turn close on load), expressed over `node:sqlite` rows instead of file bytes.
|
||||
|
||||
> **TODO:** this backend talks to `node:sqlite` directly. If a cordis database service (`cordis/db` / a `@cordisjs` SQL driver plugin) is adopted, route through that instead of holding a raw `DatabaseSync` here — the contract surface (`SessionPersistence`) would not change, only the storage driver.
|
||||
|
||||
## Storage model
|
||||
|
||||
Each `SessionEvent` maps 1:1 onto a row in an `events` table `(session_id, seq, type, time, data)` — `data` is the event payload as JSON text, so the row shape is the event verbatim (including `assistant/chunk`, keeping `seq` contiguous). Out-of-log metadata (`SessionMeta`) lives in a `sessions` row, including the mutable `SessionSummary` fields (`updatedAt`, `title`, `firstPrompt`) that `update()` rewrites without touching the event log.
|
||||
Each `SessionEvent` maps 1:1 onto a row in an `events` table `(session_id, seq, type, time, data)` — `data` is the event payload as JSON text, so the row shape is the event verbatim (including `assistant/chunk`, keeping `seq` contiguous). Out-of-log metadata (`SessionMeta`) lives in a `sessions` row, including the mutable `SessionSummary` fields (`updatedAt`, `title`, `firstPrompt`) that `update()` rewrites without touching the event log. A `sessions` row is written only by the first `append` — its existence is the lazy-materialization signal (`has`/`list` report exactly the sessions that have a row), so no separate column is needed.
|
||||
|
||||
`node:sqlite` requires Node ≥ 22.5 (this repo runs Node ≥ 24); the database opens with `foreign_keys = ON` (so `ON DELETE CASCADE` drops a session's events with its row) and `journal_mode = WAL`. The table-layout version is stored in `PRAGMA user_version` and checked on open: a fresh database is stamped with the current `SCHEMA_VERSION`; a database written by a newer, incompatible build (higher `user_version`) is rejected rather than opened against an unknown layout.
|
||||
The repo targets Node ≥ 24 (the root `engines` field), which includes the stable `node:sqlite` module. The database opens with `foreign_keys = ON` (so `ON DELETE CASCADE` drops a session's events with its row) and `journal_mode = WAL`. The table-layout version is stored in `PRAGMA user_version` and checked on open: a fresh database is stamped with the current `SCHEMA_VERSION`; a database written by a newer, incompatible build (higher `user_version`) is rejected rather than opened against an unknown layout.
|
||||
|
||||
## Contract semantics over rows
|
||||
|
||||
- **Append = a transaction.** `append` runs `BEGIN`/`COMMIT` around the batch: it runs any deferred crash-tail repair, materializes the `sessions` row (if still lazy), and INSERTs every event, asserting the contiguous-seq contract first (the first event's `seq` must equal the stored next-seq). A mid-batch failure (a UNIQUE violation on a duplicated seq) rolls back entirely, so the stored log and the in-memory cursor stay consistent.
|
||||
- **Lazy materialization.** `create()` records intent in memory only — no row is written until the first `append`. A created-but-never-appended session is absent from `has()`/`list()` (a `materialized` flag on the row, set inside the first append transaction; `has`/`list` filter to materialized rows).
|
||||
- **Crash-tail-on-load.** `load()` reads every stored event ordered by `seq` and returns only the prefix through the **last complete `turn/end`** (the `SessionPersistence.load` contract), computed from the `seq`/`type` columns so a malformed `data` in the uncommitted tail is never parsed. A batch that landed without its closing `turn/end` (a process killed mid-turn) is an uncommitted tail: `load()` stays non-mutating w.r.t. the event log and records a repair point; the **next `append`** physically DELETEs the orphaned rows inside its transaction (the one-time truncation-repair, matching the JSONL backend and the abstract contract). A `seq` gap inside the committed region makes the session unloadable. If the discarded tail was the session's only committed content, `load()` also flips the metadata row's `materialized` flag to 0 so `has()`/`list()` immediately stop reporting the now-empty session.
|
||||
- **Append = a transaction.** `append` runs `BEGIN`/`COMMIT` around the batch: it materializes the `sessions` row (if still lazy) and INSERTs every event, asserting the contiguous-seq contract first (the first event's `seq` must equal the stored next-seq). A mid-batch failure (a UNIQUE violation on a duplicated seq) rolls back entirely, so the stored log and the in-memory cursor stay consistent. (`load()` already balanced the stored log, so `append` never has to repair a crash tail.)
|
||||
- **Lazy materialization.** `create()` records intent in memory only — no row is written until the first `append`. A created-but-never-appended session has no `sessions` row, so it is absent from `has()`/`list()` (which report exactly the sessions that have a row).
|
||||
- **Interrupted-turn close on load.** `load()` reads every stored event ordered by `seq` and finds the longest seq-contiguous, parseable prefix — INCLUDING the real events of an interrupted final turn after the last `turn/end` (the loop only flushes at `turn/end`, so a process killed mid-turn leaves real, fully-written rows past it). A single turn can be huge in a long-horizon task, so those events are **preserved, never truncated**: `load()` CLOSES the orphaned turn by durably appending the minimal synthetic boundary events (a `step/end` if a step was open, then a `turn/end` carrying `{ kind: 'interrupted' }`), inside one transaction that also DELETEs any never-fully-written torn tail row. `load()` is therefore mutating — after it the stored rows are balanced and the cursor is truthful, so the next `append` continues cleanly. The boundary (last `turn/end`, torn-tail detection) is computed from the `seq`/`type` columns so a malformed `data` in a torn tail row is never parsed (discarded, not unloadable). A parse error or `seq` gap inside the committed region (at or before the last real `turn/end`) makes the session unloadable. A session whose only turn never closed keeps its metadata row and stays present in `has()`/`list()` — the same as the JSONL backend, whose file likewise survives a first append that never reached `turn/end`.
|
||||
|
||||
## Configuration (schemastery)
|
||||
|
||||
|
||||
@@ -28,8 +28,8 @@
|
||||
"schemastery": "^3.18.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-session": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session-persistence": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session": "workspace:^",
|
||||
"@deepseek-ai/dsh-session-persistence": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,10 +4,10 @@
|
||||
* A SECOND {@link SessionPersistence} implementation, built to validate that
|
||||
* the abstract seam + the shared `runPersistenceContract` suite are genuinely
|
||||
* backend-agnostic: the same append-only / contiguous-seq / lazy-materialization
|
||||
* / crash-tail-on-load semantics the JSONL backend expresses over file bytes,
|
||||
* expressed here over `node:sqlite` rows. Each `SessionEvent` maps 1:1 onto a
|
||||
* row `(session_id, seq, type, time, data)`; `append` is an INSERT inside a
|
||||
* transaction that asserts the contiguous-seq contract; the mutable
|
||||
* / interrupted-turn-close-on-load semantics the JSONL backend expresses over
|
||||
* file bytes, expressed here over `node:sqlite` rows. Each `SessionEvent` maps
|
||||
* 1:1 onto a row `(session_id, seq, type, time, data)`; `append` is an INSERT
|
||||
* inside a transaction that asserts the contiguous-seq contract; the mutable
|
||||
* `SessionSummary` lives in the `sessions` metadata row.
|
||||
*
|
||||
* Like the JSONL backend it is also the write-path plugin: it installs the
|
||||
@@ -25,10 +25,10 @@ import { DatabaseSync } from 'node:sqlite'
|
||||
import { mkdir } from 'node:fs/promises'
|
||||
import { dirname, resolve } from 'node:path'
|
||||
import { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
|
||||
import { isJsonValue } from '@deepseek-ai/dsh-session'
|
||||
import { isJsonValue, interruptedTurnClosers } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionEvent, SessionId, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session'
|
||||
import {
|
||||
cutAtLastTurnEnd, openDatabase, rowToEvent, rowToMeta, type EventRow, type SessionRow,
|
||||
openDatabase, rowToMeta, scanRows, type EventRow, type SessionRow,
|
||||
} from './schema.ts'
|
||||
|
||||
export { SCHEMA_VERSION } from './schema.ts'
|
||||
@@ -50,14 +50,6 @@ interface SessionState {
|
||||
cursor: number
|
||||
/** Whether the session has at least one persisted event (materialized). */
|
||||
materialized: boolean
|
||||
/**
|
||||
* If a load found a crash tail, the seq from which the next {@link append}
|
||||
* must DELETE before inserting (the one-time truncation-repair). load() stays
|
||||
* non-mutating w.r.t. the event log — it only records this marker — so the
|
||||
* public contract matches the JSONL backend: load returns the committed
|
||||
* prefix; the subsequent append performs the physical repair.
|
||||
*/
|
||||
repairFrom?: number
|
||||
/** The live Session that owns this state (collision detection); see onCreated. */
|
||||
owner?: Session
|
||||
}
|
||||
@@ -179,24 +171,19 @@ export class SessionPersistenceSqlite extends SessionPersistence {
|
||||
}
|
||||
}
|
||||
|
||||
// The transaction is the durability + atomicity boundary: run any deferred
|
||||
// crash-tail repair, materialize the sessions row (if lazy), and INSERT
|
||||
// every event, or roll back entirely. A BEGIN/COMMIT around the batch means
|
||||
// a mid-batch failure (a UNIQUE violation on a duplicated seq from a
|
||||
// concurrent writer) leaves the stored log untouched, so the cursor stays
|
||||
// truthful and a retry is clean.
|
||||
// The transaction is the durability + atomicity boundary: materialize the
|
||||
// sessions row (if lazy) and INSERT every event, or roll back entirely. A
|
||||
// BEGIN/COMMIT around the batch means a mid-batch failure (a UNIQUE
|
||||
// violation on a duplicated seq from a concurrent writer) leaves the stored
|
||||
// log untouched, so the cursor stays truthful and a retry is clean. (A crash
|
||||
// tail is already gone: load() physically deletes the torn fragment and
|
||||
// durably closes the interrupted turn before returning, so by the time any
|
||||
// append runs the stored log is balanced and contiguous.)
|
||||
const insertEvent = this.db.prepare(
|
||||
'INSERT INTO events (session_id, seq, type, time, data) VALUES (?, ?, ?, ?, ?)',
|
||||
)
|
||||
this.db.exec('BEGIN')
|
||||
try {
|
||||
// One-time truncation-repair: a prior load() found a crash tail and
|
||||
// deferred its physical removal to here (load stays non-mutating). DELETE
|
||||
// the orphaned rows (seq >= repairFrom) before inserting, inside the same
|
||||
// transaction, so the repair + first new append commit atomically.
|
||||
if (state.repairFrom !== undefined) {
|
||||
this.db.prepare('DELETE FROM events WHERE session_id = ? AND seq >= ?').run(id, state.repairFrom)
|
||||
}
|
||||
if (!state.materialized) this.writeRow(state.meta)
|
||||
for (const event of events) {
|
||||
insertEvent.run(id, event.seq, event.type, event.time, JSON.stringify(event.data))
|
||||
@@ -210,7 +197,6 @@ export class SessionPersistenceSqlite extends SessionPersistence {
|
||||
this.db.exec('ROLLBACK')
|
||||
throw error
|
||||
}
|
||||
delete state.repairFrom
|
||||
state.materialized = true
|
||||
state.cursor += events.length
|
||||
}
|
||||
@@ -226,57 +212,77 @@ export class SessionPersistenceSqlite extends SessionPersistence {
|
||||
const meta = rowToMeta(row)
|
||||
this.assertVersion(meta)
|
||||
|
||||
// Read every stored row ordered by seq, then cut at the last complete
|
||||
// turn/end — the same crash-tail semantics as the JSONL backend. The cut is
|
||||
// computed from seq+type COLUMNS only, so a malformed `data` in the
|
||||
// uncommitted tail is discarded (not unloadable); only `data` in the
|
||||
// COMMITTED prefix is parsed (rowToEvent), where a parse error correctly
|
||||
// surfaces. A row that landed without its closing turn/end is an
|
||||
// uncommitted tail and is excluded; a seq gap in the committed region makes
|
||||
// the session unloadable (cutAtLastTurnEnd throws).
|
||||
// Read every stored row ordered by seq, then scan for the preserved prefix:
|
||||
// the longest seq-contiguous, parseable run, INCLUDING the real events of an
|
||||
// interrupted final turn after the last turn/end (a turn can be huge — they
|
||||
// are never truncated). scanRows works off the seq+type COLUMNS for the
|
||||
// last-turn/end boundary, so a malformed `data` in a torn tail row is
|
||||
// discarded (not unloadable); only a parse error / seq gap in the COMMITTED
|
||||
// region (at or before the last turn/end) throws (genuine corruption).
|
||||
const eventRows = this.db
|
||||
.prepare('SELECT seq, type, time, data FROM events WHERE session_id = ? ORDER BY seq')
|
||||
.all(id) as unknown as EventRow[]
|
||||
const { committed, cutTail } = cutAtLastTurnEnd(eventRows)
|
||||
const events = committed.map(rowToEvent)
|
||||
const { preserved, tornFrom } = scanRows(eventRows)
|
||||
|
||||
// Do NOT delete the crash tail here: load() stays non-mutating w.r.t. the
|
||||
// event log, matching the abstract contract and the JSONL backend (load
|
||||
// returns the committed prefix; the next append performs the one-time
|
||||
// physical repair). Record the repair point so the next appendCore DELETEs
|
||||
// the orphaned tail inside its own transaction before inserting.
|
||||
const materialized = committed.length > 0
|
||||
if (committed.length === 0 && row.materialized === 1) {
|
||||
// All-tail discard: the only committed events were a crash tail, so the
|
||||
// session now has NO committed events. The metadata row, however, still
|
||||
// reads materialized = 1 from the prior append — which would make has()
|
||||
// and list() report a session that load() just emptied. Correct the
|
||||
// materialized FLAG (metadata, not the event log) so has()/list() are
|
||||
// immediately consistent. The orphaned tail rows are still removed by the
|
||||
// deferred repair on the next append.
|
||||
this.db.prepare('UPDATE sessions SET materialized = 0 WHERE id = ?').run(id)
|
||||
// Crash-recovery (mutating load, same as the JSONL backend): if the log ended
|
||||
// mid-turn, close it DURING load so disk, the returned log, and the cursor all
|
||||
// agree — both append routes then continue with no special-casing. Synthesize
|
||||
// the boundary events (a step/end if a step was open, then a
|
||||
// turn/end {kind:'interrupted'}); the interrupted turn's real events are
|
||||
// preserved, never truncated (ADR 0018).
|
||||
const closers = interruptedTurnClosers(preserved)
|
||||
const balanced = [...preserved, ...closers]
|
||||
|
||||
// Physically repair the stored log inside one transaction: DELETE the torn
|
||||
// tail fragment (if any), then INSERT the synthetic closers. After COMMIT the
|
||||
// stored rows == balanced, so the cursor is truthful and the next append
|
||||
// continues cleanly with no deferred repair. The metadata row stays as-is
|
||||
// even when preserved.length === 0 (an all-tail crash): the session WAS
|
||||
// materialized by the partial append, so has()/list() still report it — the
|
||||
// same as the JSONL backend, whose file likewise survives a first append that
|
||||
// never reached turn/end.
|
||||
if (tornFrom !== undefined || closers.length > 0) {
|
||||
this.db.exec('BEGIN')
|
||||
try {
|
||||
if (tornFrom !== undefined) {
|
||||
this.db.prepare('DELETE FROM events WHERE session_id = ? AND seq >= ?').run(id, tornFrom)
|
||||
}
|
||||
if (closers.length > 0) {
|
||||
const insertEvent = this.db.prepare('INSERT INTO events (session_id, seq, type, time, data) VALUES (?, ?, ?, ?, ?)')
|
||||
for (const event of closers) {
|
||||
insertEvent.run(id, event.seq, event.type, event.time, JSON.stringify(event.data))
|
||||
}
|
||||
}
|
||||
this.db.exec('COMMIT')
|
||||
} catch (error) {
|
||||
// The DELETE+INSERT cannot collide (a row at a closer's seq is preserved
|
||||
// or deleted as torn first); this rolls back a DB-level failure (disk
|
||||
// full, etc.), unreachable in test.
|
||||
/* v8 ignore start */
|
||||
this.db.exec('ROLLBACK')
|
||||
throw error
|
||||
/* v8 ignore stop */
|
||||
}
|
||||
}
|
||||
|
||||
// Record state so a later append continues at the committed length and runs
|
||||
// the deferred tail repair. The state keeps its OWN copy of the meta; the
|
||||
// returned value is separate so a consumer mutating loaded.meta cannot
|
||||
// corrupt the backend's row metadata.
|
||||
// Record state at the balanced length. The state keeps its OWN copy of the
|
||||
// meta; the returned value is separate so a consumer mutating loaded.meta
|
||||
// cannot corrupt the backend's row metadata.
|
||||
this.states.set(id, {
|
||||
meta: { ...meta },
|
||||
cursor: committed.length,
|
||||
materialized,
|
||||
...cutTail ? { repairFrom: committed.length } : {},
|
||||
cursor: balanced.length,
|
||||
materialized: true,
|
||||
})
|
||||
return { meta, events }
|
||||
return { meta, events: balanced }
|
||||
}
|
||||
|
||||
async list(): Promise<SessionMeta[]> {
|
||||
await this.ready
|
||||
// Materialized rows only: a created-but-never-appended (lazy) session has no
|
||||
// row at all, and a load that cut every event back to zero leaves
|
||||
// materialized = 0. Both are excluded, matching has().
|
||||
// Every metadata row is a materialized session: the row is written only by
|
||||
// the first append (a created-but-never-appended session has no row), so
|
||||
// listing all rows is exactly the materialized set.
|
||||
const rows = this.db
|
||||
.prepare('SELECT * FROM sessions WHERE materialized = 1')
|
||||
.prepare('SELECT * FROM sessions')
|
||||
.all() as unknown as SessionRow[]
|
||||
return rows.map(rowToMeta)
|
||||
}
|
||||
@@ -285,8 +291,8 @@ export class SessionPersistenceSqlite extends SessionPersistence {
|
||||
await this.ready
|
||||
const state = this.states.get(id)
|
||||
if (state?.materialized) return true
|
||||
const row = this.rowFor(id)
|
||||
return row !== undefined && row.materialized === 1
|
||||
// A metadata row exists iff the session was materialized by a first append.
|
||||
return this.rowFor(id) !== undefined
|
||||
}
|
||||
|
||||
delete(id: SessionId): Promise<void> {
|
||||
@@ -326,15 +332,15 @@ export class SessionPersistenceSqlite extends SessionPersistence {
|
||||
}
|
||||
|
||||
/**
|
||||
* Insert-or-replace a session's metadata row, marked materialized. The only
|
||||
* callers are the first materializing `append` and a post-materialization
|
||||
* `update` — a row is written only once a session has durable events, so
|
||||
* `materialized` is always 1 (a never-appended session has no row at all).
|
||||
* Insert-or-replace a session's metadata row. The only callers are the first
|
||||
* materializing `append` and a post-materialization `update`, so writing the
|
||||
* row IS the materialization (its existence is the signal `has`/`list` read);
|
||||
* a never-appended session has no row at all.
|
||||
*/
|
||||
private writeRow(meta: SessionMeta): void {
|
||||
this.db.prepare(`
|
||||
INSERT INTO sessions (id, version, created_at, cwd, parent_session, updated_at, title, first_prompt, materialized)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1)
|
||||
INSERT INTO sessions (id, version, created_at, cwd, parent_session, updated_at, title, first_prompt)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
version = excluded.version,
|
||||
created_at = excluded.created_at,
|
||||
@@ -342,8 +348,7 @@ export class SessionPersistenceSqlite extends SessionPersistence {
|
||||
parent_session = excluded.parent_session,
|
||||
updated_at = excluded.updated_at,
|
||||
title = excluded.title,
|
||||
first_prompt = excluded.first_prompt,
|
||||
materialized = excluded.materialized
|
||||
first_prompt = excluded.first_prompt
|
||||
`).run(
|
||||
meta.id,
|
||||
meta.version,
|
||||
@@ -466,7 +471,7 @@ export class SessionPersistenceSqlite extends SessionPersistence {
|
||||
}
|
||||
|
||||
const row = this.rowFor(id)
|
||||
if (row !== undefined && row.materialized === 1) {
|
||||
if (row !== undefined) {
|
||||
const stored = this.eventsFor(id)
|
||||
if (!seedCoversPrefix(seed, stored)) {
|
||||
throw new Error(`session "${id}" already has a persisted log that does not match this live session (id collision)`)
|
||||
@@ -489,14 +494,17 @@ export class SessionPersistenceSqlite extends SessionPersistence {
|
||||
if (seed.length > 0) await this.append(id, seed)
|
||||
}
|
||||
|
||||
/** The committed events for a session id (last-turn/end cut applied). */
|
||||
/** The preserved events for a session id (torn tail excluded, turn NOT yet closed). */
|
||||
private eventsFor(id: SessionId): SessionEvent[] {
|
||||
const rows = this.db
|
||||
.prepare('SELECT seq, type, time, data FROM events WHERE session_id = ? ORDER BY seq')
|
||||
.all(id) as unknown as EventRow[]
|
||||
// Cut on seq+type columns, then parse `data` only for the committed prefix
|
||||
// (a malformed tail must not throw here — same as loadCore).
|
||||
return cutAtLastTurnEnd(rows).committed.map(rowToEvent)
|
||||
// Scan on seq+type columns, parsing `data` only for the preserved prefix (a
|
||||
// malformed torn tail must not throw here — same as loadCore). Returns the
|
||||
// preserved events WITHOUT the synthetic closers, so a collision check
|
||||
// compares a live seed against the real on-disk events, mirroring the JSONL
|
||||
// backend's scanLog use in onCreated.
|
||||
return scanRows(rows).preserved
|
||||
}
|
||||
|
||||
/** Whether a live session's seed reproduces the first `cursor` stored events. */
|
||||
|
||||
@@ -18,10 +18,11 @@ import type { SessionEvent, SessionId, SessionMeta } from '@deepseek-ai/dsh-sess
|
||||
export const SCHEMA_VERSION = 1
|
||||
|
||||
/**
|
||||
* A row of the `sessions` table — the out-of-log metadata (`SessionMeta`) plus
|
||||
* the `materialized` flag that implements lazy materialization (a created-but-
|
||||
* never-appended session has `materialized = 0` and is excluded from
|
||||
* `has`/`list`, mirroring the JSONL backend's "no file until first append").
|
||||
* A row of the `sessions` table — the out-of-log metadata (`SessionMeta`). The
|
||||
* row's EXISTENCE is the materialization signal: it is written only by the
|
||||
* first `append` (lazy materialization), so a created-but-never-appended
|
||||
* session has no row and is absent from `has`/`list`, mirroring the JSONL
|
||||
* backend's "no file until first append".
|
||||
*/
|
||||
export interface SessionRow {
|
||||
id: string
|
||||
@@ -32,7 +33,6 @@ export interface SessionRow {
|
||||
updated_at: number
|
||||
title: string | null
|
||||
first_prompt: string | null
|
||||
materialized: number
|
||||
}
|
||||
|
||||
/** An `events` table row: one `SessionEvent` mapped 1:1 (`data` is JSON text). */
|
||||
@@ -81,8 +81,7 @@ export function openDatabase(path: string): DatabaseSync {
|
||||
parent_session TEXT,
|
||||
updated_at INTEGER NOT NULL,
|
||||
title TEXT,
|
||||
first_prompt TEXT,
|
||||
materialized INTEGER NOT NULL DEFAULT 0
|
||||
first_prompt TEXT
|
||||
) STRICT
|
||||
`)
|
||||
db.exec(`
|
||||
@@ -123,41 +122,69 @@ export function rowToEvent(row: EventRow): SessionEvent {
|
||||
}
|
||||
|
||||
/**
|
||||
* The committed prefix of an ordered event list: everything up to and including
|
||||
* the LAST `turn/end`, plus whether a crash tail (items after it) was cut.
|
||||
* The preserved prefix of an ordered event-row list (mirrors the JSONL
|
||||
* backend's `scanLog`): the longest prefix of complete, seq-contiguous,
|
||||
* parseable rows, PLUS the seq from which a never-committed torn tail must be
|
||||
* deleted (or `undefined` if the whole list is intact).
|
||||
*
|
||||
* Generic over anything carrying `seq` + `type` (an {@link EventRow} or a
|
||||
* {@link SessionEvent}) so the cut is computed from those COLUMNS alone — the
|
||||
* caller parses each row's `data` only for the committed items it returns,
|
||||
* never for the tail. This matters for the contract: a malformed `data` in an
|
||||
* uncommitted crash tail must be discarded, not make the session unloadable —
|
||||
* only a parse error / gap in the COMMITTED region is unloadable (see
|
||||
* `SessionPersistence.load`). Mirrors the JSONL backend's `scanLog`, which
|
||||
* likewise tolerates a corrupt tail after the last committed `turn/end`.
|
||||
* A crash can leave a durable log whose final turn never closed: real,
|
||||
* fully-written rows sit after the last `turn/end`. Those are PRESERVED — a
|
||||
* single turn can be huge in a long-horizon task, so truncating it would
|
||||
* destroy real work; the backend closes the orphaned open turn with a synthetic
|
||||
* `turn/end {kind:'interrupted'}` on load (ADR 0018). The ONLY thing excluded is
|
||||
* a torn trailing fragment — a row whose `data` never parses, or a seq gap —
|
||||
* AFTER the last committed `turn/end`; that bounds the preserved region and its
|
||||
* seq is returned as `tornFrom` so `load` can physically delete it.
|
||||
*
|
||||
* The loop only flushes at `turn/end`, so the last `turn/end` is the last
|
||||
* durable boundary; anything after it is a never-committed crash tail (a batch
|
||||
* that landed without its closing `turn/end`, e.g. a process killed mid-turn).
|
||||
* The committed region MUST be contiguous (`item.seq === i`); a gap there means
|
||||
* committed data was lost and the session is unloadable.
|
||||
* The last `turn/end` is computed from the `type` COLUMN (never parsing tail
|
||||
* `data`), so a malformed `data` in an uncommitted tail row is discarded rather
|
||||
* than making the session unloadable. A parse error or seq gap AT OR BEFORE the
|
||||
* last committed `turn/end` is committed-data corruption and throws.
|
||||
*
|
||||
* This relies on the session-log invariant that every event lives inside a turn
|
||||
* (`Session.append` enforces it): only the final turn can be open, so the
|
||||
* preserved tail is at most one unclosed turn.
|
||||
*/
|
||||
export function cutAtLastTurnEnd<T extends { seq: number; type: string }>(
|
||||
items: readonly T[],
|
||||
): { committed: T[]; cutTail: boolean } {
|
||||
let lastTurnEnd = -1
|
||||
items.forEach((item, i) => {
|
||||
if (item.type === 'turn/end') lastTurnEnd = i
|
||||
})
|
||||
// No committed turn/end anywhere: the whole list is an uncommitted first-turn
|
||||
// tail. Nothing is committed (mirrors scanLog returning zero events).
|
||||
if (lastTurnEnd < 0) {
|
||||
return { committed: [], cutTail: items.length > 0 }
|
||||
}
|
||||
const committed = items.slice(0, lastTurnEnd + 1)
|
||||
committed.forEach((item, i) => {
|
||||
if (item.seq !== i) {
|
||||
throw new Error(`corrupt session log: seq gap in committed region at index ${i} (expected ${i}, got ${item.seq})`)
|
||||
export function scanRows(rows: readonly EventRow[]): { preserved: SessionEvent[]; tornFrom?: number } {
|
||||
// Pass 1: parse each row's data; a row whose data is not valid JSON is a hole.
|
||||
// (The seq/type COLUMNS are always present even when `data` is corrupt.)
|
||||
interface Parsed { ok: boolean; event?: SessionEvent }
|
||||
const parsed: Parsed[] = rows.map((row) => {
|
||||
try {
|
||||
return { ok: true, event: rowToEvent(row) }
|
||||
} catch {
|
||||
return { ok: false }
|
||||
}
|
||||
})
|
||||
return { committed, cutTail: lastTurnEnd < items.length - 1 }
|
||||
|
||||
// The last index that is a valid `turn/end` — the last fully-committed
|
||||
// boundary (the loop flushes only at turn/end).
|
||||
let lastTurnEnd = -1
|
||||
for (let i = parsed.length - 1; i >= 0; i--) {
|
||||
if (parsed[i]?.ok && rows[i]?.type === 'turn/end') { lastTurnEnd = i; break }
|
||||
}
|
||||
|
||||
// Walk the longest PREFIX of complete, seq-contiguous, parseable rows
|
||||
// (row i has seq === i). This includes the fully-written rows of an
|
||||
// interrupted final turn AFTER the last turn/end — real work, never
|
||||
// truncated. The walk stops at the first hole:
|
||||
// - at or before the last committed turn/end → committed corruption (throw);
|
||||
// - after it (or no committed turn/end) → tolerated torn tail (stop).
|
||||
const preserved: SessionEvent[] = []
|
||||
for (let i = 0; i < rows.length; i++) {
|
||||
const p = parsed[i]
|
||||
if (!p?.ok || p.event === undefined) {
|
||||
if (i <= lastTurnEnd) throw new Error(`corrupt session log: unparsable committed event at seq ${rows[i]?.seq}`)
|
||||
break // torn tail fragment after the last turn/end — stop, tolerate
|
||||
}
|
||||
if (p.event.seq !== i) {
|
||||
if (i <= lastTurnEnd) throw new Error(`corrupt session log: seq gap in committed region (expected ${i}, got ${p.event.seq})`)
|
||||
break // gap after the last turn/end — torn tail, stop
|
||||
}
|
||||
preserved.push(p.event)
|
||||
}
|
||||
|
||||
// Any rows past the preserved prefix are a never-committed torn tail; their
|
||||
// first seq is the deletion point for load's physical repair.
|
||||
return preserved.length < rows.length ? { preserved, tornFrom: preserved.length } : { preserved }
|
||||
}
|
||||
|
||||
@@ -6,7 +6,7 @@ import { join } from 'node:path'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionEvent, SessionMeta } from '@deepseek-ai/dsh-session'
|
||||
import SessionPersistenceSqlite, { SCHEMA_VERSION } from '@deepseek-ai/dsh-session-persistence-sqlite'
|
||||
import { cutAtLastTurnEnd, openDatabase } from '../src/schema.ts'
|
||||
import { openDatabase, scanRows, type EventRow } from '../src/schema.ts'
|
||||
import { runPersistenceContract, meta, oneTurnLog } from '../../session-persistence/tests/contract.ts'
|
||||
|
||||
const dirs: string[] = []
|
||||
@@ -38,49 +38,78 @@ runPersistenceContract('sqlite', async () => {
|
||||
}
|
||||
})
|
||||
|
||||
describe('cutAtLastTurnEnd', () => {
|
||||
it('returns the prefix through the last complete turn/end and flags a cut tail', () => {
|
||||
const log = oneTurnLog()
|
||||
const withTail: SessionEvent[] = [
|
||||
...log,
|
||||
describe('scanRows', () => {
|
||||
// scanRows works off EventRows (data is a JSON string column); build them from
|
||||
// SessionEvents so the unit tests read in terms of the event vocabulary.
|
||||
const rows = (events: SessionEvent[]): EventRow[] =>
|
||||
events.map(e => ({ seq: e.seq, type: e.type, time: e.time, data: JSON.stringify(e.data) }))
|
||||
|
||||
it('preserves the full log when it ends exactly on a turn/end (no torn tail)', () => {
|
||||
const { preserved, tornFrom } = scanRows(rows(oneTurnLog()))
|
||||
expect(preserved).toEqual(oneTurnLog())
|
||||
expect(tornFrom).toBeUndefined()
|
||||
})
|
||||
|
||||
it('PRESERVES the real events of an interrupted turn after the last turn/end', () => {
|
||||
// turn 1 committed (0..5) + a crashed turn 2 (turn/start 6, step/start 7, no
|
||||
// close): all 8 rows are intact, so the whole prefix is preserved and there
|
||||
// is no torn fragment to delete. (load() then synthesizes the closers.)
|
||||
const withOpenTurn: SessionEvent[] = [
|
||||
...oneTurnLog(),
|
||||
{ type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'user/message', seq: 7, time: 8, data: { content: [{ type: 'text', text: 'q2' }], source: { kind: 'user' } } },
|
||||
{ type: 'step/start', seq: 7, time: 8, data: { turn: 2, step: 1 } },
|
||||
]
|
||||
const { committed, cutTail } = cutAtLastTurnEnd(withTail)
|
||||
expect(committed).toEqual(log)
|
||||
expect(cutTail).toBe(true)
|
||||
const { preserved, tornFrom } = scanRows(rows(withOpenTurn))
|
||||
expect(preserved.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
|
||||
expect(tornFrom).toBeUndefined()
|
||||
})
|
||||
|
||||
it('treats a log with no turn/end as fully uncommitted', () => {
|
||||
const partial: SessionEvent[] = [
|
||||
it('preserves the contiguous prefix and flags a torn tail at a seq gap', () => {
|
||||
// A gap after seq 0 (no committed turn/end): seq 0 is the preserved
|
||||
// interrupted-turn event; the gap bounds it and marks the torn fragment.
|
||||
const gapped: SessionEvent[] = [
|
||||
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'user/message', seq: 1, time: 2, data: { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } } },
|
||||
{ type: 'step/start', seq: 2, time: 2, data: { turn: 1, step: 1 } }, // seq 1 missing
|
||||
]
|
||||
expect(cutAtLastTurnEnd(partial)).toEqual({ committed: [], cutTail: true })
|
||||
const { preserved, tornFrom } = scanRows(rows(gapped))
|
||||
expect(preserved.map(e => e.seq)).toEqual([0])
|
||||
expect(tornFrom).toBe(1)
|
||||
})
|
||||
|
||||
it('reports no cut when the log ends exactly on a turn/end', () => {
|
||||
const { committed, cutTail } = cutAtLastTurnEnd(oneTurnLog())
|
||||
expect(committed).toEqual(oneTurnLog())
|
||||
expect(cutTail).toBe(false)
|
||||
it('an empty log preserves nothing and has no torn tail', () => {
|
||||
expect(scanRows([])).toEqual({ preserved: [] })
|
||||
})
|
||||
|
||||
it('an empty log is committed-empty with no tail', () => {
|
||||
expect(cutAtLastTurnEnd([])).toEqual({ committed: [], cutTail: false })
|
||||
})
|
||||
|
||||
it('throws on a seq gap inside the committed region', () => {
|
||||
it('throws on a seq gap inside the committed region (before the last turn/end)', () => {
|
||||
const gapped: SessionEvent[] = [
|
||||
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ 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' } } },
|
||||
]
|
||||
expect(() => cutAtLastTurnEnd(gapped)).toThrow(/seq gap in committed region/)
|
||||
expect(() => scanRows(rows(gapped))).toThrow(/seq gap in committed region/)
|
||||
})
|
||||
|
||||
it('throws on an unparsable row inside the committed region', () => {
|
||||
const withCorruptCommitted: EventRow[] = [
|
||||
{ seq: 0, type: 'turn/start', time: 1, data: '{not json' }, // corrupt, sits before a turn/end
|
||||
{ seq: 1, type: 'turn/end', time: 2, data: JSON.stringify({ turn: 1, reason: { kind: 'completed' } }) },
|
||||
]
|
||||
expect(() => scanRows(withCorruptCommitted)).toThrow(/unparsable committed event/)
|
||||
})
|
||||
|
||||
it('tolerates an unparsable torn-tail row after the last turn/end', () => {
|
||||
const withCorruptTail: EventRow[] = [
|
||||
...rows(oneTurnLog()),
|
||||
{ seq: 6, type: 'turn/start', time: 7, data: '{not json' }, // torn fragment, no committed turn/end after
|
||||
]
|
||||
const { preserved, tornFrom } = scanRows(withCorruptTail)
|
||||
expect(preserved).toEqual(oneTurnLog())
|
||||
expect(tornFrom).toBe(6)
|
||||
})
|
||||
})
|
||||
|
||||
describe('SessionPersistenceSqlite: durability and crash semantics', () => {
|
||||
it('a crash tail (rows after the last turn/end) is excluded on load and repaired on the next append', async () => {
|
||||
it('an interrupted turn (rows after the last turn/end) is PRESERVED and closed during load', async () => {
|
||||
const path = await freshDbPath()
|
||||
const m = meta('crash')
|
||||
// Run 1: persist a complete turn, then a half-written second turn (no turn/end).
|
||||
@@ -91,37 +120,44 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
|
||||
await ctx1.sessionPersistence.append(m.id, oneTurnLog())
|
||||
await ctx1.sessionPersistence.append(m.id, [
|
||||
{ type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'user/message', seq: 7, time: 8, data: { content: [{ type: 'text', text: 'q2' }], source: { kind: 'user' } } },
|
||||
{ type: 'step/start', seq: 7, time: 8, data: { turn: 2, step: 1 } },
|
||||
])
|
||||
await fiber1.dispose()
|
||||
|
||||
// Run 2: load returns only the committed first turn (tail excluded).
|
||||
// Run 2: load PRESERVES the interrupted turn's real events (a turn can be huge
|
||||
// — never truncated) and closes the orphaned turn with synthetic boundary
|
||||
// events: step/end (the step was open) then turn/end {interrupted}.
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
const fiber2 = await ctx2.plugin(SessionPersistenceSqlite, { path })
|
||||
const loaded = await ctx2.sessionPersistence.load(m.id)
|
||||
expect(loaded.events).toEqual(oneTurnLog())
|
||||
expect(loaded.events.map(e => e.type)).toEqual([
|
||||
'turn/start', 'user/message', 'step/start', 'assistant/message', 'step/end', 'turn/end', // turn 1
|
||||
'turn/start', 'step/start', 'step/end', 'turn/end', // turn 2: real events + synthetic closers
|
||||
])
|
||||
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
|
||||
const last = loaded.events.at(-1)!
|
||||
expect(last.type === 'turn/end' && last.data.reason).toEqual({ kind: 'interrupted' })
|
||||
|
||||
// The next append continues at seq 6 and performs the deferred truncation-
|
||||
// repair inside its transaction (DELETE seq >= 6 before inserting), so the
|
||||
// orphaned tail rows are gone and there is no UNIQUE collision.
|
||||
// load durably closed the turn, so the next append continues at the balanced
|
||||
// length (seq 10) and a reload round-trips identically.
|
||||
await ctx2.sessionPersistence.append(m.id, [
|
||||
{ type: 'turn/start', seq: 6, time: 9, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'turn/end', seq: 7, time: 10, data: { turn: 2, reason: { kind: 'completed' } } },
|
||||
{ type: 'turn/start', seq: 10, time: 9, data: { turn: 3, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'turn/end', seq: 11, time: 10, data: { turn: 3, 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])
|
||||
expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11])
|
||||
await fiber2.dispose()
|
||||
})
|
||||
|
||||
it('load() is non-mutating: the crash tail rows survive until the next append repairs them', async () => {
|
||||
it('load() durably closes the interrupted turn: the synthetic closers are on disk after load', async () => {
|
||||
const path = await freshDbPath()
|
||||
const m = meta('load-nonmutating')
|
||||
const m = meta('load-closes')
|
||||
const b1 = await backend(path)
|
||||
await b1.ctx.sessionPersistence.create(m)
|
||||
await b1.ctx.sessionPersistence.append(m.id, oneTurnLog()) // seqs 0..5
|
||||
await b1.dispose()
|
||||
// Hand-write an uncommitted tail (seq 6, no turn/end).
|
||||
// Hand-write an interrupted turn (turn/start seq 6, no turn/end).
|
||||
const db = openDatabase(path)
|
||||
db.prepare('INSERT INTO events (session_id, seq, type, time, data) VALUES (?, 6, ?, 7, ?)')
|
||||
.run(m.id, 'turn/start', JSON.stringify({ turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }))
|
||||
@@ -129,17 +165,20 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
|
||||
|
||||
const b2 = await backend(path)
|
||||
const loaded = await b2.ctx.sessionPersistence.load(m.id)
|
||||
expect(loaded.events).toEqual(oneTurnLog())
|
||||
// load() must NOT have deleted the tail row (contract: load returns the
|
||||
// prefix; the next append repairs). Verify the row is still on disk.
|
||||
// turn 2's real turn/start (seq 6) is preserved + a synthetic turn/end (seq 7).
|
||||
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
|
||||
expect(loaded.events.at(-1)!.type).toBe('turn/end')
|
||||
// load() is mutating: the synthetic turn/end MUST be on disk so the stored log
|
||||
// is balanced and the cursor is truthful (contract: load closes, not defers).
|
||||
const probe = openDatabase(path)
|
||||
const tailRows = probe.prepare('SELECT seq FROM events WHERE session_id = ? AND seq >= 6').all(m.id)
|
||||
const stored = probe.prepare('SELECT seq, type FROM events WHERE session_id = ? ORDER BY seq').all(m.id) as { seq: number; type: string }[]
|
||||
probe.close()
|
||||
expect(tailRows).toHaveLength(1)
|
||||
expect(stored.map(r => r.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
|
||||
expect(stored.at(-1)!.type).toBe('turn/end')
|
||||
await b2.dispose()
|
||||
})
|
||||
|
||||
it('all-tail load: a session whose only content is a crash tail is absent from has()/list()', async () => {
|
||||
it('all-tail load: a session whose only turn never closed is preserved and closed on load', async () => {
|
||||
const path = await freshDbPath()
|
||||
const m = meta('all-tail')
|
||||
const b1 = await backend(path)
|
||||
@@ -152,13 +191,15 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
|
||||
expect(await b1.ctx.sessionPersistence.has(m.id)).toBe(true) // materialized
|
||||
await b1.dispose()
|
||||
|
||||
// A fresh backend loads it: the committed prefix is empty (no turn/end), so
|
||||
// the session has no committed content. has()/list() must NOT report it.
|
||||
// A fresh backend loads it: the interrupted (only) turn's real events are
|
||||
// preserved and closed with a synthetic turn/end {interrupted} — NOT
|
||||
// truncated. The session was materialized, so has()/list() report it present.
|
||||
const b2 = await backend(path)
|
||||
const loaded = await b2.ctx.sessionPersistence.load(m.id)
|
||||
expect(loaded.events).toEqual([])
|
||||
expect(await b2.ctx.sessionPersistence.has(m.id)).toBe(false)
|
||||
expect((await b2.ctx.sessionPersistence.list()).map(x => x.id)).not.toContain(m.id)
|
||||
expect(loaded.events.map(e => e.type)).toEqual(['turn/start', 'user/message', 'turn/end'])
|
||||
expect(loaded.events.at(-1)!.type === 'turn/end' && loaded.events.at(-1)!.data).toMatchObject({ reason: { kind: 'interrupted' } })
|
||||
expect(await b2.ctx.sessionPersistence.has(m.id)).toBe(true)
|
||||
expect((await b2.ctx.sessionPersistence.list()).map(x => x.id)).toContain(m.id)
|
||||
await b2.dispose()
|
||||
})
|
||||
|
||||
@@ -205,10 +246,11 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
|
||||
await b1.ctx.sessionPersistence.append(m.id, oneTurnLog()) // committed: seqs 0..5
|
||||
await b1.dispose()
|
||||
|
||||
// Hand-insert an uncommitted tail row (seq 6, no closing turn/end) whose
|
||||
// `data` is invalid JSON. The contract: only a parse error in the COMMITTED
|
||||
// region is unloadable; a corrupt tail must be discarded (load cuts at the
|
||||
// last turn/end using seq+type columns, never parsing tail `data`).
|
||||
// Hand-insert a torn tail row (seq 6, no closing turn/end) whose `data` is
|
||||
// invalid JSON. The contract: only a parse error in the COMMITTED region is
|
||||
// unloadable; a torn tail must be discarded. scanRows finds the last
|
||||
// turn/end on the seq+type columns (never parsing tail `data`), so the
|
||||
// unparsable row after it bounds the preserved prefix and is deleted by load.
|
||||
const db = openDatabase(path)
|
||||
db.prepare('INSERT INTO events (session_id, seq, type, time, data) VALUES (?, 6, ?, 7, ?)')
|
||||
.run(m.id, 'turn/start', '{not valid json')
|
||||
@@ -216,8 +258,8 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
|
||||
|
||||
const b2 = await backend(path)
|
||||
const loaded = await b2.ctx.sessionPersistence.load(m.id)
|
||||
expect(loaded.events).toEqual(oneTurnLog()) // tail discarded, committed intact
|
||||
// The corrupt tail row was physically deleted, so a fresh append continues.
|
||||
expect(loaded.events).toEqual(oneTurnLog()) // torn tail discarded, committed intact (turn 1 already balanced → no closers)
|
||||
// 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, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'turn/end', seq: 7, time: 9, data: { turn: 2, reason: { kind: 'completed' } } },
|
||||
@@ -269,7 +311,7 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
|
||||
const path = await freshDbPath()
|
||||
// Materialize a row with version 2 directly via the real schema.
|
||||
const db = openDatabase(path)
|
||||
db.prepare('INSERT INTO sessions (id, version, created_at, updated_at, materialized) VALUES (?, ?, ?, ?, 1)')
|
||||
db.prepare('INSERT INTO sessions (id, version, created_at, updated_at) VALUES (?, ?, ?, ?)')
|
||||
.run('v2', 2, 1, 1)
|
||||
db.close()
|
||||
|
||||
|
||||
@@ -10,14 +10,14 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
|
||||
|---|---|
|
||||
| `create(meta): Promise<void>` | Register a new session's metadata. MAY defer the physical write until the first `append` (lazy materialization). |
|
||||
| `append(id, events): Promise<void>` | Durably persist a batch (from the `session/flush` drain). Append-only; first event `seq` == stored next-seq after any repair; rejects non-JSON-serializable data naming the offending type. |
|
||||
| `load(id): Promise<{ meta; events }>` | Reload meta + log up to the last complete `turn/end`; events contiguous (`events[i].seq === i`); rejects a mid-log gap/parse error or unknown `version`. |
|
||||
| `load(id): Promise<{ meta; events }>` | Reload meta + log. Preserves an interrupted (unclosed) final turn and closes it with synthetic `step/end?`+`turn/end {interrupted}` (a turn can be huge — never truncated); only a torn tail fragment is dropped. Events contiguous (`events[i].seq === i`); rejects a committed-region gap/parse error or unknown `version`. |
|
||||
| `list(): Promise<SessionMeta[]>` | Lightweight listing from metadata, no full-log parse. |
|
||||
| `has(id)` / `delete(id)` | Existence / removal. A zero-event lazily-materialized session is absent from `has`/`list`. |
|
||||
| `update(id, summary): Promise<void>` | Update mutable `SessionSummary` fields without touching the append-only log. |
|
||||
|
||||
## Invariants every backend must honor
|
||||
|
||||
- **Append-only.** Committed events (at or below a flushed `turn/end`) are never rewritten. The only exception is the one-time truncation-repair of a never-committed crash tail on the first `append` after a `load`.
|
||||
- **Append-only; a crashed turn is closed, not truncated.** Committed events (at or below a flushed `turn/end`) are never rewritten. A crash can leave an unclosed final turn whose events are real and possibly large; `load` preserves them and durably appends synthetic closers (`step/end?`+`turn/end {interrupted}`) to balance the log. Only a never-fully-written torn tail fragment is discarded.
|
||||
- **Contiguous seq.** `load` rejects a `seq` gap/parse error in the MIDDLE of the log; `append`'s first `seq` must equal the stored next-seq.
|
||||
- **JSON-serializable data.** `append` rejects non-serializable `event.data`; backends snapshot each event when buffering (the live `session.events` object is mutable).
|
||||
- **Durability.** `append` returns only once the batch is durable.
|
||||
|
||||
@@ -24,7 +24,7 @@
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-session": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -42,14 +42,16 @@ declare module 'cordis' {
|
||||
* Contracts every implementation MUST honor (a DB backend asserts them inside
|
||||
* a transaction; a file backend appends at EOF):
|
||||
*
|
||||
* - **Append-only.** Committed events — those at or below a flushed `turn/end`
|
||||
* — are never rewritten. The ONLY exception is the one-time truncation-repair
|
||||
* of a never-committed crash tail on the first {@link append} after a
|
||||
* {@link load} (see {@link load}).
|
||||
* - **Append-only; a crashed turn is closed, not truncated.** Committed events
|
||||
* — those at or below a flushed `turn/end` — are never rewritten. A crash can
|
||||
* leave an unclosed final turn whose events are real (and possibly large);
|
||||
* {@link load} preserves them and closes the orphaned turn with synthetic
|
||||
* boundary events (see {@link load}). Only a never-fully-written torn tail
|
||||
* fragment is discarded.
|
||||
* - **Contiguous seq.** A persisted log is contiguous: `events[i].seq === i`.
|
||||
* {@link load} rejects a parse error or a `seq` gap in the MIDDLE of the log
|
||||
* {@link load} rejects a parse error or a `seq` gap in the COMMITTED region
|
||||
* (unloadable); {@link append}'s first event `seq` MUST equal the backend's
|
||||
* stored next-seq after any repair.
|
||||
* stored next-seq (after `load` has balanced any interrupted turn).
|
||||
* - **JSON-serializable data.** `SessionEventMap` is merge-extensible and
|
||||
* `event.data` is typed only as `SessionEventMap[K]`, so {@link append}
|
||||
* REJECTS non-JSON-serializable data with an error naming the offending
|
||||
@@ -75,9 +77,9 @@ export abstract class SessionPersistence extends Service {
|
||||
/**
|
||||
* Durably persist a batch of events (called from the write-behind drain at
|
||||
* the `session/flush` checkpoint). Honors the append-only and contiguous-seq
|
||||
* contracts: the first event's `seq` MUST equal the stored next-seq after
|
||||
* any truncation-repair of a crash tail. Rejects non-JSON-serializable
|
||||
* `event.data` with an error naming the offending event type.
|
||||
* contracts: the first event's `seq` MUST equal the stored next-seq (after
|
||||
* `load` has durably closed any interrupted turn). Rejects non-JSON-
|
||||
* serializable `event.data` with an error naming the offending event type.
|
||||
*/
|
||||
abstract append(id: SessionId, events: readonly SessionEvent[]): Promise<void>
|
||||
|
||||
@@ -86,13 +88,19 @@ export abstract class SessionPersistence extends Service {
|
||||
* durable checkpoint. Returns `meta` AND `events` so the live session is
|
||||
* reconstructed with its `cwd`/lineage, not just its log.
|
||||
*
|
||||
* The loop only flushes at `turn/end`, so a crash can leave a half-written
|
||||
* final turn below the last committed checkpoint. `load` returns events only
|
||||
* up to the **last complete `turn/end`**; a subsequent {@link append} runs
|
||||
* the one-time truncation-repair that physically discards the orphaned tail
|
||||
* before writing. Returned events are contiguous (`events[i].seq === i`); a
|
||||
* parse error or a `seq` gap in the MIDDLE of the log makes the session
|
||||
* unloadable (reject). Rejects an unknown format `version`.
|
||||
* The loop only flushes at `turn/end`, so a crash can leave a durable log
|
||||
* whose final turn never closed: real, fully-written events sit after the last
|
||||
* `turn/end`. Those events are PRESERVED — a single turn can be huge in a
|
||||
* long-horizon task, so truncating it would destroy real work — and `load`
|
||||
* CLOSES the orphaned turn by durably appending the minimal synthetic boundary
|
||||
* events (a `step/end` if a step was open, then a `turn/end` carrying the
|
||||
* `{ kind: 'interrupted' }` reason). The returned `events` therefore end on a
|
||||
* balanced `turn/end` and are immediately usable as a session seed. Only a
|
||||
* never-fully-written TORN tail fragment (a half-written final record) is
|
||||
* discarded. Returned events are contiguous (`events[i].seq === i`); a parse
|
||||
* error or a `seq` gap in the COMMITTED region (at or before the last real
|
||||
* `turn/end`) makes the session unloadable (reject). Rejects an unknown format
|
||||
* `version`. See ADR 0018 for the crash-recovery contract.
|
||||
*/
|
||||
abstract load(id: SessionId): Promise<{ meta: SessionMeta; events: SessionEvent[] }>
|
||||
|
||||
|
||||
@@ -64,6 +64,44 @@ export function runPersistenceContract(name: string, make: () => Promise<Contrac
|
||||
}
|
||||
})
|
||||
|
||||
it('crash recovery: load preserves an interrupted (unclosed) turn and closes it with turn/end {interrupted}', async () => {
|
||||
const { persistence, dispose } = await make()
|
||||
try {
|
||||
const m = meta('interrupted')
|
||||
await persistence.create(m)
|
||||
await persistence.append(m.id, oneTurnLog()) // turn 1, committed (seqs 0..5)
|
||||
// A second turn that crashed mid-flight: turn/start + step/start were
|
||||
// durably written, but no step/end / turn/end ever arrived.
|
||||
await persistence.append(m.id, [
|
||||
{ type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'step/start', seq: 7, time: 8, data: { turn: 2, step: 1 } },
|
||||
])
|
||||
|
||||
// load PRESERVES the interrupted turn's events (a turn can be huge — they
|
||||
// must not be truncated) and closes the orphaned turn with synthetic
|
||||
// boundary events: step/end (the step was open) then turn/end {interrupted}.
|
||||
const loaded = await persistence.load(m.id)
|
||||
expect(loaded.events.map(e => e.type)).toEqual([
|
||||
'turn/start', 'user/message', 'step/start', 'assistant/message', 'step/end', 'turn/end', // turn 1
|
||||
'turn/start', 'step/start', 'step/end', 'turn/end', // turn 2: real events + synthetic closers
|
||||
])
|
||||
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
|
||||
const last = loaded.events.at(-1)!
|
||||
expect(last.type === 'turn/end' && last.data.reason).toEqual({ kind: 'interrupted' })
|
||||
|
||||
// The closed log is durable and continuable: a fresh append continues at
|
||||
// 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, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'turn/end', seq: 11, time: 10, data: { turn: 3, 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])
|
||||
} finally {
|
||||
await dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('has()/list() exclude a created-but-never-appended (zero-event) session', async () => {
|
||||
const { persistence, dispose } = await make()
|
||||
try {
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { SessionId, isJsonValue } from '@deepseek-ai/dsh-session'
|
||||
import { SessionId, isJsonValue, interruptedTurnClosers } from '@deepseek-ai/dsh-session'
|
||||
import type { SessionEvent, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session'
|
||||
import { SessionPersistence } from '../src/index.ts'
|
||||
import { runPersistenceContract, meta, oneTurnLog } from './contract.ts'
|
||||
@@ -46,6 +46,11 @@ class MemoryPersistence extends SessionPersistence {
|
||||
async load(id: SessionId): Promise<{ meta: SessionMeta; events: SessionEvent[] }> {
|
||||
const entry = this.store.get(id)
|
||||
if (!entry) throw new Error(`session "${id}" not found`)
|
||||
// Honor the crash-recovery contract: if the stored log ends mid-turn, close
|
||||
// the orphaned turn durably with synthetic boundary events and continue from
|
||||
// the balanced length.
|
||||
const closers = interruptedTurnClosers(entry.events)
|
||||
if (closers.length > 0) entry.events.push(...structuredClone(closers))
|
||||
return { meta: structuredClone(entry.meta), events: structuredClone(entry.events) }
|
||||
}
|
||||
|
||||
|
||||
@@ -24,7 +24,7 @@
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-llm": "^0.0.1",
|
||||
"@deepseek-ai/dsh-llm": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -15,6 +15,7 @@ import { isJsonValue } from './json.ts'
|
||||
|
||||
export * from './types.ts'
|
||||
export { isJsonValue } from './json.ts'
|
||||
export { interruptedTurnClosers } from './repair.ts'
|
||||
|
||||
declare module 'cordis' {
|
||||
interface Context {
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
/**
|
||||
* JSON-serializability validation for session event data.
|
||||
*
|
||||
* The session event log is the durable source of truth (ADR 0003/0016): every
|
||||
* The session event log is the durable source of truth (ADR 0003/0018): every
|
||||
* `event.data` must round-trip losslessly through JSON so any persistence
|
||||
* backend can store and reload it byte-identically. This invariant belongs to
|
||||
* the log itself — `Session.append` enforces it at the source, so a
|
||||
|
||||
79
packages/session/src/repair.ts
Normal file
79
packages/session/src/repair.ts
Normal file
@@ -0,0 +1,79 @@
|
||||
/**
|
||||
* Crash-recovery repair for an interrupted session log.
|
||||
*
|
||||
* A persistence backend flushes only at `turn/end`, so a crash can leave a
|
||||
* durable log whose final turn never closed: real, fully-written events sit
|
||||
* after the last `turn/end` with no closing boundary. A single turn can be huge
|
||||
* in a long-horizon task (many steps, large tool output), so those events MUST
|
||||
* be preserved — truncating the turn would silently destroy real work. Instead,
|
||||
* on reload the backend CLOSES the orphaned turn by appending the minimal
|
||||
* synthetic boundary events (a `step/end` if a step was still open, then a
|
||||
* `turn/end` carrying the merge-extensible `{ kind: 'interrupted' }` reason).
|
||||
* The marker records that the turn was cut short by a crash, not completed by
|
||||
* the model. See ADR 0018.
|
||||
*
|
||||
* This module computes those synthetic closers from an event list; the backend
|
||||
* returns them inline from `load` (so the reconstructed session is balanced and
|
||||
* immediately usable) and persists them on the first post-load `append`.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-session/repair
|
||||
*/
|
||||
|
||||
import type { SessionEvent } from './types.ts'
|
||||
|
||||
/**
|
||||
* Scan `events` for an open turn/step at the tail and return the synthetic
|
||||
* boundary events that close them, with `seq` continuing the log and `time`
|
||||
* copied from the last real event (the closers stand in for the crash moment;
|
||||
* reusing the last timestamp keeps them deterministic and never invents a
|
||||
* "future" time). Returns an empty array when the log is already balanced
|
||||
* (ends on a `turn/end`, or is empty) — the common, non-crash case.
|
||||
*
|
||||
* Only the LAST turn can be open: the invariants plugin guarantees a `turn/end`
|
||||
* before any later `turn/start`, so an interior open turn is impossible in a
|
||||
* valid committed log. Likewise at most one step is open within that turn.
|
||||
*/
|
||||
export function interruptedTurnClosers(events: readonly SessionEvent[]): SessionEvent[] {
|
||||
let openTurn: number | null = null
|
||||
let openStep: number | null = null
|
||||
for (const event of events) {
|
||||
switch (event.type) {
|
||||
case 'turn/start':
|
||||
openTurn = event.data.turn
|
||||
break
|
||||
case 'turn/end':
|
||||
openTurn = null
|
||||
openStep = null
|
||||
break
|
||||
case 'step/start':
|
||||
openStep = event.data.step
|
||||
break
|
||||
case 'step/end':
|
||||
openStep = null
|
||||
break
|
||||
// Other event types do not move the turn/step boundary cursor.
|
||||
default:
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
// Balanced log (no crash mid-turn): nothing to close. An open turn implies
|
||||
// `events` is non-empty (its turn/start was logged), so `last` exists.
|
||||
const last = events.at(-1)
|
||||
if (openTurn === null || last === undefined) return []
|
||||
|
||||
// The last real event supplies the seq base and the timestamp for the
|
||||
// synthetic closers (reusing the last timestamp keeps them deterministic and
|
||||
// never invents a "future" time).
|
||||
let seq = last.seq + 1
|
||||
const time = last.time
|
||||
const closers: SessionEvent[] = []
|
||||
|
||||
// Close an open step first — a turn/end while a step is open is an invariant
|
||||
// violation, so the step's boundary must be synthesized before the turn's.
|
||||
if (openStep !== null) {
|
||||
closers.push({ type: 'step/end', seq: seq++, time, data: { turn: openTurn, step: openStep } })
|
||||
}
|
||||
closers.push({ type: 'turn/end', seq: seq++, time, data: { turn: openTurn, reason: { kind: 'interrupted' } } })
|
||||
return closers
|
||||
}
|
||||
@@ -113,6 +113,17 @@ export interface TurnEndReasonMap {
|
||||
error: { kind: 'error'; message: string; code?: string }
|
||||
disposed: { kind: 'disposed' }
|
||||
'max-tokens': { kind: 'max-tokens' }
|
||||
/**
|
||||
* The turn never ended on its own: the process crashed mid-turn and a
|
||||
* persistence backend later closed the orphaned (open) turn on reload so the
|
||||
* log stays balanced. SYNTHESIZED by the backend's crash-recovery repair — no
|
||||
* loop ever emits this. Its events are real (they were durably appended before
|
||||
* the crash) and are PRESERVED, not discarded: a single turn can be huge in a
|
||||
* long-horizon task (many steps, large tool output), so truncating it would
|
||||
* lose real work. The marker records that the turn was cut short, not that the
|
||||
* model completed it. See ADR 0018.
|
||||
*/
|
||||
interrupted: { kind: 'interrupted' }
|
||||
}
|
||||
|
||||
export type TurnEndReason = TurnEndReasonMap[keyof TurnEndReasonMap]
|
||||
|
||||
@@ -24,7 +24,7 @@
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-llm": "^0.0.1",
|
||||
"@deepseek-ai/dsh-llm": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -27,14 +27,14 @@
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-agent": "^0.0.1",
|
||||
"@deepseek-ai/dsh-agent-loop": "^0.0.1",
|
||||
"@deepseek-ai/dsh-bash": "^0.0.1",
|
||||
"@deepseek-ai/dsh-bash-local": "^0.0.1",
|
||||
"@deepseek-ai/dsh-llm": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session": "^0.0.1",
|
||||
"@deepseek-ai/dsh-system-prompt": "^0.0.1",
|
||||
"@deepseek-ai/dsh-tools": "^0.0.1",
|
||||
"@deepseek-ai/dsh-agent": "workspace:^",
|
||||
"@deepseek-ai/dsh-agent-loop": "workspace:^",
|
||||
"@deepseek-ai/dsh-bash": "workspace:^",
|
||||
"@deepseek-ai/dsh-bash-local": "workspace:^",
|
||||
"@deepseek-ai/dsh-llm": "workspace:^",
|
||||
"@deepseek-ai/dsh-session": "workspace:^",
|
||||
"@deepseek-ai/dsh-system-prompt": "workspace:^",
|
||||
"@deepseek-ai/dsh-tools": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -26,9 +26,9 @@
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-agent": "^0.0.1",
|
||||
"@deepseek-ai/dsh-llm": "^0.0.1",
|
||||
"@deepseek-ai/dsh-system-prompt": "^0.0.1",
|
||||
"@deepseek-ai/dsh-agent": "workspace:^",
|
||||
"@deepseek-ai/dsh-llm": "workspace:^",
|
||||
"@deepseek-ai/dsh-system-prompt": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user