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

Inline session fork parameters

Browse Source
This commit is contained in:
Hypatia May
2026-07-07 09:04:49 +08:00
parent 9c5da8d81f
commit c7fba41ba9
8 changed files with 38 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/core/session/README.md
View File

@@ -9,7 +9,7 @@ Creates and holds event-sourced `Session` instances. Persistence is intentionall
### Public API
- `ctx.sessions.create(id?: SessionId, options?: { seed?: SessionEvent[]; meta?: { cwd?: string; parentSession?: SessionId; createdAt?: number; seedLength?: number } }): Session` — Create a session. `options.seed` replays/forks an existing event log; `options.meta` attaches creation metadata (validated absolute `cwd`, `parentSession` lineage, seed boundary) as the immutable `SessionHeader`. The store fills `version`/`id` and defaults `createdAt` to now; a caller reconstructing a persisted session passes the original `createdAt` and persisted `seedLength` to preserve them. Disposed with the calling fiber.
- `ctx.sessions.fork({ source, boundary?, childSessionId? }): Session` — Resolve a live session object or id, select a seed through the inclusive `boundary` event seq (default: current last event), require that boundary to be `turn/end`, and create a live child session with lineage metadata.
- `ctx.sessions.fork(source, boundary?, childSessionId?): Session` — Resolve a live session object or id, select a seed through the inclusive `boundary` event seq (default: current last event), require that boundary to be `turn/end`, and create a live child session with lineage metadata.
- `ctx.sessions.get(id: SessionId): Session | undefined`
- `ctx.sessions.list(): Session[]`
@@ -73,7 +73,7 @@ Every `SessionEvent` carries two optional top-level fields (structural metadata)
### Extension points
- Persistence plugins: subscribe to `session/event` (write-behind) and drain on `session/flush` (awaited) and fiber dispose. A durable backend reads the log and reloads it into a live session; the metadata seam (`SessionHeader`, `session.header`) is what such a backend stores beside the log.
- Replay/fork: `ctx.sessions.create(id, { seed })` seeds a new session with an existing event log. The surface rebuilds deterministically from `surfaceOp` markers in the seeded events. The seed is validated to the SAME always-on invariants `append` enforces — contiguous seqs, JSON-serializable data, and required `surfaceOp` markers on surface-eligible events — so marker-less message events are rejected at construction rather than silently vanishing from `deriveMessages()`. Broader turn-enclosure checks stay in `dsh-invariants` and persistence repair. Ordinary live-session forks use `ctx.sessions.fork({ source, boundary?, childSessionId? })`, where `boundary` is the inclusive source event seq to fork through.
- Replay/fork: `ctx.sessions.create(id, { seed })` seeds a new session with an existing event log. The surface rebuilds deterministically from `surfaceOp` markers in the seeded events. The seed is validated to the SAME always-on invariants `append` enforces — contiguous seqs, JSON-serializable data, and required `surfaceOp` markers on surface-eligible events — so marker-less message events are rejected at construction rather than silently vanishing from `deriveMessages()`. Broader turn-enclosure checks stay in `dsh-invariants` and persistence repair. Ordinary live-session forks use `ctx.sessions.fork(source, boundary?, childSessionId?)`, where `boundary` is the inclusive source event seq to fork through.
- Compaction: the `dsh-compact-basic` plugin appends a `user/message` with `surfaceOp: { op: 'replace', start, end }` to shadow old surface nodes behind a summary checkpoint.
### What is NOT here (TODO)

36
packages/core/session/src/index.ts
View File

