fix: collapse session fork to one api

This commit is contained in:
Hypatia May
2026-07-06 13:57:59 +08:00
parent 37f3aedc0b
commit 37aac7f313
10 changed files with 284 additions and 177 deletions

View File

@@ -108,7 +108,7 @@ Every session event is turn-enclosed. Reloading a crashed session preserves the
The session log is the source of truth. `deriveMessages()` projects session events into the `Message[]` sent to the model; raw `assistant/chunk` events stay in the log for replay and UI fidelity. Replay, fork, resume, transcript rendering, telemetry, and persistence all derive from the same event stream. The session log is the source of truth. `deriveMessages()` projects session events into the `Message[]` sent to the model; raw `assistant/chunk` events stay in the log for replay and UI fidelity. Replay, fork, resume, transcript rendering, telemetry, and persistence all derive from the same event stream.
For live forks, `ctx.sessions.snapshot(source)` validates an empty or turn-ended source and returns seed metadata; `ctx.sessions.fork()` creates the child session from it. For live forks, `ctx.sessions.fork({ source, boundary?, childSessionId? })` creates a child from a turn-enclosed source prefix.
Durability is a plugin concern. Persistence backends buffer synchronous `session/event` notifications and the loop awaits a turn-end checkpoint before moving on. The `SessionPersistence` seam stores `SessionEvent` directly, with metadata in `SessionHeader`; JSONL and SQLite share one contract suite. Durability is a plugin concern. Persistence backends buffer synchronous `session/event` notifications and the loop awaits a turn-end checkpoint before moving on. The `SessionPersistence` seam stores `SessionEvent` directly, with metadata in `SessionHeader`; JSONL and SQLite share one contract suite.
@@ -143,6 +143,6 @@ New behavior should attach to a documented seam; changing the shipped loop requi
| Intercept prompts, requests, tool use, or continuation | listen on the relevant `agent/*` or `tools/*` waterfall | | Intercept prompts, requests, tool use, or continuation | listen on the relevant `agent/*` or `tools/*` waterfall |
| Add UI or editor integration | drive `ctx.agents` and render from `session/event` | | Add UI or editor integration | drive `ctx.agents` and render from `session/event` |
| Add durable session state | add a `SessionEventMap` member and render/replay from the log | | Add durable session state | add a `SessionEventMap` member and render/replay from the log |
| Fork a live session | use `ctx.sessions.snapshot()` or `ctx.sessions.fork()` | | Fork a live session | use `ctx.sessions.fork({ source, boundary?, childSessionId? })` |
The [extension cookbook](cookbook/extension-cookbook.md) carries plugin skeletons and the feature-to-seam map; step-by-step guides cover [packages](cookbook/adding-a-package.md), [tools](cookbook/adding-a-tool.md), [LLM adapters](cookbook/adding-an-llm-adapter.md), and [vendored packages](cookbook/adding-a-vendored-package.md). The [extension cookbook](cookbook/extension-cookbook.md) carries plugin skeletons and the feature-to-seam map; step-by-step guides cover [packages](cookbook/adding-a-package.md), [tools](cookbook/adding-a-tool.md), [LLM adapters](cookbook/adding-an-llm-adapter.md), and [vendored packages](cookbook/adding-a-vendored-package.md).

View File

@@ -161,11 +161,10 @@ enter(session: Session): () => void
announce(session: Session): void announce(session: Session): void
get(id: SessionId): Session | undefined get(id: SessionId): Session | undefined
list(): Session[] list(): Session[]
snapshot(source: SessionForkSource): SessionForkSeed
fork(options: ForkSessionOptions): Session fork(options: ForkSessionOptions): Session
``` ```
Source: [`packages/core/session/src/index.ts:369`](../../packages/core/session/src/index.ts) Source: [`packages/core/session/src/index.ts:358`](../../packages/core/session/src/index.ts)
## `ctx.subagents` — `SubagentService` ## `ctx.subagents` — `SubagentService`

View File

