diff --git a/CLAUDE.md b/CLAUDE.md index 71270822..75116458 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -38,7 +38,7 @@ A run executes in one of **three execution modes** behind the one `LLMProvider` engine is identical across all three. See [ADR-0012](docs/decisions/0012-managed-inference-dual-mode.md) to [ADR-0015](docs/decisions/0015-managed-mode-data-handling-and-compliance.md) and [docs/architecture/managed-inference.md](docs/architecture/managed-inference.md). -**Status:** Phase 1 in progress — milestone M1 (LLM seam proven) reached (PR #9, 2026-06-07); the `FallbackChain` runner (1.K) landed, completing 1.m2 with the cost tracker (PR #13, 2026-06-11); the run loop (1.N — `WorkflowEngine` + `RunEventBus`) landed (PR #17, 2026-06-13) **completing 1.m3** (parse → DAG → run loop emits the canonical event stream), with the built-in `ToolRegistry` (1.T, a 1.m4 component) landing alongside it as the other `AgentRunner` (1.O) join prerequisite; the **`AgentRunner` (1.O) — per-node LLM execution behind the seam — landed (PR #18, 2026-06-14)**; and the **node-type handlers (1.P) — the six non-agent `NodeExecutor` arms (condition / transform / fan_out / fan_in / input / output) behind a dispatching executor — landed (PR #20, 2026-06-14)**; and **checkpoint/resume (1.R) + the human gate (1.Q) landed (PR #22, 2026-06-15)** — the derived `Checkpointer` + cross-process `resumeFromCheckpoint`, and the `human_in_the_loop` gate with the one-shot timeout port; and **node retry (1.S) — the above-chain whole-node retry budget ([ADR-0040](docs/decisions/0040-node-retry-budget-above-the-chain.md), amending ADR-0038) — landed (PR #24, 2026-06-15)**, re-dispatching a whole node on a retryable failure up to `retry.max` attempts (with `node:retrying`, abort-aware backoff, and `retry_on` filtering), with retry-from-node (ADR-0040 Part B) deferred to Phase-2; and the **pre-egress budget governor (1.AC, [ADR-0028](docs/decisions/0028-workflow-resource-governance.md)) + the `AgentSession` agent-first entry point (1.V, [ADR-0024](docs/decisions/0024-agent-first-entry-point-agentsession.md)) landed together (PR #26, 2026-06-16)** — 1.AC was the last 1.m4 component, so **1.m4 is complete** (the full engine stack: node handlers, gate, checkpoint/resume, retry, tools, sandbox, budget governor), and 1.V opens the Lane-C agent-first sub-spine (1.m5); then the **end-to-end Node harness (1.U) landed (PR #27, 2026-06-16), reaching 🎯 M2** — the engine runs end-to-end (live streaming + per-node-boundary checkpointing + cross-process resume + node retry + provider failover, gap-free), **completing the Phase-1 engine critical path**. The remaining Phase-1 work is additive and off the critical path (Lane C: the **`session:*` namespace (1.W) has since landed (PR #28, 2026-06-17)** — the `SessionEventSink`→`RunEventBus` adapter + per-session `sequenceNumber`, the `SessionHandle`, and the combined `RunOrSessionEventSchema` gate — leaving persistence 1.X next ‖ 1.Y/1.Z/1.AA; the 1.m6 multimodal sub-spine); **Phase 2 (CLI, M3) is unblocked**. +**Status:** Phase 1 in progress — milestone M1 (LLM seam proven) reached (PR #9, 2026-06-07); the `FallbackChain` runner (1.K) landed, completing 1.m2 with the cost tracker (PR #13, 2026-06-11); the run loop (1.N — `WorkflowEngine` + `RunEventBus`) landed (PR #17, 2026-06-13) **completing 1.m3** (parse → DAG → run loop emits the canonical event stream), with the built-in `ToolRegistry` (1.T, a 1.m4 component) landing alongside it as the other `AgentRunner` (1.O) join prerequisite; the **`AgentRunner` (1.O) — per-node LLM execution behind the seam — landed (PR #18, 2026-06-14)**; and the **node-type handlers (1.P) — the six non-agent `NodeExecutor` arms (condition / transform / fan_out / fan_in / input / output) behind a dispatching executor — landed (PR #20, 2026-06-14)**; and **checkpoint/resume (1.R) + the human gate (1.Q) landed (PR #22, 2026-06-15)** — the derived `Checkpointer` + cross-process `resumeFromCheckpoint`, and the `human_in_the_loop` gate with the one-shot timeout port; and **node retry (1.S) — the above-chain whole-node retry budget ([ADR-0040](docs/decisions/0040-node-retry-budget-above-the-chain.md), amending ADR-0038) — landed (PR #24, 2026-06-15)**, re-dispatching a whole node on a retryable failure up to `retry.max` attempts (with `node:retrying`, abort-aware backoff, and `retry_on` filtering), with retry-from-node (ADR-0040 Part B) deferred to Phase-2; and the **pre-egress budget governor (1.AC, [ADR-0028](docs/decisions/0028-workflow-resource-governance.md)) + the `AgentSession` agent-first entry point (1.V, [ADR-0024](docs/decisions/0024-agent-first-entry-point-agentsession.md)) landed together (PR #26, 2026-06-16)** — 1.AC was the last 1.m4 component, so **1.m4 is complete** (the full engine stack: node handlers, gate, checkpoint/resume, retry, tools, sandbox, budget governor), and 1.V opens the Lane-C agent-first sub-spine (1.m5); then the **end-to-end Node harness (1.U) landed (PR #27, 2026-06-16), reaching 🎯 M2** — the engine runs end-to-end (live streaming + per-node-boundary checkpointing + cross-process resume + node retry + provider failover, gap-free), **completing the Phase-1 engine critical path**. The remaining Phase-1 work is additive and off the critical path (Lane C: the **`session:*` namespace (1.W) landed (PR #28, 2026-06-17)** — the `SessionEventSink`→`RunEventBus` adapter + per-session `sequenceNumber`, the `SessionHandle`, and the combined `RunOrSessionEventSchema` gate — and **session persistence (1.X) landed (PR #29, 2026-06-17)** — the `agent_sessions`/`session_messages` tables + migration, `SessionMessageSchema`/`AgentSessionSchema`, and the `SessionStore` + domain↔row mappers (data-layer only; the per-turn `AgentSession`→store wiring + resume are 1.Y/1.AA) — leaving 1.Y/1.Z/1.AA next ‖ the 1.m6 multimodal sub-spine); **Phase 2 (CLI, M3) is unblocked**. Phase 0 (M0, 2026-06-04) landed the monorepo, strict toolchain + CI, `@relavium/shared` (the full Zod contract set), the no-vendor-type seam fence, and `@relavium/db`. Phase 1 has since landed `@relavium/llm` — the `LLMProvider` seam + all three adapters (Anthropic, OpenAI/DeepSeek, @@ -63,8 +63,8 @@ plus the one-shot `setTimer` timeout port — `approve` auto-resolves, `reject` The pre-egress budget governor (1.AC) + the agent-first `AgentSession` (1.V) landed together (PR #26) **completing 1.m4**; then the end-to-end Node harness (1.U) landed (PR #27, 2026-06-16) **reaching M2** — the Phase-1 engine critical path is complete. Active work is now the additive Lane-C sub-spine (session -events **1.W ✅ Done (PR #28, 2026-06-17)**; persistence 1.X next) and the multimodal sub-spine; Phase 2 -(CLI) is unblocked. See +events **1.W ✅ Done (PR #28)** + persistence **1.X ✅ Done (PR #29, 2026-06-17)**; 1.Y/1.Z/1.AA next) and +the multimodal sub-spine; Phase 2 (CLI) is unblocked. See [docs/roadmap/current.md](docs/roadmap/current.md). See [README.md](README.md) for the public overview. ## Non-negotiable rules for AI agents diff --git a/README.md b/README.md index 7c8aa4d0..d1d88b2a 100644 --- a/README.md +++ b/README.md @@ -52,6 +52,6 @@ both landed in **PR #26 (2026-06-16)**. With the budget governor in, **milestone **end-to-end Node harness** (1.U) landed (**PR #27, 2026-06-16**) — **🎯 reaching milestone M2**: the engine runs end-to-end (live streaming + checkpoint + cross-process resume + retry + provider failover), completing the Phase-1 engine critical path. The remaining Phase-1 work is additive (the agent-first sub-spine — **session -events 1.W ✅ (PR #28, 2026-06-17)**, persistence 1.X next, … — and the multimodal sub-spine); **Phase 2 (the -CLI) is unblocked**. See +events 1.W ✅ (PR #28)** + **persistence 1.X ✅ (PR #29, 2026-06-17)**, 1.Y/1.Z/1.AA next … — and the +multimodal sub-spine); **Phase 2 (the CLI) is unblocked**. See [docs/roadmap/current.md](docs/roadmap/current.md) for live status. diff --git a/docs/reference/contracts/agent-session-spec.md b/docs/reference/contracts/agent-session-spec.md index 18e18e7d..adafbae5 100644 --- a/docs/reference/contracts/agent-session-spec.md +++ b/docs/reference/contracts/agent-session-spec.md @@ -182,6 +182,40 @@ The export **produces** the format owned by [workflow-yaml-spec.md](workflow-yam **mapping** (session turn → `agent` node, transcript → metadata) is the contract owned here. The desktop "Export to Canvas" affordance and the CLI `relavium chat-export` both drive this one contract. +**Precise mapping (1.Z).** Given a loaded `AgentSessionRecord` + its ordered `SessionMessage[]`, the +exporter builds a `WorkflowDefinition` deterministically (no wall-clock / randomness, so the artifact is +reproducible and round-trips): + +- **Nodes** — a single `input` node (`id: input`), then **one `agent` node per COMPLETED logical turn** in + `sequenceNumber` order (`id: turn-1`, `turn-2`, … — 1-based), then one `output` node (`id: output`). A + *logical turn* is the contiguous `user` message(s) plus the assistant/tool messages answering them (a host + may persist a single turn as split rows — `user → assistant(tool_call) → tool → assistant(text)`); it is one + node, not one per assistant message. A turn is *completed* only if it produced final assistant **text** — an + unanswered or interrupted-mid-tool-loop turn (no final text) is **omitted from the chain** (kept verbatim in + `metadata`), so export and `reconstructSessionState`'s rollback (1.Y) agree on what a turn is. Each `agent` + node carries: `agent_ref` = the session's `agentSlug`; `prompt_template` = the **text** of the turn's + `user` message(s), with interpolation openers neutralized (omitted if empty); `tools` = the deduped union of + tool names invoked across the turn's assistant messages (the `tool_call` parts), omitted when none. No + `model`/`temperature`/`max_tokens`/`retry`/`output_schema` are emitted — those are authoring concerns the + user adds on the canvas, not replay fields. +- **Edges** — a straight linear chain `input → turn-1 → … → turn-n → output` (just `{ from, to }`); when a + session has no completed turn the chain is `input → output`. No parallel/conditional/loop edges (ADR-0026). +- **Workflow `id`** — a deterministic kebab slug of the title (ASCII alphanumerics only, matching + `kebabIdSchema`; non-ASCII is stripped), falling back to `exported-session`. The scaffold's id is + human-reviewed and renameable on the canvas. +- **`agents`** — the session's frozen `agentSnapshot` (an inline `Agent`) is emitted as the sole `agents[]` + entry so `agent_ref` resolves; when no snapshot was captured, `agents` is omitted and `agent_ref` resolves + against the workspace agent registry at author time (the file still parses — `agent_ref` resolution is the + engine's job, not the schema's). +- **`metadata`** — the full transcript under a single reserved key: `metadata.relaviumExport = { source: + 'session', sessionId, agentSlug, title?, createdAt, updatedAt, messages: SessionMessage[] }`. It is a real + schema field (`z.record`), so it survives parse → serialize round-trips. +- **Determinism + exclusions** — the YAML emitter (1.Z, `serializeWorkflow`; 1.L is parse-only) sorts map + keys alphabetically and preserves array order, so `parse → serialize` is byte-stable. No `secret` value can + appear (secrets never enter a message — [ADR-0029](../../decisions/0029-tool-policy-hardening.md)) and no + reasoning `signature` can appear (the transcript is `DurableContentPart`, which structurally omits it — + [ADR-0030](../../decisions/0030-llm-seam-shape-amendment-reasoning-response-format-provider-executed.md)). + ## Validation and persistence - Validated against `AgentSessionSchema` / `SessionMessageSchema` / `SessionContextSchema` (Zod, in diff --git a/docs/roadmap/current.md b/docs/roadmap/current.md index 4724bb87..51ca8a75 100644 --- a/docs/roadmap/current.md +++ b/docs/roadmap/current.md @@ -167,8 +167,10 @@ the H3 approve-continues bypass, the per-attempt `FallbackChain` enforcement) emission via an injected `SessionEventSink`) **then landed together — ✅ Done (PR #26, 2026-06-16)**. 1.AC closed **1.m4** (the full engine stack), so the critical path now reaches **1.U — the end-to-end Node harness (the M2 milestone)**, now unblocked; in parallel, **Lane C** (the 1.m5 sub-spine) continued from 1.V through **1.W** -(the `SessionEventSink`→`RunEventBus` wiring + per-session `sequenceNumber`/`SessionHandle` — ✅ Done, PR #28) to -**1.X** (session persistence, next), with cost-event persistence still a tracked deferral. +(the `SessionEventSink`→`RunEventBus` wiring + per-session `sequenceNumber`/`SessionHandle` — ✅ Done, PR #28) +and **1.X** (session persistence — the `agent_sessions`/`session_messages` tables + `SessionMessageSchema` + +`SessionStore`, data-layer only — ✅ Done, PR #29), leaving **1.Y/1.Z/1.AA** next on Lane C, with cost-event +persistence still a tracked deferral. > **Multimodal I/O — the shape is landed (1.AD ✅ Done, PR #11, 2026-06-10).** First-class > image/audio/video I/O (input **and** output, incl. generate-media-by-rule) was decided on 2026-06-08: @@ -204,8 +206,11 @@ milestone)**, now unblocked; in parallel, **Lane C** (the 1.m5 sub-spine) contin > end-to-end). **The Phase-1 engine critical path is complete.** On the additive Lane C, the **`session:*` > namespace (1.W) is ✅ Done (PR #28, 2026-06-17)** — the `SessionEventSink`→`RunEventBus` adapter (per-session > `sequenceNumber`, `next`/`emit` overloads keeping the engine run-precise), the `SessionHandle`, and the -> combined `RunOrSessionEventSchema` gate. The remaining Phase-1 work is **additive and off the critical -> path**: Lane C (1.X persistence ‖ 1.Y/1.Z/1.AA) and the 1.m6 multimodal sub-spine (1.AE–1.AH). **Phase 2 +> combined `RunOrSessionEventSchema` gate — and **session persistence (1.X) is ✅ Done (PR #29, 2026-06-17)** — +> the `agent_sessions`/`session_messages` tables + migration, `SessionMessageSchema`/`AgentSessionSchema`, and +> the `SessionStore` + domain↔row mappers (data-layer only; engine wiring is 1.Y/1.AA). The remaining Phase-1 +> work is **additive and off the critical path**: Lane C (1.Y/1.Z/1.AA) and the 1.m6 multimodal sub-spine +> (1.AE–1.AH). **Phase 2 > (CLI, milestone M3) is unblocked.** Carry-over hardening is tracked in [deferred-tasks.md](deferred-tasks.md) — pick items up as Phase 1 diff --git a/docs/roadmap/phases/phase-1-engine-and-llm.md b/docs/roadmap/phases/phase-1-engine-and-llm.md index acd72a2a..68b16734 100644 --- a/docs/roadmap/phases/phase-1-engine-and-llm.md +++ b/docs/roadmap/phases/phase-1-engine-and-llm.md @@ -875,7 +875,8 @@ These build the `AgentSession` entry point ([ADR-0024](../../decisions/0024-agen - **1.W — `session:*` event namespace — ✅ Done (PR #28, 2026-06-17).** Emit session lifecycle events on the shared `RunEventBus` with the same `sequenceNumber` gap/resync logic ([sse-event-schema.md](../../reference/contracts/sse-event-schema.md)). *Acceptance:* session events are disjoint from `run:*` and gap-detected identically. - *Also owns (from the 1.V injected-sink split, recorded 2026-06-15):* the `SessionEventSink → RunEventBus` adapter; a **`SessionHandle`** (the async-iterable session stream + `cancel`, mirroring `createRunHandle`); and **reconciling the bus session-event gate** — `RunEventBus` validated against `RunEventSchema` and its `RunEventDraft` type both **excluded** the five `session:*` variants (they live only in the separate `SessionEventSchema`), so the bus needed a combined Run+Session gate before it could carry the session lifecycle events 1.V emits. - *Landed mechanism (PR #28):* the combined **`RunOrSessionEventSchema`** gate (`@relavium/shared`); `next`/`emit` **overloads** (run draft → `RunEvent`, session draft → `SessionEvent`, either → `RunOrSessionEvent`) keep the engine **run-precise** rather than widening `RunEventDraft`; `createSessionEventSink` attaches the `sessionId` — defensively dropping the run-only `agent:file_patch_proposed` at the seam — and the bus stamps the per-session `sequenceNumber`; `createSessionHandle` mirrors `createRunHandle` (terminal **only** on `session:cancelled`) over the shared `BoundedEventStream` (extracted from `RunHandle`, with an `onClose` hook that unsubscribes on early consumer abandon); `correlationKey` enforces the runId-XOR-sessionId invariant, and the bus/stream "can never happen" asserts are the typed `RunLoopInvariantError`. See [sse-event-schema.md](../../reference/contracts/sse-event-schema.md) §"The session stream". -- **1.X — Session persistence.** `agent_sessions` + `session_messages` via `@relavium/db` into `history.db` ([database-schema.md](../../reference/desktop/database-schema.md)). **Authors the durable `SessionMessageSchema`** in `@relavium/shared` (deferred from 1.V, which runs on the in-flight `LlmMessage` form): `{ id, sessionId, sequenceNumber, role, content: DurableContentPart[], modelId?, timestamp }` — the persisted transcript type these tables store. *Acceptance:* a session round-trips to the DB and resumes. **Note:** adding these two tables requires a regenerated Drizzle migration snapshot (the schema-migration drift CI gate). **ADR-0030 ephemerality:** a `reasoning` part's `signature`/`redacted` continuity token must **not** be persisted to `session_messages` — strip it (keep reasoning *text* if a transcript needs it, drop the opaque signature). *Acceptance also asserts:* a round-tripped session row carries no reasoning `signature`. +- **1.X — Session persistence — ✅ Done (PR #29, 2026-06-17).** `agent_sessions` + `session_messages` via `@relavium/db` into `history.db` ([database-schema.md](../../reference/desktop/database-schema.md)). **Authors the durable `SessionMessageSchema`** in `@relavium/shared` (deferred from 1.V, which runs on the in-flight `LlmMessage` form): `{ id, sessionId, sequenceNumber, role, content: DurableContentPart[], modelId?, timestamp }` — the persisted transcript type these tables store. *Acceptance:* a session round-trips to the DB and resumes. **Note:** adding these two tables requires a regenerated Drizzle migration snapshot (the schema-migration drift CI gate). **ADR-0030 ephemerality:** a `reasoning` part's `signature`/`redacted` continuity token must **not** be persisted to `session_messages` — strip it (keep reasoning *text* if a transcript needs it, drop the opaque signature). *Acceptance also asserts:* a round-tripped session row carries no reasoning `signature`. + - *Landed mechanism (PR #29):* **data-layer only** (no engine change — `packages/core` never imports `@relavium/db`; the per-turn `AgentSession`→store wiring + cross-restart resume are 1.Y/1.AA). `@relavium/shared` gained `session.ts` (`SessionMessageSchema` / `AgentSessionSchema` (inferred `AgentSessionRecord`, to avoid clashing with the `AgentSession` engine class) / `SessionStatusSchema`), reusing the shared `DurableContentPart` so a reasoning `signature` + inline media are **structurally impossible** to persist; `modelId` is a `model_catalog` id reference (host-resolved). `@relavium/db` added the two tables (migration `0001_pale_scorpion.sql` + snapshot + journal; the drift gate re-generates clean; a 0001 fidelity snapshot test pins the DDL), the `SessionStore`, and the pure domain↔row mappers — the **single validation boundary** (parse on write AND read); `content_parts` is the canonical body (the scalar columns are optional denormalized metadata), and `updateSession` freezes `created_at`. Verified by a multi-dimensional adversarial review (8 dimensions); all confirmed findings folded in. - **1.Y — Session checkpoint/resume.** Reuse the idempotency-key logic so a session resumes after a restart. - **1.Z — Export-to-workflow serializer.** Session → `.relavium.yaml` **linear-chain scaffold + transcript** ([ADR-0026](../../decisions/0026-session-export-to-workflow.md)). Includes a **`WorkflowDefinition` → YAML emitter** (deterministic key ordering, the `metadata` transcript block, secret exclusion) — 1.L is parse-only, so this workstream owns serialization. *Acceptance:* an exported session parses as a valid workflow whose agent nodes mirror the turns; **parse → serialize round-trips** (including `metadata`); no `secret` value is serialized; and **no reasoning `signature` is serialized** (ADR-0030 ephemerality — the signature is a transient same-provider token, never written to a committable artifact, same exclusion as `secret`). - **1.AA — Node-harness chat regression.** The session counterpart of 1.U: a multi-turn chat with a tool call and an export, run green in CI. @@ -1146,7 +1147,7 @@ flowchart LR | 1.U | B | 1.P, 1.S, 1.Q, 1.R, 1.T, 1.AC | **M2** | ✅ — **Done (PR #27) — M2 reached** | | 1.V | C | 1.O | 1.W, 1.X, 1.Z | ◇ — **Done (PR #26)** | | 1.W | C | 1.V, 1.N, 1.L.0 | 1.AA | ✅ — **Done (PR #28, 2026-06-17)** | -| 1.X | C | 1.V, `@relavium/db` (new migration) | 1.Y, 1.AA | ◇ | +| 1.X | C | 1.V, `@relavium/db` (new migration) | 1.Y, 1.AA | ✅ — **Done (PR #29, 2026-06-17)** | | 1.Y | C | 1.X, 1.R | 1.AA | ◇ | | 1.Z | C | 1.V, 1.L | 1.AA | ◇ | | 1.AA | C | 1.V, 1.W, 1.X, 1.Y, 1.Z | **1.m5** | ◇ | diff --git a/packages/core/src/engine/agent-session.ts b/packages/core/src/engine/agent-session.ts index e056b89d..e3426333 100644 --- a/packages/core/src/engine/agent-session.ts +++ b/packages/core/src/engine/agent-session.ts @@ -54,6 +54,7 @@ import { import { BudgetPauseError } from './budget-governor.js'; import type { AbortControllerLike } from './execution-host.js'; import type { NodeStreamEvent } from './node-executor.js'; +import type { SessionResumeState } from './session-resume.js'; /** The default hard turn cap when {@link SessionDeps.maxTurns} is omitted — a finite DoS fail-safe. */ export const DEFAULT_SESSION_MAX_TURNS = 50; @@ -192,6 +193,27 @@ export class AgentSession { this.#limits = params.deps.limits ?? DEFAULT_AGENT_TURN_LIMITS; } + /** + * Resume a session in a NEW process (1.Y) from its reconstructed state + * ({@link reconstructSessionState}, built from the persisted transcript). Preloads the in-flight + * transcript, the hard-cap turn count, and the running cost, then lands directly at `idle` **without** + * emitting `session:started` — the session already started in the prior process; re-emitting it would + * double a terminal-less lifecycle event. The next {@link sendMessage} continues the conversation. This + * **replaces** {@link start}: the returned session is past `created`, so calling `start()` on it throws. + */ + static resume(params: AgentSessionParams, state: SessionResumeState): AgentSession { + const session = new AgentSession(params); + session.#messages.push(...state.messages); + session.#turnCount = state.turnCount; + session.#cumulativeCostMicrocents = state.cumulativeCostMicrocents; + // Sync a host-wired budget governor with the carried-over spend so the FIRST resumed turn's pre-egress + // check sees the real cumulative — not 0 — before any cost:updated fires (mirrors #onTurnEmit). Without + // this, a resumed session's first turn could bypass a near-exhausted budget cap. + session.#deps.updateCost?.(state.cumulativeCostMicrocents); + session.#status = 'idle'; + return session; + } + /** Open the session: emit `session:started` and move to idle. Idempotent-guarded (one start per session). */ start(): void { if (this.#status !== 'created') { diff --git a/packages/core/src/engine/session-resume.test.ts b/packages/core/src/engine/session-resume.test.ts new file mode 100644 index 00000000..1bc10456 --- /dev/null +++ b/packages/core/src/engine/session-resume.test.ts @@ -0,0 +1,287 @@ +import type { CapabilityFlags, LlmMessage, LlmProvider, StreamChunk } from '@relavium/llm'; +import { + AgentSchema, + SessionContextSchema, + type AgentSessionRecord, + type SessionMessage, +} from '@relavium/shared'; +import { describe, expect, it } from 'vitest'; + +import { AgentSession, type SessionDeps, type SessionStreamEvent } from './agent-session.js'; +import { createAbortController } from './execution-host.js'; +import { reconstructSessionState, type SessionResumeState } from './session-resume.js'; +import type { ToolRegistry } from '../tools/types.js'; + +const TS = '2026-06-17T08:00:00.000Z'; +const CTX = SessionContextSchema.parse({ workingDir: '/workspace/s', fsScopeTier: 'sandboxed' }); +const AGENT = AgentSchema.parse({ + id: 'chatter', + model: 'claude-opus-4-8', + provider: 'anthropic', + system_prompt: 'You are concise.', +}); + +const record = (overrides: Partial = {}): AgentSessionRecord => ({ + id: 'sess-1', + agentSlug: 'chatter', + context: CTX, + status: 'idle', + totalInputTokens: 0, + totalOutputTokens: 0, + totalCostMicrocents: 0, + createdAt: TS, + updatedAt: TS, + ...overrides, +}); + +const msg = ( + sequenceNumber: number, + role: SessionMessage['role'], + content: SessionMessage['content'], +): SessionMessage => ({ + id: `m-${sequenceNumber}`, + sessionId: 'sess-1', + sequenceNumber, + role, + content, + timestamp: TS, +}); + +describe('reconstructSessionState (1.Y)', () => { + it('projects user/assistant text turns and re-seeds turnCount + cost', () => { + const state = reconstructSessionState(record({ totalCostMicrocents: 4200 }), [ + msg(0, 'user', [{ type: 'text', text: 'hi' }]), + msg(1, 'assistant', [{ type: 'text', text: 'hello' }]), + msg(2, 'user', [{ type: 'text', text: 'more' }]), + msg(3, 'assistant', [{ type: 'text', text: 'sure' }]), + ]); + expect(state.messages).toEqual([ + { role: 'user', content: [{ type: 'text', text: 'hi' }] }, + { role: 'assistant', content: [{ type: 'text', text: 'hello' }] }, + { role: 'user', content: [{ type: 'text', text: 'more' }] }, + { role: 'assistant', content: [{ type: 'text', text: 'sure' }] }, + ]); + expect(state.turnCount).toBe(2); + expect(state.cumulativeCostMicrocents).toBe(4200); + }); + + it('rolls back a trailing unanswered user turn (the incomplete-turn idempotency)', () => { + const state = reconstructSessionState(record(), [ + msg(0, 'user', [{ type: 'text', text: 'hi' }]), + msg(1, 'assistant', [{ type: 'text', text: 'hello' }]), + msg(2, 'user', [{ type: 'text', text: 'interrupted — no reply persisted' }]), + ]); + expect(state.messages).toEqual([ + { role: 'user', content: [{ type: 'text', text: 'hi' }] }, + { role: 'assistant', content: [{ type: 'text', text: 'hello' }] }, + ]); + expect(state.turnCount).toBe(1); + }); + + it('drops system/tool-role messages and non-text parts (text-only in-flight transcript)', () => { + const state = reconstructSessionState(record(), [ + msg(0, 'system', [{ type: 'text', text: 'system prompt' }]), + msg(1, 'user', [{ type: 'text', text: 'q' }]), + msg(2, 'assistant', [ + { type: 'reasoning', text: 'thinking' }, + { type: 'text', text: 'a' }, + ]), + ]); + expect(state.messages).toEqual([ + { role: 'user', content: [{ type: 'text', text: 'q' }] }, + { role: 'assistant', content: [{ type: 'text', text: 'a' }] }, + ]); + }); + + it('sorts by sequenceNumber before reconstructing', () => { + const state = reconstructSessionState(record(), [ + msg(1, 'assistant', [{ type: 'text', text: 'hello' }]), + msg(0, 'user', [{ type: 'text', text: 'hi' }]), + ]); + expect(state.messages.map((m) => m.role)).toEqual(['user', 'assistant']); + }); + + it('rolls back an interrupted tool-loop turn ending in a tool result — no dangling user', () => { + const state = reconstructSessionState(record(), [ + msg(0, 'user', [{ type: 'text', text: 'q1' }]), + msg(1, 'assistant', [{ type: 'text', text: 'a1' }]), // a completed exchange + msg(2, 'user', [{ type: 'text', text: 'q2 — use a tool' }]), // the interrupted turn begins + msg(3, 'assistant', [ + { type: 'tool_call', id: 'c1', name: 'read_file', args: { path: 'x' } }, + ]), + msg(4, 'tool', [{ type: 'tool_result', toolCallId: 'c1', result: 'ok', isError: false }]), // died here + ]); + // the entire interrupted turn (user + tool_call + tool) is rolled back — the projection drops the + // tool/text-less-assistant rows and the trailing-user rollback removes the originating q2. + expect(state.messages).toEqual([ + { role: 'user', content: [{ type: 'text', text: 'q1' }] }, + { role: 'assistant', content: [{ type: 'text', text: 'a1' }] }, + ]); + expect(state.turnCount).toBe(1); + }); + + it('rolls back a turn whose assistant produced only a tool_call (no committed text)', () => { + const state = reconstructSessionState(record(), [ + msg(0, 'user', [{ type: 'text', text: 'q' }]), + msg(1, 'assistant', [{ type: 'tool_call', id: 'c1', name: 'read_file', args: {} }]), + ]); + expect(state.messages).toEqual([]); // no completed exchange survives + expect(state.turnCount).toBe(0); + }); + + it('counts a completed tool-loop turn once (assistant tool_call → tool → assistant text)', () => { + const state = reconstructSessionState(record(), [ + msg(0, 'user', [{ type: 'text', text: 'q' }]), + msg(1, 'assistant', [{ type: 'tool_call', id: 'c1', name: 'read_file', args: {} }]), // within-turn + msg(2, 'tool', [{ type: 'tool_result', toolCallId: 'c1', result: 'ok', isError: false }]), + msg(3, 'assistant', [{ type: 'text', text: 'final answer' }]), // the completing text + ]); + expect(state.messages).toEqual([ + { role: 'user', content: [{ type: 'text', text: 'q' }] }, + { role: 'assistant', content: [{ type: 'text', text: 'final answer' }] }, + ]); + expect(state.turnCount).toBe(1); // one logical turn — the tool_call-only assistant row is not counted + }); +}); + +// --- AgentSession.resume integration (stub provider, mirroring agent-session.test.ts) ---------------- + +const CAPS: CapabilityFlags = { + tools: true, + streaming: true, + parallelToolCalls: true, + vision: false, + promptCache: false, + reasoning: true, + media: { + input: { image: false, audio: false, video: false, document: false }, + outputCombinations: [], + }, +}; + +async function* streamOf(chunks: readonly StreamChunk[]): AsyncGenerator { + await Promise.resolve(); + for (const chunk of chunks) yield chunk; +} + +const textTurn = (text: string): StreamChunk[] => [ + { type: 'text_delta', text }, + { type: 'stop', stopReason: 'stop', usage: { inputTokens: 5, outputTokens: 3 } }, +]; + +const noToolRegistry: ToolRegistry = { + has: () => false, + list: () => [], + dispatch: () => Promise.reject(new Error('no tool dispatch expected')), +}; + +/** A provider that records the `messages` of each request and replies with a fixed text turn. */ +function capturingProvider(seen: LlmMessage[][]): LlmProvider { + return { + id: 'anthropic', + supports: CAPS, + generate: () => { + throw new Error('unused'); + }, + stream: (req) => { + seen.push(req.messages.map((m) => ({ role: m.role, content: [...m.content] }))); + return streamOf(textTurn('resumed reply')); + }, + }; +} + +function depsFor( + provider: LlmProvider, + events: SessionStreamEvent[], + maxTurns?: number, +): SessionDeps { + return { + resolveProvider: () => provider, + registry: noToolRegistry, + tools: [], + keyFor: () => 'key', + sleep: () => Promise.resolve(), + newAbortController: createAbortController, + emit: (event) => { + events.push(event); + }, + ...(maxTurns === undefined ? {} : { maxTurns }), + }; +} + +const params = (deps: SessionDeps) => ({ + sessionId: 'sess-1', + agentRef: AGENT.id, + agent: AGENT, + context: CTX, + deps, +}); + +describe('AgentSession.resume (1.Y)', () => { + it('resumes without re-emitting session:started, and the next turn sees the prior transcript', async () => { + const seen: LlmMessage[][] = []; + const events: SessionStreamEvent[] = []; + const state = reconstructSessionState(record(), [ + msg(0, 'user', [{ type: 'text', text: 'hi' }]), + msg(1, 'assistant', [{ type: 'text', text: 'hello' }]), + ]); + const session = AgentSession.resume(params(depsFor(capturingProvider(seen), events)), state); + + // resume does NOT re-emit session:started (the session already started in the prior process). + expect(events.map((e) => e.type)).not.toContain('session:started'); + + await session.sendMessage('again'); + + // the provider's call saw the resumed transcript followed by the new user message. + expect(seen[0]).toEqual([ + { role: 'user', content: [{ type: 'text', text: 'hi' }] }, + { role: 'assistant', content: [{ type: 'text', text: 'hello' }] }, + { role: 'user', content: [{ type: 'text', text: 'again' }] }, + ]); + const types = events.map((e) => e.type); + expect(types).toContain('session:turn_started'); + expect(types).toContain('session:turn_completed'); + }); + + it('honors the hard turn cap across a restart (the resumed turnCount counts)', async () => { + const seen: LlmMessage[][] = []; + const events: SessionStreamEvent[] = []; + // maxTurns 1, resumed already at 1 completed turn → the next turn is over the cap. + const state: SessionResumeState = { + messages: [ + { role: 'user', content: [{ type: 'text', text: 'hi' }] }, + { role: 'assistant', content: [{ type: 'text', text: 'hello' }] }, + ], + turnCount: 1, + cumulativeCostMicrocents: 0, + }; + const session = AgentSession.resume(params(depsFor(capturingProvider(seen), events, 1)), state); + + await session.sendMessage('again'); + + expect(seen).toHaveLength(0); // blocked loud BEFORE any egress — the provider was never called + const completed = events.filter((e) => e.type === 'session:turn_completed'); + expect(completed).toHaveLength(1); + const only = completed[0]; + expect(only?.type === 'session:turn_completed' && only.error?.code).toBe('turn_limit'); + }); + + it('syncs a host-wired budget governor with the carried-over cost on resume', () => { + const events: SessionStreamEvent[] = []; + const costs: number[] = []; + const deps: SessionDeps = { + ...depsFor(capturingProvider([]), events), + updateCost: (cost) => { + costs.push(cost); + }, + }; + AgentSession.resume(params(deps), { + messages: [{ role: 'user', content: [{ type: 'text', text: 'hi' }] }], + turnCount: 1, + cumulativeCostMicrocents: 4200, + }); + // the governor is seeded once, at resume, with the absolute cumulative — so the first resumed turn's + // pre-egress check sees the real spend, not 0 (before any cost:updated fires). + expect(costs).toEqual([4200]); + }); +}); diff --git a/packages/core/src/engine/session-resume.ts b/packages/core/src/engine/session-resume.ts new file mode 100644 index 00000000..ad9019b0 --- /dev/null +++ b/packages/core/src/engine/session-resume.ts @@ -0,0 +1,106 @@ +/** + * Session resume (1.Y) — reconstruct an `AgentSession`'s in-flight state from its persisted transcript so a + * conversation continues after a process restart. Sessions are **directly stored** (ADR-0003 governs *runs*, + * not sessions), so resume RELOADS rows ({@link SessionStore.loadFull}, 1.X) rather than replaying an event + * log, and it reuses the run-side idempotency **principle** (1.R): an interrupted, never-completed turn is + * rolled back — the `sessionId+sequenceNumber` analog of "re-run the incomplete node" — so resume yields only + * **completed exchanges** and the next turn re-prompts rather than replaying a half turn. + * + * Platform-free: it operates only on `@relavium/shared` types (the host loads via `@relavium/db`; the engine + * never imports it). The projection mirrors `AgentSession`'s cross-turn invariant — the in-flight transcript + * is **text-only** (the turn core keeps within-turn `tool_use`/`tool_result` internal, and a reasoning + * `signature` must not span turns, ADR-0030/0039) — so a resumed next turn stays protocol-valid (no orphaned + * `tool_use`). Full-fidelity history lives durably (1.X) and in an export (1.Z); this is what the model sees next. + */ + +import type { LlmMessage } from '@relavium/llm'; +import type { AgentSessionRecord, DurableContentPart, SessionMessage } from '@relavium/shared'; + +/** + * The reconstructed in-memory state {@link AgentSession.resume} preloads — its `#messages` (in-flight + * transcript), `#turnCount` (the hard-cap counter), and `#cumulativeCostMicrocents` (the running cost). + * + * Build it via {@link reconstructSessionState}: `messages` must be the **text-only** `user`/`assistant` + * projection (AgentSession's cross-turn invariant). `resume` preloads these verbatim, so a hand-built state + * carrying `tool_call`/`tool_result`/`reasoning` parts would be replayed to the provider on the next turn — + * risking an orphaned `tool_use` or a non-alternating request. Do not assemble one by hand. + */ +export interface SessionResumeState { + readonly messages: readonly LlmMessage[]; + readonly turnCount: number; + readonly cumulativeCostMicrocents: number; +} + +/** The concatenated `text` parts of a durable content array (non-text parts are dropped). */ +function textOf(content: readonly DurableContentPart[]): string { + return content + .filter((part): part is Extract => part.type === 'text') + .map((part) => part.text) + .join('\n\n'); +} + +/** + * Project the persisted transcript into the in-flight cross-turn form `AgentSession` continues from: + * **text-only** `user`/`assistant` messages (mirroring how `AgentSession` builds `#messages` — `system` is + * the agent prompt, not a turn; `tool` results stay within a turn). An empty-text message is dropped (the + * same `length > 0` guard the assistant-append uses). + */ +function durableToLlmMessages(messages: readonly SessionMessage[]): LlmMessage[] { + const out: LlmMessage[] = []; + for (const message of messages) { + if (message.role !== 'user' && message.role !== 'assistant') { + continue; + } + const text = textOf(message.content); + if (text.length === 0) { + continue; + } + out.push({ role: message.role, content: [{ type: 'text', text }] }); + } + return out; +} + +/** + * Drop trailing turns that did not complete — the incomplete-turn rollback, applied to the ALREADY-PROJECTED + * transcript. After projection the only roles are `user` and text-bearing `assistant`, so a completed exchange + * ends in `assistant`; any trailing `user` is an unanswered turn — the process died mid-turn, whether **before + * the assistant replied** or **mid-tool-loop** (which projects away its `tool` / text-less `assistant` + * tool_call rows and re-exposes the originating `user`). Re-prompting it on resume is the + * `sessionId+sequenceNumber` idempotency analog of re-running the run-side incomplete node. + */ +function trimTrailingUserTurn(messages: readonly LlmMessage[]): LlmMessage[] { + const committed = [...messages]; + while (committed.at(-1)?.role === 'user') { + committed.pop(); + } + return committed; +} + +/** + * Reconstruct the {@link SessionResumeState} from a loaded session record + its transcript (any order). Sorts + * by `sequenceNumber`, **projects first** (to the text-only in-flight transcript), then rolls back a trailing + * unanswered turn, and re-seeds the turn count + the running cost (the record's total). Pure and + * deterministic — the host passes the result to {@link AgentSession.resume}. + * + * Projecting BEFORE trimming is load-bearing: an interrupted mid-tool-loop turn leaves a `tool` / text-less + * `assistant` tail in the durable record; the projection drops those, so the trailing-`user` rollback then + * sees and removes the originating unanswered `user` — otherwise it would survive as a dangling turn and the + * next `sendMessage` would emit two consecutive `user` messages (a non-alternating, provider-rejected request). + * + * NOTE: `turnCount` counts the **text-producing** assistant turns that survive projection — one per completed + * logical exchange (a within-turn tool_call-only assistant row is not double-counted). A turn that engaged a + * provider but produced no committed text leaves no exchange, so the resumed hard-cap counter is a lower bound + * (the cap is a safety limit, not exact accounting; AgentSessionRecord carries no turn counter to make it exact). + */ +export function reconstructSessionState( + record: AgentSessionRecord, + messages: readonly SessionMessage[], +): SessionResumeState { + const ordered = [...messages].sort((a, b) => a.sequenceNumber - b.sequenceNumber); + const committed = trimTrailingUserTurn(durableToLlmMessages(ordered)); + return { + messages: committed, + turnCount: committed.filter((message) => message.role === 'assistant').length, + cumulativeCostMicrocents: record.totalCostMicrocents, + }; +} diff --git a/packages/core/src/export/serializer.test.ts b/packages/core/src/export/serializer.test.ts new file mode 100644 index 00000000..04f503d3 --- /dev/null +++ b/packages/core/src/export/serializer.test.ts @@ -0,0 +1,226 @@ +import { + AgentSchema, + SessionContextSchema, + type AgentSessionRecord, + type SessionMessage, +} from '@relavium/shared'; +import { describe, expect, it } from 'vitest'; + +import { parseWorkflow } from '../parser.js'; +import { serializeWorkflow, sessionToWorkflow } from './serializer.js'; + +const TS = '2026-06-17T08:00:00.000Z'; +const CTX = SessionContextSchema.parse({ workingDir: '/workspace/s', fsScopeTier: 'sandboxed' }); +const AGENT = AgentSchema.parse({ + id: 'chatter', + model: 'claude-opus-4-8', + provider: 'anthropic', + system_prompt: 'You are concise.', +}); + +const session = (overrides: Partial = {}): AgentSessionRecord => ({ + id: 'sess-1', + agentSlug: 'chatter', + agentSnapshot: AGENT, + title: 'My Chat', + context: CTX, + status: 'ended', + totalInputTokens: 0, + totalOutputTokens: 0, + totalCostMicrocents: 0, + createdAt: TS, + updatedAt: TS, + ...overrides, +}); + +const msg = ( + sequenceNumber: number, + role: SessionMessage['role'], + content: SessionMessage['content'], +): SessionMessage => ({ + id: `m-${sequenceNumber}`, + sessionId: 'sess-1', + sequenceNumber, + role, + content, + timestamp: TS, +}); + +describe('sessionToWorkflow (1.Z) — linear-chain scaffold', () => { + it('maps assistant turns to a linear agent-node chain (input → turn-n → output) with edges', () => { + const def = sessionToWorkflow(session(), [ + msg(0, 'user', [{ type: 'text', text: 'hello' }]), + msg(1, 'assistant', [{ type: 'text', text: 'hi there' }]), + msg(2, 'user', [{ type: 'text', text: 'use a tool' }]), + msg(3, 'assistant', [ + { type: 'tool_call', id: 'c1', name: 'read_file', args: { path: 'x' } }, + { type: 'text', text: 'done' }, + ]), + ]); + const { nodes, edges } = def.workflow; + expect(nodes.map((n) => n.id)).toEqual(['input', 'turn-1', 'turn-2', 'output']); + expect(nodes.map((n) => n.type)).toEqual(['input', 'agent', 'agent', 'output']); + + const turn1 = nodes[1]; + expect(turn1?.type === 'agent' && turn1.agent_ref).toBe('chatter'); + expect(turn1?.type === 'agent' && turn1.prompt_template).toBe('hello'); + expect(turn1?.type === 'agent' && turn1.tools).toBeUndefined(); // text-only turn → no node-level tools + + const turn2 = nodes[2]; + expect(turn2?.type === 'agent' && turn2.prompt_template).toBe('use a tool'); + expect(turn2?.type === 'agent' && turn2.tools).toEqual(['read_file']); + + expect(edges).toEqual([ + { from: 'input', to: 'turn-1' }, + { from: 'turn-1', to: 'turn-2' }, + { from: 'turn-2', to: 'output' }, + ]); + expect(def.workflow.agents).toEqual([AGENT]); + expect(def.workflow.name).toBe('My Chat'); + expect(def.workflow.id).toBe('my-chat'); // kebab slug of the title + }); + + it('preserves the full transcript under metadata.relaviumExport', () => { + const messages = [ + msg(0, 'user', [{ type: 'text', text: 'hi' }]), + msg(1, 'assistant', [{ type: 'text', text: 'yo' }]), + ]; + const def = sessionToWorkflow(session(), messages); + const serialized = JSON.stringify(def.workflow.metadata); + expect(serialized).toContain('relaviumExport'); + expect(serialized).toContain('"sessionId":"sess-1"'); + expect(serialized).toContain('"agentSlug":"chatter"'); + // the entire transcript (both messages) is carried, not just the assistant turns + expect(serialized).toContain('"sequenceNumber":0'); + expect(serialized).toContain('"sequenceNumber":1'); + }); + + it('handles a session with no assistant turns (input → output)', () => { + const def = sessionToWorkflow(session(), [msg(0, 'user', [{ type: 'text', text: 'hi' }])]); + expect(def.workflow.nodes.map((n) => n.id)).toEqual(['input', 'output']); + expect(def.workflow.edges).toEqual([{ from: 'input', to: 'output' }]); + expect(() => parseWorkflow(serializeWorkflow(def))).not.toThrow(); + }); + + it('omits agents when no snapshot was captured (the scaffold still parses)', () => { + const def = sessionToWorkflow(session({ agentSnapshot: undefined }), [ + msg(0, 'user', [{ type: 'text', text: 'hi' }]), + msg(1, 'assistant', [{ type: 'text', text: 'yo' }]), + ]); + expect(def.workflow.agents).toBeUndefined(); + expect(() => parseWorkflow(serializeWorkflow(def))).not.toThrow(); + }); + + it('collapses a split-row tool-loop turn into ONE node (prompt + tools), not one per assistant message', () => { + const def = sessionToWorkflow(session(), [ + msg(0, 'user', [{ type: 'text', text: 'read the file' }]), + msg(1, 'assistant', [ + { type: 'tool_call', id: 'c1', name: 'read_file', args: { path: 'x' } }, + ]), + msg(2, 'tool', [ + { type: 'tool_result', toolCallId: 'c1', result: 'contents', isError: false }, + ]), + msg(3, 'assistant', [{ type: 'text', text: 'here is what it says' }]), + ]); + expect(def.workflow.nodes.map((n) => n.id)).toEqual(['input', 'turn-1', 'output']); // ONE turn node + const turn1 = def.workflow.nodes[1]; + expect(turn1?.type === 'agent' && turn1.prompt_template).toBe('read the file'); + expect(turn1?.type === 'agent' && turn1.tools).toEqual(['read_file']); + }); + + it('dedupes a tool invoked multiple times within one turn', () => { + const def = sessionToWorkflow(session(), [ + msg(0, 'user', [{ type: 'text', text: 'read two files' }]), + msg(1, 'assistant', [ + { type: 'tool_call', id: 'c1', name: 'read_file', args: { path: 'a' } }, + { type: 'tool_call', id: 'c2', name: 'read_file', args: { path: 'b' } }, + ]), + msg(2, 'assistant', [{ type: 'text', text: 'done' }]), + ]); + const turn1 = def.workflow.nodes[1]; + expect(turn1?.type === 'agent' && turn1.tools).toEqual(['read_file']); // not ['read_file', 'read_file'] + }); + + it('derives a clean kebab workflow id (no leading/trailing/double dashes; fallback when empty)', () => { + expect(sessionToWorkflow(session({ title: ' Hello -- World! ' }), []).workflow.id).toBe( + 'hello-world', + ); + expect(sessionToWorkflow(session({ title: '???' }), []).workflow.id).toBe('exported-session'); + }); + + it('omits a trailing interrupted tool-loop turn from the node chain, keeping it in metadata (matches resume)', () => { + const def = sessionToWorkflow(session(), [ + msg(0, 'user', [{ type: 'text', text: 'q1' }]), + msg(1, 'assistant', [{ type: 'text', text: 'a1' }]), // a completed exchange + msg(2, 'user', [{ type: 'text', text: 'q2 use a tool' }]), // interrupted — no final assistant text + msg(3, 'assistant', [{ type: 'tool_call', id: 'c1', name: 'read_file', args: {} }]), + msg(4, 'tool', [{ type: 'tool_result', toolCallId: 'c1', result: 'ok', isError: false }]), + ]); + // only the completed turn becomes a node (consistent with reconstructSessionState's rollback)… + expect(def.workflow.nodes.map((n) => n.id)).toEqual(['input', 'turn-1', 'output']); + const turn1 = def.workflow.nodes[1]; + expect(turn1?.type === 'agent' && turn1.prompt_template).toBe('q1'); + // …but the interrupted attempt is still preserved verbatim in the transcript + expect(JSON.stringify(def.workflow.metadata)).toContain('q2 use a tool'); + }); +}); + +describe('serializeWorkflow (1.Z) — deterministic, round-trippable YAML emitter', () => { + const transcript = [ + msg(0, 'user', [{ type: 'text', text: 'hello' }]), + msg(1, 'assistant', [{ type: 'text', text: 'hi there' }]), + ]; + + it('round-trips: an exported session parses as a valid workflow, and parse→serialize is byte-stable', () => { + const def = sessionToWorkflow(session(), transcript); + const yaml1 = serializeWorkflow(def); + const parsed = parseWorkflow(yaml1); // must not throw — the export is a valid workflow + const yaml2 = serializeWorkflow(parsed); + expect(yaml2).toBe(yaml1); // byte-stable round-trip (including metadata) + // the parsed workflow's agent nodes mirror the turns + expect(parsed.workflow.nodes.filter((n) => n.type === 'agent').map((n) => n.id)).toEqual([ + 'turn-1', + ]); + }); + + it('is deterministic (sorted keys) and stable across repeated emits', () => { + const def = sessionToWorkflow(session(), transcript); + expect(serializeWorkflow(def)).toBe(serializeWorkflow(def)); + const yaml = serializeWorkflow(def); + // sorted map keys: within the document, `schema_version` precedes `workflow` + expect(yaml.indexOf('schema_version')).toBeLessThan(yaml.indexOf('workflow:')); + }); + + it('serializes no reasoning signature and no secret (the durable transcript is structurally clean)', () => { + const yaml = serializeWorkflow( + sessionToWorkflow(session(), [ + msg(0, 'user', [{ type: 'text', text: 'think about it' }]), + msg(1, 'assistant', [ + { type: 'reasoning', text: 'internal reasoning here' }, + { type: 'text', text: 'the answer' }, + ]), + ]), + ); + expect(yaml).toContain('internal reasoning here'); // reasoning TEXT is preserved in the transcript + expect(yaml).not.toContain('signature'); // …but the ephemeral signature never is (ADR-0030) + expect(yaml).not.toMatch(/secret:\s*true/); // no MaskedSecret value (ADR-0029) + }); + + it('neutralizes interpolation syntax a user typed, so the export still parses + round-trips', () => { + // A user literally typing `{{ secrets.X }}` in chat would otherwise make prompt_template fail the + // parse-time secret-taint gate (ADR-0029) — breaking the round-trip. The opener is neutralized. + const def = sessionToWorkflow(session(), [ + msg(0, 'user', [{ type: 'text', text: 'how do I reference {{ secrets.OPENAI_KEY }} here?' }]), + msg(1, 'assistant', [{ type: 'text', text: 'you do not — it stays in the keychain' }]), + ]); + const turn1 = def.workflow.nodes[1]; + expect(turn1?.type === 'agent' && turn1.prompt_template).toBe( + 'how do I reference { { secrets.OPENAI_KEY }} here?', + ); + const yaml1 = serializeWorkflow(def); + expect(() => parseWorkflow(yaml1)).not.toThrow(); // the neutralized prompt is no longer a secret reference + expect(serializeWorkflow(parseWorkflow(yaml1))).toBe(yaml1); // still byte-stable + // the FULL verbatim text is preserved untouched in the metadata transcript (not interpolation-scanned) + expect(JSON.stringify(def.workflow.metadata)).toContain('{{ secrets.OPENAI_KEY }}'); + }); +}); diff --git a/packages/core/src/export/serializer.ts b/packages/core/src/export/serializer.ts new file mode 100644 index 00000000..7e696a37 --- /dev/null +++ b/packages/core/src/export/serializer.ts @@ -0,0 +1,222 @@ +/** + * Session **export-to-workflow** (1.Z) — the inverse of the parser (1.L is parse-only). Two pure functions: + * + * - {@link serializeWorkflow} emits a {@link WorkflowDefinition} as **deterministic, round-trippable** YAML + * (map keys sorted alphabetically, array order preserved) so `parse → serialize` is byte-stable. + * - {@link sessionToWorkflow} maps a persisted session (`AgentSessionRecord` + its ordered + * `SessionMessage[]`, 1.X) into a **linear-chain scaffold** `WorkflowDefinition` + * ([ADR-0026](../../../../docs/decisions/0026-session-export-to-workflow.md); the mapping contract lives + * in [agent-session-spec.md](../../../../docs/reference/contracts/agent-session-spec.md) §"Export to + * workflow" §"Precise mapping"): one `agent` node per assistant turn, `input → turn-1 → … → output` + * edges, the bound agent inline, and the **full transcript** under `metadata.relaviumExport`. + * + * Both are deterministic (no wall-clock / randomness) and platform-free — the host (CLI `relavium + * chat-export`, the desktop "Export to Canvas") loads the session via the `@relavium/db` `SessionStore` + * and writes the file; this module never touches the DB or the filesystem. A `secret` value can never + * appear (secrets never reach a message, ADR-0029) and a reasoning `signature` can never appear (the + * transcript is `DurableContentPart`, which structurally omits it, ADR-0030). + */ + +import { + SCHEMA_VERSION, + type AgentSessionRecord, + type DurableContentPart, + type SessionMessage, +} from '@relavium/shared'; +import { stringify as stringifyYaml } from 'yaml'; + +import type { WorkflowDefinition } from '../parser.js'; + +type WorkflowSpec = WorkflowDefinition['workflow']; +type WorkflowNode = WorkflowSpec['nodes'][number]; +type WorkflowEdge = WorkflowSpec['edges'][number]; +type AgentNode = Extract; + +/** + * Emit a {@link WorkflowDefinition} as YAML. Map keys are sorted alphabetically and array element order is + * preserved, so the output is deterministic and `parse(serialize(def))` followed by another serialize is + * byte-stable. Faithful: it serializes exactly what it is given — the secret/signature exclusion is upheld + * by the durable types that feed it (see {@link sessionToWorkflow}), not by scrubbing here. + */ +export function serializeWorkflow(workflow: WorkflowDefinition): string { + return stringifyYaml(workflow, { sortMapEntries: true }); +} + +/** The concatenated `text` parts of a durable content array (non-text parts are dropped). */ +function textOf(content: readonly DurableContentPart[]): string { + return content + .filter((part): part is Extract => part.type === 'text') + .map((part) => part.text) + .join('\n\n'); +} + +/** The tool names invoked in a turn — the `tool_call` parts of an assistant message's content. */ +function toolsUsedIn(content: readonly DurableContentPart[]): string[] { + return content + .filter( + (part): part is Extract => + part.type === 'tool_call', + ) + .map((part) => part.name); +} + +/** + * Neutralize interpolation openers in user free-text destined for `prompt_template`. The exporter copies a + * chat message verbatim, but `prompt_template` is parsed as a **template** (workflow-yaml-spec.md), so a + * stray `{{ secrets.X }}` a user happened to type would be rejected by the parse-time secret-taint gate + * (ADR-0029) and break the round-trip — and `{{ inputs.X }}` would silently become a live reference, which + * literal chat text is not. Inserting a space after each `{` that opens a `{{` digraph keeps the text + * readable, prevents any synthetic reference, and round-trips byte-stably. The FULL verbatim text is + * preserved untouched under `metadata.relaviumExport` (metadata is not interpolation-scanned). + */ +function neutralizeInterpolation(text: string): string { + return text.replace(/\{(?=\{)/g, '{ '); +} + +/** A logical turn: the contiguous `user` message(s) plus the assistant/tool messages that answer them. */ +interface TurnDraft { + promptSegments: string[]; + toolNames: string[]; + /** Any assistant message (incl. a tool_call-only one) — marks the turn answered, so the next `user` is new. */ + hasAssistant: boolean; + /** An assistant message that produced final text — marks a COMPLETED exchange (gates node promotion). */ + hasAssistantText: boolean; +} + +/** + * Segment an ordered transcript into logical TURNS. The spec maps one `agent` node per **turn**, not per + * assistant message — a host may persist a single turn as split rows (`user → assistant(tool_call) → tool → + * assistant(text)`), and emitting a node per assistant message would split that one turn into two (the second + * losing its prompt + tools). A turn begins at a `user` message that follows an already-answered turn; + * contiguous `user` messages merge into one prompt; `tool`/`system` messages are not delimiters and add no + * node content (a turn's tools come from its assistant messages' `tool_call` parts). Mirrors the turn model + * `reconstructSessionState` (1.Y) uses. + */ +function groupIntoTurns(ordered: readonly SessionMessage[]): TurnDraft[] { + const turns: TurnDraft[] = []; + let current: TurnDraft | null = null; + for (const message of ordered) { + if (message.role === 'user') { + if (current?.hasAssistant) { + turns.push(current); + current = null; + } + current ??= { + promptSegments: [], + toolNames: [], + hasAssistant: false, + hasAssistantText: false, + }; + const text = textOf(message.content); + if (text.length > 0) { + current.promptSegments.push(text); + } + } else if (message.role === 'assistant') { + current ??= { + promptSegments: [], + toolNames: [], + hasAssistant: false, + hasAssistantText: false, + }; + current.hasAssistant = true; + if (textOf(message.content).length > 0) { + current.hasAssistantText = true; // a final-text assistant message completes the exchange + } + current.toolNames.push(...toolsUsedIn(message.content)); + } + } + if (current !== null) { + turns.push(current); + } + return turns; +} + +/** + * A kebab-case workflow id derived deterministically from the session (its title, else a fixed default). + * Built by splitting on non-alphanumeric runs and rejoining with single dashes — identical output to a + * collapse-then-trim, but with no anchored-alternation regex (the only pattern left is one bounded char + * class, which is linear), so it avoids the false-positive ReDoS hotspot Sonar raises on `/^-+|-+$/`. + * + * Only ASCII alphanumerics survive — `kebabIdSchema` (the id's contract) is ASCII, so a non-ASCII title is + * stripped to its ASCII run(s) (e.g. Turkish "İstanbul Sohbeti" → "stanbul-sohbeti") or falls back to + * `exported-session` when nothing ASCII remains. Unicode slugs are a separate future concern; the scaffold's + * id is human-reviewed and renameable on the canvas. + */ +function workflowIdFor(record: AgentSessionRecord): string { + const slug = (record.title ?? '') + .toLowerCase() + .split(/[^a-z0-9]+/) + .filter((segment) => segment.length > 0) + .join('-'); + return slug.length > 0 ? slug : 'exported-session'; +} + +/** + * Map a persisted session + its ordered transcript into a linear-chain scaffold `WorkflowDefinition` + * (ADR-0026). Deterministic — the same `record` + `messages` always produce the same definition (no + * wall-clock / randomness), so the emitted YAML round-trips. Assumes a well-formed, user-initiated + * transcript (each turn opens with a `user` message, as `AgentSession` emits); a malformed assistant-first + * transcript still yields a valid workflow, just with a prompt-less leading agent node. + */ +export function sessionToWorkflow( + record: AgentSessionRecord, + messages: readonly SessionMessage[], +): WorkflowDefinition { + const ordered = [...messages].sort((a, b) => a.sequenceNumber - b.sequenceNumber); + + const nodes: WorkflowNode[] = [{ id: 'input', type: 'input' }]; + const edges: WorkflowEdge[] = []; + let previousNodeId = 'input'; + let turnIndex = 0; + + for (const turn of groupIntoTurns(ordered)) { + if (!turn.hasAssistantText) { + // Promote only COMPLETED exchanges (a turn that produced final assistant text). A user-only turn or an + // interrupted tool-loop turn (assistant tool_call + tool result, no final text) is skipped — matching + // reconstructSessionState's rollback (1.Y) so export and resume agree on what a turn is. The raw attempt + // is still preserved verbatim under metadata.relaviumExport. + continue; + } + turnIndex += 1; + const nodeId = `turn-${turnIndex}`; + const prompt = neutralizeInterpolation(turn.promptSegments.join('\n\n')); + const tools = [...new Set(turn.toolNames)]; // dedupe across the turn, first-seen order (determinism-safe) + const node: AgentNode = { + id: nodeId, + type: 'agent', + agent_ref: record.agentSlug, + ...(prompt.length > 0 ? { prompt_template: prompt } : {}), + ...(tools.length > 0 ? { tools } : {}), + }; + nodes.push(node); + edges.push({ from: previousNodeId, to: nodeId }); + previousNodeId = nodeId; + } + + nodes.push({ id: 'output', type: 'output' }); + edges.push({ from: previousNodeId, to: 'output' }); + + // The full transcript under one reserved metadata key (agent-session-spec.md §"Precise mapping"). + const metadata: Record = { + relaviumExport: { + source: 'session', + sessionId: record.id, + agentSlug: record.agentSlug, + ...(record.title === undefined ? {} : { title: record.title }), + createdAt: record.createdAt, + updatedAt: record.updatedAt, + messages: ordered, + }, + }; + + const spec: WorkflowSpec = { + id: workflowIdFor(record), + ...(record.title === undefined ? {} : { name: record.title }), + ...(record.agentSnapshot === undefined ? {} : { agents: [record.agentSnapshot] }), + nodes, + edges, + metadata, + }; + + return { schema_version: SCHEMA_VERSION, workflow: spec }; +} diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index a8fa7330..860ef344 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -183,12 +183,22 @@ export type { SessionStreamEvent, SessionLifecycleEvent, } from './engine/agent-session.js'; +// Session checkpoint/resume (1.Y) — reconstruct the in-flight state from a persisted transcript (1.X) so a +// session continues after a restart; the host loads via the @relavium/db SessionStore and hands the result +// to AgentSession.resume. Directly-stored, not event-sourced (ADR-0003); reuses the 1.R idempotency principle. +export { reconstructSessionState } from './engine/session-resume.js'; +export type { SessionResumeState } from './engine/session-resume.js'; // 1.W — the session:* namespace on the shared bus: the SessionEventSink→RunEventBus adapter (attaches the // sessionId; the bus stamps the per-session sequenceNumber) and the SessionHandle (mirrors RunHandle, // scoped to sessionId, terminal on session:cancelled). See sse-event-schema.md §"The session stream". export { createSessionHandle, createSessionEventSink } from './engine/session-handle.js'; export type { SessionHandle, SessionStreamHandleEvent } from './engine/session-handle.js'; +// Session export-to-workflow (1.Z) — the inverse of parseWorkflow (1.L is parse-only). `serializeWorkflow` +// emits a WorkflowDefinition as deterministic, round-trippable YAML; `sessionToWorkflow` maps a persisted +// session + transcript into a linear-chain scaffold (ADR-0026; agent-session-spec.md §"Export to workflow"). +export { serializeWorkflow, sessionToWorkflow } from './export/serializer.js'; + // Node-type handlers (1.P) — the six non-agent NodeExecutor arms (condition / transform / fan_out / // fan_in / input / output) plus the dispatcher that composes them (and the 1.O agent arm) into the one // executor the engine holds. `createStandardNodeExecutor` is the convenience wirer; the per-type diff --git a/packages/shared/src/session.test.ts b/packages/shared/src/session.test.ts index 15a01c31..aacd8424 100644 --- a/packages/shared/src/session.test.ts +++ b/packages/shared/src/session.test.ts @@ -146,6 +146,11 @@ describe('AgentSessionSchema', () => { }; expect(AgentSessionSchema.safeParse(noContext).success).toBe(false); expect(AgentSessionSchema.safeParse(baseSession({ agentSlug: '' })).success).toBe(false); + // agentSlug is kebab-case (it IS an agent_ref) — a non-kebab slug is rejected, so it transfers + // verbatim into an exported workflow's kebab `agent_ref` without breaking the round-trip (1.Z). + expect(AgentSessionSchema.safeParse(baseSession({ agentSlug: 'Not Kebab' })).success).toBe( + false, + ); expect(AgentSessionSchema.safeParse(baseSession({ totalInputTokens: -1 })).success).toBe(false); expect(AgentSessionSchema.safeParse(baseSession({ totalCostMicrocents: -1 })).success).toBe( false, diff --git a/packages/shared/src/session.ts b/packages/shared/src/session.ts index c81a3ff7..4eb1cf41 100644 --- a/packages/shared/src/session.ts +++ b/packages/shared/src/session.ts @@ -1,7 +1,7 @@ import { z } from 'zod'; import { AgentSchema } from './agent.js'; -import { nonEmptyString, nonNegativeInt } from './common.js'; +import { kebabIdSchema, nonEmptyString, nonNegativeInt } from './common.js'; import { DurableContentPartSchema } from './content.js'; import { SessionContextSchema } from './run-event.js'; @@ -67,8 +67,9 @@ export type SessionMessage = z.infer; */ export const AgentSessionSchema = z.object({ id: nonEmptyString, - /** The authored `agent_ref` the session is bound to (`agent_sessions.agent_slug`). */ - agentSlug: nonEmptyString, + /** The authored `agent_ref` the session is bound to (`agent_sessions.agent_slug`) — kebab-case, like every + * `agent.id` / `agent_ref`, so it transfers verbatim into an exported workflow's `agent_ref` (1.Z). */ + agentSlug: kebabIdSchema, /** The catalog `agents.id` this session resolved to, when the agent was a stored catalog entry. */ agentId: nonEmptyString.optional(), /** Frozen agent config for reproducibility (`agent_sessions.agent_snapshot`). */