@@ -365,19 +365,6 @@ export class Session {
/** A fork source: either the live session object or its live store id. */
export type SessionForkSource = Session | SessionId
/** Inputs for live session forking. */
export interface ForkSessionOptions {
/** Live source session object or id. */
source: SessionForkSource
/**
* Inclusive source event seq to fork through. Omitted means the source's
* current last event; omitted on an empty source forks an empty child.
*/
boundary?: number
/** Optional child session id; omitted delegates to SessionStore's id policy. */
childSessionId?: SessionId
}
export type SessionForkErrorCode =
| 'SESSION_NOT_FOUND'
| 'SESSION_NOT_LIVE'
@@ -533,20 +520,25 @@ export class SessionStore extends Service {
* `boundary` is an inclusive source event seq; omitted means the source's
* current last event. A non-empty selected slice must end at `turn/end`.
*
* @param options Source, optional boundary, and optional child id for the fork.
* @param source - Live source session object or id.
* @param boundary - Inclusive source event seq to fork through; omitted means
* the source's current last event, and omitted on an empty source forks an
* empty child.
* @param childSessionId - Optional child session id; omitted delegates to
* `SessionStore`'s id policy.
* @returns The created live child session.
*/
fork(options: ForkSessionOptions): Session {
if (options.childSessionId !== undefined && this.get(options.childSessionId) !== undefined) {
throw new SessionForkError(`session "${options.childSessionId}" already exists`, 'SESSION_ALREADY_EXISTS')
fork(source: SessionForkSource, boundary?: number, childSessionId?: SessionId): Session {
if (childSessionId !== undefined && this.get(childSessionId) !== undefined) {
throw new SessionForkError(`session "${childSessionId}" already exists`, 'SESSION_ALREADY_EXISTS')
}
const source = this._resolveForkSource(options.source)
const seed = this._forkSeed(source, options.boundary)
return this.create(options.childSessionId, {
const liveSource = this._resolveForkSource(source)
const seed = this._forkSeed(liveSource, boundary)
return this.create(childSessionId, {
seed,
meta: {
...source.header.cwd !== undefined ? { cwd: source.header.cwd } : {},
parentSession: source.id,
...liveSource.header.cwd !== undefined ? { cwd: liveSource.header.cwd } : {},
parentSession: liveSource.id,
seedLength: seed.length,
},
})

40
packages/core/session/tests/fork.spec.ts
View File

@@ -49,7 +49,7 @@ describe('SessionStore.fork', () => {
const { ctx, sessions } = await setup()
const source = ctx.sessions.create(SessionId('empty-parent'), { meta: { cwd: '/workspace' } })
const child = sessions.fork({ source, childSessionId: SessionId('empty-child') })
const child = sessions.fork(source, undefined, SessionId('empty-child'))
expect(child.events).toEqual([])
expect(child.header).toMatchObject({
@@ -65,7 +65,7 @@ describe('SessionStore.fork', () => {
const source = ctx.sessions.create(SessionId('parent'), { meta: { cwd: '/workspace' } })
appendClosedTurn(source, 1, 'hello')
const child = sessions.fork({ source: SessionId('parent'), childSessionId: SessionId('child') })
const child = sessions.fork(SessionId('parent'), undefined, SessionId('child'))
expect(child.events).toEqual(source.events)
expect(child.events).not.toBe(source.events)
@@ -88,11 +88,7 @@ describe('SessionStore.fork', () => {
appendClosedTurn(source, 2, 'second')
appendOpenTurn(source, 3)
const child = sessions.fork({
source,
boundary: firstBoundary,
childSessionId: SessionId('child-from-first'),
})
const child = sessions.fork(source, firstBoundary, SessionId('child-from-first'))
expect(child.events).toEqual(source.events.slice(0, firstBoundary + 1))
expect(child.header.seedLength).toBe(firstBoundary + 1)
@@ -114,11 +110,7 @@ describe('SessionStore.fork', () => {
const source = ctx.sessions.create(SessionId(`parent-${reason.kind}`))
appendClosedTurn(source, 1, reason.kind, reason)
const child = sessions.fork({
source,
boundary: lastSeq(source),
childSessionId: SessionId(`child-${reason.kind}`),
})
const child = sessions.fork(source, lastSeq(source), SessionId(`child-${reason.kind}`))
expect(child.events.at(-1)?.type).toBe('turn/end')
expect(child.header.seedLength).toBe(source.events.length)
@@ -128,19 +120,19 @@ describe('SessionStore.fork', () => {
it('rejects invalid boundaries before creating a child', async () => {
const { ctx, sessions } = await setup()
const empty = ctx.sessions.create(SessionId('empty'))
expect(() => sessions.fork({ source: empty, boundary: 0, childSessionId: SessionId('empty-child') }))
expect(() => sessions.fork(empty, 0, SessionId('empty-child')))
.toThrow(new SessionForkError('fork boundary 0 does not exist in session "empty" (last seq: none)', 'INVALID_BOUNDARY'))
expect(ctx.sessions.get(SessionId('empty-child'))).toBeUndefined()
const source = ctx.sessions.create(SessionId('parent'))
appendClosedTurn(source, 1)
expect(() => sessions.fork({ source, boundary: -1, childSessionId: SessionId('negative') }))
expect(() => sessions.fork(source, -1, SessionId('negative')))
.toThrow(/non-negative safe integer/)
expect(() => sessions.fork({ source, boundary: 0.5, childSessionId: SessionId('fraction') }))
expect(() => sessions.fork(source, 0.5, SessionId('fraction')))
.toThrow(/non-negative safe integer/)
expect(() => sessions.fork({ source, boundary: Number.MAX_SAFE_INTEGER + 1, childSessionId: SessionId('unsafe') }))
expect(() => sessions.fork(source, Number.MAX_SAFE_INTEGER + 1, SessionId('unsafe')))
.toThrow(/non-negative safe integer/)
expect(() => sessions.fork({ source, boundary: source.seq, childSessionId: SessionId('past-end') }))
expect(() => sessions.fork(source, source.seq, SessionId('past-end')))
.toThrow(new SessionForkError(`fork boundary ${source.seq} does not exist in session "parent" (last seq: ${source.seq - 1})`, 'INVALID_BOUNDARY'))
})
@@ -151,7 +143,7 @@ describe('SessionStore.fork', () => {
const mutableLog = (source as unknown as { log: SessionEvent[] }).log
mutableLog[2] = { ...mutableLog[2]!, seq: 99 }
expect(() => sessions.fork({ source, boundary: 2, childSessionId: SessionId('corrupt-child') }))
expect(() => sessions.fork(source, 2, SessionId('corrupt-child')))
.toThrow(new SessionForkError('fork boundary 2 does not match a contiguous event seq in session "corrupt-parent"', 'INVALID_BOUNDARY'))
expect(ctx.sessions.get(SessionId('corrupt-child'))).toBeUndefined()
})
@@ -159,7 +151,7 @@ describe('SessionStore.fork', () => {
it('rejects an unknown live session id', async () => {
const { sessions } = await setup()
expect(() => sessions.fork({ source: SessionId('missing') }))
expect(() => sessions.fork(SessionId('missing')))
.toThrow(new SessionForkError('session "missing" not found', 'SESSION_NOT_FOUND'))
})
@@ -167,7 +159,7 @@ describe('SessionStore.fork', () => {
const { sessions } = await setup()
const detached = new Session(SessionId('detached'))
expect(() => sessions.fork({ source: detached }))
expect(() => sessions.fork(detached))
.toThrow(new SessionForkError('session "detached" not found', 'SESSION_NOT_FOUND'))
})
@@ -176,7 +168,7 @@ describe('SessionStore.fork', () => {
ctx.sessions.create(SessionId('same-id'))
const stale = new Session(SessionId('same-id'))
expect(() => sessions.fork({ source: stale }))
expect(() => sessions.fork(stale))
.toThrow(new SessionForkError('session "same-id" is not the live store instance', 'SESSION_NOT_LIVE'))
})
@@ -221,7 +213,7 @@ describe('SessionStore.fork', () => {
const source = ctx.sessions.create(SessionId(`open-${lastType}`))
const boundary = build(source)
expect(() => sessions.fork({ source, boundary }))
expect(() => sessions.fork(source, boundary))
.toThrow(new SessionForkError(`fork boundary ${boundary} in session "open-${lastType}" must be turn/end, got ${lastType}`, 'OPEN_TURN'))
}
})
@@ -232,7 +224,7 @@ describe('SessionStore.fork', () => {
appendClosedTurn(source, 1)
ctx.sessions.create(SessionId('child'))
expect(() => sessions.fork({ source, childSessionId: SessionId('child') }))
expect(() => sessions.fork(source, undefined, SessionId('child')))
.toThrow(new SessionForkError('session "child" already exists', 'SESSION_ALREADY_EXISTS'))
})
@@ -242,7 +234,7 @@ describe('SessionStore.fork', () => {
source.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
ctx.sessions.create(SessionId('child'))
expect(() => sessions.fork({ source, childSessionId: SessionId('child') }))
expect(() => sessions.fork(source, undefined, SessionId('child')))
.toThrow(new SessionForkError('session "child" already exists', 'SESSION_ALREADY_EXISTS'))
})
})

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

@@ -144,7 +144,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
const source = ctx.sessions.create(SessionId('persist-parent'), { meta: { cwd: '/workspace' } })
appendClosedTurn(source)
const child = ctx.sessions.fork({ source, childSessionId: SessionId('persist-child') })
const child = ctx.sessions.fork(source, undefined, SessionId('persist-child'))
await ctx.parallel('session/flush', child)
const loaded = await ctx.sessionPersistence.load(child.id)