@@ -159,14 +159,13 @@ export interface SurfaceNode {
Everything else (`turn/*`, `step/*`) is structural and does not project into a message. Token usage is observed on `assistant/message.usage` (the step that produced it); an operational error's step number is on `turn/end.reason` for `kind: 'error'`. Everything else (`turn/*`, `step/*`) is structural and does not project into a message. Token usage is observed on `assistant/message.usage` (the step that produced it); an operational error's step number is on `turn/end.reason` for `kind: 'error'`.
## Live-session fork helpers ## Live-session fork API
`ctx.sessions.create(id, { seed, meta })` is the low-level replay/fork primitive. For ordinary live-session forks, `SessionStore` adds two policy helpers: `ctx.sessions.create(id, { seed, meta })` is the low-level replay/fork primitive. For ordinary live-session forks, `SessionStore` exposes one policy API:
- `snapshot(source)` accepts a live `Session` object or live `SessionId`, validates the source log is empty or ends at `turn/end`, then returns a deep-cloned `SessionEvent[]` seed plus child metadata (`parentSession`, `seedLength`, and inherited `cwd`). - `fork({ source, boundary?, childSessionId? })` accepts a live `Session` object or live `SessionId`, selects source events through the inclusive `boundary` seq (default: current last event), validates that selected prefix is turn-enclosed and empty or ends at `turn/end`, then creates a live child session with deep-cloned seed events plus child metadata (`parentSession`, `seedLength`, and inherited `cwd`).
- `fork({ source, sessionId? })` calls `snapshot(source)` and immediately creates the live child via `ctx.sessions.create(sessionId, { seed, meta })`.
The split is intentional: `snapshot()` is the reusable seed/metadata computation for callers that create an agent or defer session creation; `fork()` is the convenience path when a caller only needs a child `Session`. Both reject open-turn sources instead of clipping to an older prefix. `dsh-subagent-fork` keeps its completed-prefix clipping because tool-time delegation usually starts while the parent turn is open; ordinary session branching should not silently drop the parent turn tail. An explicit `boundary` lets callers fork from a previous completed turn even if the source has newer events or an open current turn. The API rejects open or malformed selected prefixes instead of clipping silently. `dsh-subagent-fork` keeps its completed-prefix clipping because tool-time delegation usually starts while the parent turn is open; ordinary session branching should make the requested boundary explicit.
## What started a turn: `TurnTriggerMap` ## What started a turn: `TurnTriggerMap`

View File

@@ -58,7 +58,7 @@ Generated by `pnpm run gen-rfc-index` from the RFC tree — never edit by hand;
| [dsh-hooks-claude + dsh-hooks-codex — the Claude Code / Codex hook bridges](implemented/feature/2026-06-30-hook-bridges.md) | 2026-06-30 | | [dsh-hooks-claude + dsh-hooks-codex — the Claude Code / Codex hook bridges](implemented/feature/2026-06-30-hook-bridges.md) | 2026-06-30 |
| [dsh-hook-protocol — the shared Claude Code / Codex hook wire-protocol core](implemented/feature/2026-06-30-hook-protocol-lib.md) | 2026-06-30 | | [dsh-hook-protocol — the shared Claude Code / Codex hook wire-protocol core](implemented/feature/2026-06-30-hook-protocol-lib.md) | 2026-06-30 |
| [Interception seams — the typed-Decision surface a hook programs against](implemented/feature/2026-06-30-interception-seams.md) | 2026-06-30 | | [Interception seams — the typed-Decision surface a hook programs against](implemented/feature/2026-06-30-interception-seams.md) | 2026-06-30 |
| [SessionStore fork helpers](implemented/feature/2026-06-30-session-store-fork-helpers.md) | 2026-06-30 | | [SessionStore fork API](implemented/feature/2026-06-30-session-store-fork-api.md) | 2026-06-30 |
| [Subagent lifecycle enrichment — lastAssistantMessage (observe-only)](implemented/feature/2026-06-30-subagent-observe-enrich.md) | 2026-06-30 | | [Subagent lifecycle enrichment — lastAssistantMessage (observe-only)](implemented/feature/2026-06-30-subagent-observe-enrich.md) | 2026-06-30 |
### Simplification ### Simplification

View File

@@ -0,0 +1,47 @@
# RFC: SessionStore fork API
Status: implemented
## Problem
The event-sourced session log already has the primitive a fork needs: create a new session with a seed event prefix, then derive model history from that seeded log exactly as replay does. That primitive is intentionally low-level: `ctx.sessions.create(id, { seed, meta })` accepts any valid seed, but ordinary live-session branching needs policy around which prefix can be copied, which metadata is stamped on the child, and how errors are classified.
The semantic hazard is the fork boundary. A valid user-visible fork seed must be contiguous and turn-enclosed. Forking inside an active turn would copy an open `turn/start`, possibly an open `step/start`, and possibly dangling tool calls. That violates turn-enclosure and provider-transcript invariants, and it creates a misleading child history that appears to have participated in an unfinished parent turn. The existing [subagent seam](../../implemented/feature/2026-06-21-subagent-capability-seam.md) deliberately solves a different problem: tool-triggered subagent forks usually happen while the parent turn is open, so `dsh-subagent-fork` clips the seed to the parent's last completed-turn prefix. A general session fork should not silently clip; it should either fork the requested boundary or reject it.
## Decision
`dsh-session` owns ordinary live-session forking directly on `ctx.sessions`. There is no separate `dsh-session-fork` package or `ctx.sessionFork` service: the API has no independent backend, event vocabulary, lifecycle, or persistence behavior, and all durable work delegates to the existing session store and persistence backends.
The store exposes one operation:
```ts ignore-check
type SessionForkSource = Session | SessionId
interface ForkSessionOptions {
source: SessionForkSource
boundary?: number
childSessionId?: SessionId
}
class SessionStore extends Service {
fork(options: ForkSessionOptions): Session
}
```
`boundary` is the inclusive source event `seq` to copy through. When omitted, it defaults to the source session's current last event; on an empty source, omitted `boundary` creates an empty child. The selected prefix is deep-cloned into the child seed. The child inherits the source session's `cwd`, stamps `parentSession` to the source id, and sets `seedLength` to the copied prefix length. When `childSessionId` is omitted, `SessionStore` generates one using its existing id policy.
The boundary rule is structural: an empty selected prefix is forkable, and any non-empty selected prefix must be turn-enclosed and end at `turn/end`, regardless of the turn-end reason (`completed`, `aborted`, `error`, `disposed`, `max-tokens`, `interrupted`, or a future merge-extensible reason). Any selected prefix whose boundary is not an existing event seq, ends inside a turn, contains events outside a turn, contains nested turns, or has an orphan `turn/end` is rejected with a typed `SessionForkError` code. The API also classifies non-live source ids (`SESSION_NOT_FOUND`), stale `Session` object references whose id is live on a different instance (`SESSION_NOT_LIVE`), duplicate requested child ids (`SESSION_ALREADY_EXISTS`), and invalid boundary values (`INVALID_BOUNDARY`).
## Alternatives considered
**Separate `ctx.sessionFork` service.** This was the first implementation, but review showed it overfit the capability-seam pattern. The code had no swappable backend, no extra event surface, no independent ownership lifecycle, and no durable behavior beyond `ctx.sessions.create({ seed, meta })`. Keeping a separate package would make callers discover and install a second service just to perform policy around a session-store primitive.
**Two functions: `snapshot()` plus `fork()`.** This preserved a reusable seed/metadata computation, but the only supported consumer created a session immediately. It also made the surface feel more abstract than the concrete operation users need. A single `fork()` with an explicit `boundary` keeps the API direct while still supporting previous-point forks.
**Silently clip open turns to the last completed boundary.** That is correct for `dsh-subagent-fork`, where delegation often starts while the parent turn is open and the child should inherit only the completed prefix. It is wrong for ordinary user/session branching because it hides that the requested fork point was not actually a valid boundary and silently drops the parent turn tail.
## Consequences
The public surface stays small and discoverable: live session branching is part of `ctx.sessions`, next to `create({ seed })`, rather than a standalone service or a two-step helper pair. Persistence continues to work through existing `session/created` and `session/flush` behavior: a forked child starts life with seeded events, so existing backends persist that seed once and preserve `parentSession` / `seedLength` in the header.
The v1 scope still excludes ACP `session/fork`, unloaded persisted-session forking, model-facing tools, and subagent refactors. If a future ACP method is added, it should advertise the capability only after it has transcript/snapshot coverage; this RFC adds no editor-facing updates, so no ACP snapshot is required now. Fork-child replay remains covered by the existing [seed-boundary testing RFC](../../implemented/testing/2026-06-22-fork-child-replay-seed-boundary.md), while this API gets focused `dsh-session` unit tests plus JSONL persistence coverage.

View File

@@ -1,59 +0,0 @@
# RFC: SessionStore fork helpers
Status: implemented
## Problem
The event-sourced session log already has the primitive a fork needs: create a new session with a seed event prefix, then derive model history from that seeded log exactly as replay does. That primitive is intentionally low-level: `ctx.sessions.create(id, { seed, meta })` accepts any valid seed, but ordinary live-session branching needs policy around where the seed may be taken, which metadata is stamped on the child, and how errors are classified.
The semantic hazard is the fork boundary. A session event log is only a valid user-visible fork seed when it is contiguous and balanced. Forking inside an active turn would copy an open `turn/start`, possibly an open `step/start`, and possibly dangling tool calls. That violates the turn-enclosure and provider-transcript invariants, and it creates a misleading child history that appears to have participated in an unfinished parent turn. The existing [subagent seam](../../implemented/feature/2026-06-21-subagent-capability-seam.md) deliberately solves a different problem: a tool-triggered subagent fork usually happens while the parent turn is open, so `dsh-subagent-fork` clips the seed to the parent's last completed-turn prefix. A general session fork should not silently clip; it should reject attempts made away from a boundary.
## Decision
`dsh-session` owns ordinary live-session fork helpers directly on `ctx.sessions`. There is no separate `dsh-session-fork` package or `ctx.sessionFork` service: the helpers have no independent backend, event vocabulary, lifecycle, or persistence behavior, and all durable work delegates to the existing session store and persistence backends.
The store exposes two operations:
```ts ignore-check
type SessionForkSource = Session | SessionId
interface SessionForkSeed {
source: Session
seed: SessionEvent[]
meta: {
parentSession: SessionId
seedLength: number
cwd?: string
}
}
interface ForkSessionOptions {
source: SessionForkSource
sessionId?: SessionId
}
class SessionStore extends Service {
snapshot(source: SessionForkSource): SessionForkSeed
fork(options: ForkSessionOptions): Session
}
```
`snapshot()` is the reusable half. It resolves only live sessions from `ctx.sessions`; v1 does not load unloaded persisted sessions by id. It validates the source is at a turn boundary, deep-clones the source events, and returns the seed plus metadata a caller can pass to a later session or agent creation path. This keeps the fork computation reusable for future ACP or agent-facing consumers without coupling `dsh-session` to `ctx.agents`.
`fork()` is the convenience half. It calls `snapshot()`, then creates a live child session via `ctx.sessions.create(sessionId, { seed, meta })`. The child inherits the source session's `cwd`, stamps `parentSession` to the source id, and sets `seedLength` to the seeded prefix length. When `sessionId` is omitted, `SessionStore` generates one using its existing id policy.
The boundary rule is structural: an empty source log is forkable, and any source whose last event is `turn/end` is forkable regardless of the turn-end reason (`completed`, `aborted`, `error`, `disposed`, `max-tokens`, `interrupted`, or a future merge-extensible reason). Any non-empty source whose last event is not `turn/end` is inside a turn or otherwise not at the boundary and is rejected with a typed `SessionForkError` code. The helpers also classify non-live source ids (`SESSION_NOT_FOUND`), stale `Session` object references whose id is live on a different instance (`SESSION_NOT_LIVE`), and duplicate requested child ids (`SESSION_ALREADY_EXISTS`) instead of leaking lower-level store errors.
## Alternatives considered
**Separate `ctx.sessionFork` service.** This was the first implementation, but review showed it overfit the capability-seam pattern. The code had no swappable backend, no extra event surface, no independent ownership lifecycle, and no durable behavior beyond `ctx.sessions.create({ seed, meta })`. Keeping a separate package would make callers discover and install a second service just to perform policy around a session-store primitive.
**Only expose `fork()`.** A one-function API is simpler for immediate child-session creation, but it forces callers that need a seed for another creation path to create a detached child session just to get the seed. `snapshot()` keeps the seed/metadata computation reusable without importing `ctx.agents` into `dsh-session`; `fork()` remains the simple one-call convenience.
**Silently clip open turns to the last completed boundary.** That is correct for `dsh-subagent-fork`, where delegation often starts while the parent turn is open and the child should inherit only the completed prefix. It is wrong for ordinary user/session branching because it hides that the requested fork point was not actually a valid boundary and silently drops the parent turn tail.
## Consequences
The public surface stays small and discoverable: live session branching is part of `ctx.sessions`, next to `create({ seed })`, rather than a standalone service. Persistence continues to work through existing `session/created` and `session/flush` behavior: a forked child starts life with seeded events, so existing backends persist that seed once and preserve `parentSession` / `seedLength` in the header.
The v1 scope still excludes ACP `session/fork`, unloaded persisted-session forking, model-facing tools, and subagent refactors. Those can consume `snapshot()` later. If a future ACP method is added, it should advertise the capability only after it has transcript/snapshot coverage; this RFC adds no editor-facing updates, so no ACP snapshot is required now. Fork-child replay remains covered by the existing [seed-boundary testing RFC](../../implemented/testing/2026-06-22-fork-child-replay-seed-boundary.md), while these helpers get focused `dsh-session` unit tests plus JSONL persistence coverage.

View File

@@ -9,8 +9,7 @@ Creates and holds event-sourced `Session` instances. Persistence is intentionall
### Public API ### 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.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.snapshot(source: Session | SessionId): SessionForkSeed` — Resolve a live session object or id, reject non-boundary logs, and return a deep-cloned seed plus `parentSession` / `seedLength` metadata. Use this when the caller will pass the seed/meta into another creation path instead of creating a detached session immediately. - `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 selected prefix to be turn-enclosed, and create a live child session with lineage metadata.
- `ctx.sessions.fork({ source, sessionId? }): Session` — Convenience wrapper around `snapshot(source)` + `create(sessionId, { seed, meta })`; creates a live child session with lineage metadata.
- `ctx.sessions.get(id: SessionId): Session | undefined` - `ctx.sessions.get(id: SessionId): Session | undefined`
- `ctx.sessions.list(): Session[]` - `ctx.sessions.list(): Session[]`
@@ -69,9 +68,9 @@ Every `SessionEvent` carries two optional top-level fields (structural metadata)
### Extension points ### 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. - 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 invariants `append` enforces — including that every surface-eligible event (`SurfaceEventType`) carries a `surfaceOp` marker — so a marker-less message event is rejected at construction rather than silently vanishing from `deriveMessages()` (the surface is the sole derivation path) on resume. Ordinary live-session forks use `ctx.sessions.snapshot()` to validate an empty or `turn/end` boundary and build reusable seed metadata, or `ctx.sessions.fork()` to create the child session immediately. - 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 invariants `append` enforces — including that every surface-eligible event (`SurfaceEventType`) carries a `surfaceOp` marker — so a marker-less message event is rejected at construction rather than silently vanishing from `deriveMessages()` (the surface is the sole derivation path) on resume. 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. - 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) ### What is NOT here (TODO)
- **Session branching/tree** (pi-style entry tree) — deferred unless needed beyond turn-boundary `snapshot()` / `fork()`. - **Session branching/tree** (pi-style entry tree) — deferred unless needed beyond boundary-based `fork()`.

View File

@@ -321,35 +321,24 @@ export class Session {
/** A fork source: either the live session object or its live store id. */ /** A fork source: either the live session object or its live store id. */
export type SessionForkSource = Session | SessionId export type SessionForkSource = Session | SessionId
/** Metadata and seed events that can create a forked child session or agent. */ /** Inputs for live session forking. */
export interface SessionForkSeed {
/** The resolved live source session. */
source: Session
/** Deep-cloned seed events copied from the source session at a turn boundary. */
seed: SessionEvent[]
/** Session creation metadata for the forked child. */
meta: {
/** The source session id. */
parentSession: SessionId
/** How many leading child events were inherited rather than produced. */
seedLength: number
/** The source session workspace, inherited by the child when present. */
cwd?: string
}
}
/** Inputs for the convenience session-creation path. */
export interface ForkSessionOptions { export interface ForkSessionOptions {
/** Live source session object or id. */ /** Live source session object or id. */
source: SessionForkSource 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. */ /** Optional child session id; omitted delegates to SessionStore's id policy. */
sessionId?: SessionId childSessionId?: SessionId
} }
export type SessionForkErrorCode = export type SessionForkErrorCode =
| 'SESSION_NOT_FOUND' | 'SESSION_NOT_FOUND'
| 'SESSION_NOT_LIVE' | 'SESSION_NOT_LIVE'
| 'SESSION_ALREADY_EXISTS' | 'SESSION_ALREADY_EXISTS'
| 'INVALID_BOUNDARY'
| 'OPEN_TURN' | 'OPEN_TURN'
/** Typed error for session fork rejections. */ /** Typed error for session fork rejections. */
@@ -496,47 +485,67 @@ export class SessionStore extends Service {
} }
/** /**
* Resolve and validate a live source session, then return a reusable deep- * Create a live child session from a turn-enclosed prefix of a live source.
* cloned fork seed. A non-empty source must end exactly at `turn/end`; this * `boundary` is an inclusive source event seq; omitted means the source's
* rejects open turns rather than clipping to an older boundary. * current last event. A non-empty selected slice must be turn-enclosed and end
* at `turn/end`; this rejects open turns rather than clipping silently.
* *
* @param source Live session object or live store id to snapshot. * @param options Source, optional boundary, and optional child id for the fork.
* @returns Deep-cloned seed events plus child session metadata.
*/
snapshot(source: SessionForkSource): SessionForkSeed {
const session = this._resolveForkSource(source)
this._assertForkBoundary(session)
const seed = session.events.map(event => structuredClone(event))
return {
source: session,
seed,
meta: {
...session.header.cwd !== undefined ? { cwd: session.header.cwd } : {},
parentSession: session.id,
seedLength: seed.length,
},
}
}
/**
* Convenience path: create a live child session from a fork snapshot. Callers
* that create agents can use {@link snapshot} and pass its seed/meta through
* `ctx.agents.create` instead.
*
* @param options Source and optional child session id for the fork.
* @returns The created live child session. * @returns The created live child session.
*/ */
fork(options: ForkSessionOptions): Session { fork(options: ForkSessionOptions): Session {
if (options.sessionId !== undefined && this.get(options.sessionId) !== undefined) { if (options.childSessionId !== undefined && this.get(options.childSessionId) !== undefined) {
throw new SessionForkError(`session "${options.sessionId}" already exists`, 'SESSION_ALREADY_EXISTS') throw new SessionForkError(`session "${options.childSessionId}" already exists`, 'SESSION_ALREADY_EXISTS')
} }
const snapshot = this.snapshot(options.source) const source = this._resolveForkSource(options.source)
return this.create(options.sessionId, { const seed = this._forkSeed(source, options.boundary)
seed: snapshot.seed, return this.create(options.childSessionId, {
meta: snapshot.meta, seed,
meta: {
...source.header.cwd !== undefined ? { cwd: source.header.cwd } : {},
parentSession: source.id,
seedLength: seed.length,
},
}) })
} }
private _forkSeed(session: Session, requestedBoundary: number | undefined): SessionEvent[] {
const events = session.events
const lastEvent = events.at(-1)
let boundary: number
if (requestedBoundary !== undefined) {
boundary = requestedBoundary
} else {
if (lastEvent === undefined) return []
boundary = lastEvent.seq
}
if (!Number.isSafeInteger(boundary) || boundary < 0) {
throw new SessionForkError(
`fork boundary for session "${session.id}" must be a non-negative safe integer, got ${String(boundary)}`,
'INVALID_BOUNDARY',
)
}
if (boundary >= events.length) {
const lastSeq = events.at(-1)?.seq
throw new SessionForkError(
`fork boundary ${boundary} does not exist in session "${session.id}" (last seq: ${lastSeq ?? 'none'})`,
'INVALID_BOUNDARY',
)
}
const boundaryEvent = events[boundary]
if (boundaryEvent === undefined || boundaryEvent.seq !== boundary) {
throw new SessionForkError(
`fork boundary ${boundary} does not match a contiguous event seq in session "${session.id}"`,
'INVALID_BOUNDARY',
)
}
const seed = events.slice(0, boundary + 1)
this._assertForkBoundary(session, seed, boundary)
return seed.map(event => structuredClone(event))
}
private _resolveForkSource(source: SessionForkSource): Session { private _resolveForkSource(source: SessionForkSource): Session {
if (typeof source === 'string') { if (typeof source === 'string') {
const session = this.get(source) const session = this.get(source)
@@ -552,11 +561,46 @@ export class SessionStore extends Service {
return source return source
} }
private _assertForkBoundary(session: Session): void { private _assertForkBoundary(session: Session, seed: readonly SessionEvent[], boundary: number): void {
const last = session.events.at(-1) let openTurn: SessionEvent<'turn/start'> | undefined
if (last !== undefined && last.type !== 'turn/end') { for (const event of seed) {
switch (event.type) {
case 'turn/start': {
if (openTurn !== undefined) {
throw new SessionForkError(
`cannot fork session "${session.id}" at boundary ${boundary}: turn ${event.data.turn} starts before turn ${openTurn.data.turn} ended`,
'OPEN_TURN',
)
}
openTurn = event
break
}
case 'turn/end': {
if (openTurn === undefined) {
throw new SessionForkError(
`cannot fork session "${session.id}" at boundary ${boundary}: turn/end at seq ${event.seq} has no matching turn/start`,
'OPEN_TURN',
)
}
openTurn = undefined
break
}
default: {
if (openTurn === undefined) {
throw new SessionForkError(
`cannot fork session "${session.id}" at boundary ${boundary}: event ${event.seq} (${event.type}) is outside a turn`,
'OPEN_TURN',
)
}
break
}
}
}
const last = seed.at(-1)
if (openTurn !== undefined || last?.type !== 'turn/end') {
throw new SessionForkError( throw new SessionForkError(
`cannot fork session "${session.id}" inside an open turn (last event: ${last.type})`, `cannot fork session "${session.id}" at boundary ${boundary}: slice ends inside an open turn (last event: ${last?.type ?? 'none'})`,
'OPEN_TURN', 'OPEN_TURN',
) )
} }

View File

@@ -10,13 +10,26 @@ async function setup(): Promise<{ ctx: Context; sessions: SessionStore }> {
return { ctx, sessions: ctx.sessions } return { ctx, sessions: ctx.sessions }
} }
function appendClosedTurn(session: Session, reason: TurnEndReason = { kind: 'completed' }): void { function appendClosedTurn(
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) session: Session,
turn: number,
text = `hello ${turn}`,
reason: TurnEndReason = { kind: 'completed' },
): void {
session.append('turn/start', { turn, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('user/message', { session.append('user/message', {
content: [{ type: 'text', text: 'hello' }], content: [{ type: 'text', text }],
source: { kind: 'user' },
}, { surfaceOp: 'append' })
session.append('turn/end', { turn, reason })
}
function appendOpenTurn(session: Session, turn: number): void {
session.append('turn/start', { turn, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('user/message', {
content: [{ type: 'text', text: `open ${turn}` }],
source: { kind: 'user' }, source: { kind: 'user' },
}, { surfaceOp: 'append' }) }, { surfaceOp: 'append' })
session.append('turn/end', { turn: 1, reason })
} }
function firstUserMessage(events: readonly SessionEvent[]): SessionEvent<'user/message'> { function firstUserMessage(events: readonly SessionEvent[]): SessionEvent<'user/message'> {
@@ -25,43 +38,68 @@ function firstUserMessage(events: readonly SessionEvent[]): SessionEvent<'user/m
return event return event
} }
describe('SessionStore fork helpers', () => { function lastSeq(session: Session): number {
it('snapshots an empty live session as an empty seed with lineage metadata', async () => { const event = session.events.at(-1)
if (event === undefined) throw new Error('missing last event')
return event.seq
}
describe('SessionStore.fork', () => {
it('forks an empty live session as an empty child with lineage metadata', async () => {
const { ctx, sessions } = await setup() const { ctx, sessions } = await setup()
const source = ctx.sessions.create(SessionId('empty-parent'), { meta: { cwd: '/workspace' } }) const source = ctx.sessions.create(SessionId('empty-parent'), { meta: { cwd: '/workspace' } })
const snapshot = sessions.snapshot(source) const child = sessions.fork({ source, childSessionId: SessionId('empty-child') })
expect(snapshot.source).toBe(source) expect(child.events).toEqual([])
expect(snapshot.seed).toEqual([]) expect(child.header).toMatchObject({
expect(snapshot.meta).toEqual({ id: SessionId('empty-child'),
cwd: '/workspace', cwd: '/workspace',
parentSession: SessionId('empty-parent'), parentSession: SessionId('empty-parent'),
seedLength: 0, seedLength: 0,
}) })
}) })
it('snapshots a completed boundary by live session id and deep-clones seed events', async () => { it('forks the latest completed boundary by default and deep-clones seed events', async () => {
const { ctx, sessions } = await setup() const { ctx, sessions } = await setup()
const source = ctx.sessions.create(SessionId('parent'), { meta: { cwd: '/workspace' } }) const source = ctx.sessions.create(SessionId('parent'), { meta: { cwd: '/workspace' } })
appendClosedTurn(source) appendClosedTurn(source, 1, 'hello')
const snapshot = sessions.snapshot(SessionId('parent')) const child = sessions.fork({ source: SessionId('parent'), childSessionId: SessionId('child') })
expect(snapshot.source).toBe(source) expect(child.events).toEqual(source.events)
expect(snapshot.seed).toEqual(source.events) expect(child.events).not.toBe(source.events)
expect(snapshot.seed).not.toBe(source.events) expect(child.events[1]).not.toBe(source.events[1])
expect(snapshot.seed[1]).not.toBe(source.events[1]) firstUserMessage(child.events).data.content[0] = { type: 'text', text: 'child mutation' }
firstUserMessage(snapshot.seed).data.content[0] = { type: 'text', text: 'mutated' }
expect(firstUserMessage(source.events).data.content).toEqual([{ type: 'text', text: 'hello' }]) expect(firstUserMessage(source.events).data.content).toEqual([{ type: 'text', text: 'hello' }])
expect(snapshot.meta).toEqual({ expect(child.header).toMatchObject({
id: SessionId('child'),
cwd: '/workspace', cwd: '/workspace',
parentSession: SessionId('parent'), parentSession: SessionId('parent'),
seedLength: source.events.length, seedLength: source.events.length,
}) })
}) })
it('accepts every turn/end reason as a fork boundary', async () => { it('forks from an earlier turn boundary even when the source currently has an open tail', async () => {
const { ctx, sessions } = await setup()
const source = ctx.sessions.create(SessionId('parent'), { meta: { cwd: '/workspace' } })
appendClosedTurn(source, 1, 'first')
const firstBoundary = lastSeq(source)
appendClosedTurn(source, 2, 'second')
appendOpenTurn(source, 3)
const child = sessions.fork({
source,
boundary: firstBoundary,
childSessionId: SessionId('child-from-first'),
})
expect(child.events).toEqual(source.events.slice(0, firstBoundary + 1))
expect(child.header.seedLength).toBe(firstBoundary + 1)
expect(child.deriveMessages()).toEqual([{ role: 'user', content: [{ type: 'text', text: 'first' }] }])
})
it('accepts every turn/end reason as an explicit fork boundary', async () => {
const { ctx, sessions } = await setup() const { ctx, sessions } = await setup()
const reasons: TurnEndReason[] = [ const reasons: TurnEndReason[] = [
{ kind: 'completed' }, { kind: 'completed' },
@@ -74,19 +112,42 @@ describe('SessionStore fork helpers', () => {
for (const reason of reasons) { for (const reason of reasons) {
const source = ctx.sessions.create(SessionId(`parent-${reason.kind}`)) const source = ctx.sessions.create(SessionId(`parent-${reason.kind}`))
appendClosedTurn(source, reason) appendClosedTurn(source, 1, reason.kind, reason)
const snapshot = sessions.snapshot(source) const child = sessions.fork({
source,
boundary: lastSeq(source),
childSessionId: SessionId(`child-${reason.kind}`),
})
expect(snapshot.seed.at(-1)?.type).toBe('turn/end') expect(child.events.at(-1)?.type).toBe('turn/end')
expect(snapshot.meta.seedLength).toBe(source.events.length) expect(child.header.seedLength).toBe(source.events.length)
} }
}) })
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') }))
.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') }))
.toThrow(/non-negative safe integer/)
expect(() => sessions.fork({ source, boundary: 0.5, childSessionId: SessionId('fraction') }))
.toThrow(/non-negative safe integer/)
expect(() => sessions.fork({ source, boundary: Number.MAX_SAFE_INTEGER + 1, childSessionId: SessionId('unsafe') }))
.toThrow(/non-negative safe integer/)
expect(() => sessions.fork({ source, boundary: source.seq, childSessionId: SessionId('past-end') }))
.toThrow(new SessionForkError(`fork boundary ${source.seq} does not exist in session "parent" (last seq: ${source.seq - 1})`, 'INVALID_BOUNDARY'))
})
it('rejects an unknown live session id', async () => { it('rejects an unknown live session id', async () => {
const { sessions } = await setup() const { sessions } = await setup()
expect(() => sessions.snapshot(SessionId('missing'))) expect(() => sessions.fork({ source: SessionId('missing') }))
.toThrow(new SessionForkError('session "missing" not found', 'SESSION_NOT_FOUND')) .toThrow(new SessionForkError('session "missing" not found', 'SESSION_NOT_FOUND'))
}) })
@@ -94,7 +155,7 @@ describe('SessionStore fork helpers', () => {
const { sessions } = await setup() const { sessions } = await setup()
const detached = new Session(SessionId('detached')) const detached = new Session(SessionId('detached'))
expect(() => sessions.snapshot(detached)) expect(() => sessions.fork({ source: detached }))
.toThrow(new SessionForkError('session "detached" not found', 'SESSION_NOT_FOUND')) .toThrow(new SessionForkError('session "detached" not found', 'SESSION_NOT_FOUND'))
}) })
@@ -103,28 +164,32 @@ describe('SessionStore fork helpers', () => {
ctx.sessions.create(SessionId('same-id')) ctx.sessions.create(SessionId('same-id'))
const stale = new Session(SessionId('same-id')) const stale = new Session(SessionId('same-id'))
expect(() => sessions.snapshot(stale)) expect(() => sessions.fork({ source: stale }))
.toThrow(new SessionForkError('session "same-id" is not the live store instance', 'SESSION_NOT_LIVE')) .toThrow(new SessionForkError('session "same-id" is not the live store instance', 'SESSION_NOT_LIVE'))
}) })
it('rejects non-empty logs whose last event is not turn/end', async () => { it('rejects selected slices whose boundary is inside an open turn', async () => {
const { ctx, sessions } = await setup() const { ctx, sessions } = await setup()
const cases: [string, (session: Session) => void][] = [ const cases: [string, (session: Session) => number][] = [
['turn/start', (session) => { ['turn/start', (session) => {
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
return lastSeq(session)
}], }],
['step/start', (session) => { ['step/start', (session) => {
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('step/start', { turn: 1, step: 1 }) session.append('step/start', { turn: 1, step: 1 })
return lastSeq(session)
}], }],
['user/message', (session) => { ['user/message', (session) => {
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('user/message', { content: [{ type: 'text', text: 'open' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) session.append('user/message', { content: [{ type: 'text', text: 'open' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
return lastSeq(session)
}], }],
['assistant/message', (session) => { ['assistant/message', (session) => {
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('step/start', { turn: 1, step: 1 }) session.append('step/start', { turn: 1, step: 1 })
session.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'partial' }] }, { surfaceOp: 'append' }) session.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'partial' }] }, { surfaceOp: 'append' })
return lastSeq(session)
}], }],
['tool/call', (session) => { ['tool/call', (session) => {
const callId = CallId('call-open') const callId = CallId('call-open')
@@ -136,51 +201,64 @@ describe('SessionStore fork helpers', () => {
content: [{ type: 'tool-call', id: callId, name: 'bash', arguments: '{}' }], content: [{ type: 'tool-call', id: callId, name: 'bash', arguments: '{}' }],
}, { surfaceOp: 'append' }) }, { surfaceOp: 'append' })
session.append('tool/call', { turn: 1, step: 1, callId, name: 'bash', arguments: '{}' }) session.append('tool/call', { turn: 1, step: 1, callId, name: 'bash', arguments: '{}' })
return lastSeq(session)
}], }],
] ]
for (const [lastType, build] of cases) { for (const [lastType, build] of cases) {
const source = ctx.sessions.create(SessionId(`open-${lastType}`)) const source = ctx.sessions.create(SessionId(`open-${lastType}`))
build(source) const boundary = build(source)
expect(() => sessions.snapshot(source)) expect(() => sessions.fork({ source, boundary }))
.toThrow(new SessionForkError(`cannot fork session "open-${lastType}" inside an open turn (last event: ${lastType})`, 'OPEN_TURN')) .toThrow(new SessionForkError(`cannot fork session "open-${lastType}" at boundary ${boundary}: slice ends inside an open turn (last event: ${lastType})`, 'OPEN_TURN'))
} }
}) })
it('creates a forked child session with the seed and lineage metadata', async () => { it('rejects malformed turn enclosure in the selected slice', async () => {
const { ctx, sessions } = await setup() const { ctx, sessions } = await setup()
const source = ctx.sessions.create(SessionId('parent'), { meta: { cwd: '/workspace' } }) const outside = ctx.sessions.create(SessionId('outside'), {
appendClosedTurn(source) seed: [
{ type: 'step/start', seq: 0, time: 1, data: { turn: 1, step: 1 } },
],
})
expect(() => sessions.fork({ source: outside, boundary: 0 }))
.toThrow(new SessionForkError('cannot fork session "outside" at boundary 0: event 0 (step/start) is outside a turn', 'OPEN_TURN'))
const child = sessions.fork({ source, sessionId: SessionId('child') }) const nested = ctx.sessions.create(SessionId('nested'), {
seed: [
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
{ type: 'turn/start', seq: 1, time: 2, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
],
})
expect(() => sessions.fork({ source: nested, boundary: 1 }))
.toThrow(new SessionForkError('cannot fork session "nested" at boundary 1: turn 2 starts before turn 1 ended', 'OPEN_TURN'))
expect(child.id).toBe(SessionId('child')) const orphanEnd = ctx.sessions.create(SessionId('orphan-end'), {
expect(child.events).toEqual(source.events) seed: [
expect(child.header.parentSession).toBe(source.id) { type: 'turn/end', seq: 0, time: 1, data: { turn: 1, reason: { kind: 'completed' } } },
expect(child.header.seedLength).toBe(source.events.length) ],
expect(child.header.cwd).toBe('/workspace') })
firstUserMessage(child.events).data.content[0] = { type: 'text', text: 'child mutation' } expect(() => sessions.fork({ source: orphanEnd, boundary: 0 }))
expect(firstUserMessage(source.events).data.content).toEqual([{ type: 'text', text: 'hello' }]) .toThrow(new SessionForkError('cannot fork session "orphan-end" at boundary 0: turn/end at seq 0 has no matching turn/start', 'OPEN_TURN'))
}) })
it('rejects a child session id that is already live with a typed fork error', async () => { it('rejects a child session id that is already live with a typed fork error', async () => {
const { ctx, sessions } = await setup() const { ctx, sessions } = await setup()
const source = ctx.sessions.create(SessionId('parent')) const source = ctx.sessions.create(SessionId('parent'))
appendClosedTurn(source) appendClosedTurn(source, 1)
ctx.sessions.create(SessionId('child')) ctx.sessions.create(SessionId('child'))
expect(() => sessions.fork({ source, sessionId: SessionId('child') })) expect(() => sessions.fork({ source, childSessionId: SessionId('child') }))
.toThrow(new SessionForkError('session "child" already exists', 'SESSION_ALREADY_EXISTS')) .toThrow(new SessionForkError('session "child" already exists', 'SESSION_ALREADY_EXISTS'))
}) })
it('rejects a duplicate child session id before validating the source boundary', async () => { it('rejects a duplicate child session id before validating the boundary', async () => {
const { ctx, sessions } = await setup() const { ctx, sessions } = await setup()
const source = ctx.sessions.create(SessionId('open-parent')) const source = ctx.sessions.create(SessionId('open-parent'))
source.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) source.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
ctx.sessions.create(SessionId('child')) ctx.sessions.create(SessionId('child'))
expect(() => sessions.fork({ source, sessionId: SessionId('child') })) expect(() => sessions.fork({ source, childSessionId: SessionId('child') }))
.toThrow(new SessionForkError('session "child" already exists', 'SESSION_ALREADY_EXISTS')) .toThrow(new SessionForkError('session "child" already exists', 'SESSION_ALREADY_EXISTS'))
}) })
}) })

View File

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