diff --git a/docs/architecture/session-copy-cost-baseline.md b/docs/architecture/session-copy-cost-baseline.md new file mode 100644 index 0000000000..cc434e2c08 --- /dev/null +++ b/docs/architecture/session-copy-cost-baseline.md @@ -0,0 +1,94 @@ + + +# Session copy cost baseline + +This document records the current storage boundary for Session branch/revision +copies and defines the reproducible baseline for issue #5439. + +## Current boundary + +The Host treats a branch or revision as a materialized, self-contained Session. +`packages/runtime-host/src/server/session-revision-coordinator.ts` captures a +settled source slice, creates the target Session, then publishes copied state +across the following stores: + +| Domain | Current copy behavior | Identity after copy | +| --- | --- | --- | +| Session header | New target row; source metadata is projected into it | New Session id | +| transcript messages | Runtime ledger is replayed into target projections | New message/run/event ids where the domain owns identity | +| Runtime Events | Immutable events are cloned into the target Session | New target Session/run/event identities | +| AgentRun operational events | Copied for the selected run closure | New target run/event identities | +| model-call facts | Rebuilt from the copied run ledger and projection transitions | New target ledger facts; source usage facts are not shared | +| task ledger / Todo | Target Todo state is initialized from the selected branch boundary | New target Session ownership | +| artifacts | Payloads are copied into target-owned artifact records | New artifact ids and target paths | +| context-offload references | Reference rows are copied, while the content-addressed `blobId` is reused | New target reference; shared blob payload | +| linked child Sessions | Rejected, preserved with validated external ids, or snapshotted depending on copy kind | Explicitly controlled per reference mode | + +The important existing sharing boundary is context offload: `copyReferences` +creates target-Session references without duplicating the blob bytes. The +ordinary conversation copy path does not currently apply that ownership/payload +split to transcript, Runtime Events, model-call facts, Todo state, or artifact +payloads. + +## Measurement + +Run the benchmark after building the workspaces: + +```sh +npm run build +npm --workspace @maka/runtime-host run benchmark:session-copy +``` + +The benchmark creates real settled turns through the Runtime Host, invokes the +production `session.branch.create` handler, and reports: + +- median copy latency in milliseconds after one warmup copy and five measured + samples; +- storage bytes added under the temporary interactive root; +- source and child projected message counts; +- source and child invocation counts; +- message-count amplification. + +The default fixtures use 4, 16, 64, and 128 settled turns. Each turn contains a +256-byte user payload and uses the deterministic FakeBackend. This is a +transcript/runtime-ledger baseline; it intentionally does not claim artifact or +context-offload coverage until fixtures for those payload classes are added. + +The byte delta is measured after the warmup copy and immediately before and +after the measured copies, then reported per measured copy. It includes SQLite +journaling and metadata writes, so the number is an operational write-volume +proxy rather than a logical row-size estimate. Run the same command on a clean +machine and record the output with the commit, Node version, filesystem, and +platform. + +## Interpretation + +This baseline answers the first question in #5439: whether the current copy +cost grows with the retained history and whether the dominant growth is in +runtime/transcript state. It does not choose a shared-prefix design. A future +prototype must add comparable measurements for: + +1. artifact payloads and `session_file` references; +2. context-offload references and shared blob bytes; +3. export/import closure materialization; +4. parent continuation, parent deletion, repair, and restart stability. + +The baseline should be kept as the materialized-copy control group when those +experiments land. diff --git a/packages/runtime-host/package.json b/packages/runtime-host/package.json index d0c8f08a9e..f285d6a564 100644 --- a/packages/runtime-host/package.json +++ b/packages/runtime-host/package.json @@ -24,6 +24,7 @@ "scripts": { "clean": "node ../../scripts/clean-paths.mjs dist tsconfig.tsbuildinfo", "build": "tsc -p tsconfig.json", + "benchmark:session-copy": "node scripts/session-copy-cost-benchmark.mjs", "benchmark:transcript": "node scripts/transcript-data-plane-benchmark.mjs", "typecheck": "tsc -p tsconfig.json --noEmit", "test:dist": "node --test \"dist/**/*.test.js\"" diff --git a/packages/runtime-host/scripts/session-copy-cost-benchmark.mjs b/packages/runtime-host/scripts/session-copy-cost-benchmark.mjs new file mode 100644 index 0000000000..f82a8d92f1 --- /dev/null +++ b/packages/runtime-host/scripts/session-copy-cost-benchmark.mjs @@ -0,0 +1,211 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { performance } from 'node:perf_hooks'; +import { mkdtemp, readdir, rm, stat } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { randomUUID } from 'node:crypto'; +import { FakeBackend } from '@maka/runtime/test-only/fake-backend'; +import { SessionManager } from '@maka/runtime/session-manager'; +import { resolveStorageRoot, tryAcquireInteractiveRootOwner } from '@maka/storage/root-authority'; +import { createExecutionRuntimeHostComposition } from '../dist/server/execution-composition.js'; + +const HISTORY_SIZES = [4, 16, 64, 128]; +const COPY_SAMPLE_COUNT = 5; +const COPY_RETRY_ATTEMPTS = 20; +const COPY_RETRY_DELAY_MS = 25; +const CONNECTION_ID = 'bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb'; +const operationContext = { + hostEpoch: 'session-copy-cost-benchmark', + connectionId: 'benchmark-client', + principal: 'local_os_user', + acquireResidency: () => ({ release() {} }), +}; + +const results = []; +for (const historySize of HISTORY_SIZES) { + results.push(await runFixture(historySize)); +} +console.table(results); + +async function runFixture(historySize) { + const root = await mkdtemp(join(tmpdir(), 'maka-session-copy-benchmark-')); + const capability = await resolveStorageRoot({ path: root, kind: 'interactive' }); + const owner = await tryAcquireInteractiveRootOwner(capability); + if (!owner) throw new Error(`Could not acquire benchmark storage root: ${root}`); + + let composition; + let manager; + const originalRecover = SessionManager.prototype.recoverInterruptedSessionsStrict; + SessionManager.prototype.recoverInterruptedSessionsStrict = async function (stores) { + manager = this; + return originalRecover.call(this, stores); + }; + try { + composition = await createExecutionRuntimeHostComposition( + { + owner, + hostEpoch: operationContext.hostEpoch, + acquireResidency: () => ({ release() {} }), + retainUntilProcessExit: () => undefined, + requestDrain: () => undefined, + }, + {}, + { + primaryBackendFactory: (context) => new FakeBackend(context), + }, + ); + await composition.recover(); + + if (!manager) throw new Error('Execution composition did not construct a SessionManager'); + const source = await manager.createSession({ + cwd: root, + llmConnectionId: CONNECTION_ID, + llmConnectionSlug: 'fake', + model: 'fake-model', + permissionMode: 'bypass', + name: `copy-cost-${historySize}`, + }); + let sourceTurnId; + for (let index = 0; index < historySize; index += 1) { + sourceTurnId = `turn-${index}`; + const started = await composition.handlers['turn.start']( + { + sessionId: source.id, + turnId: sourceTurnId, + content: { text: `benchmark turn ${index} ${'x'.repeat(256)}` }, + }, + operationContext, + ); + if (!started.ok) throw new Error(`Turn ${index} failed to start: ${started.reason}`); + await waitForCompletedTurn(manager, source.id, sourceTurnId); + } + + const sourceSummary = (await manager.listSessions()).find((item) => item.id === source.id); + if (!sourceSummary || !sourceTurnId) throw new Error('Benchmark source did not settle'); + + await copySession({ composition, manager, sourceSessionId: source.id, sourceTurnId }); + const beforeBytes = await directoryBytes(root); + const copyDurations = []; + let targetSessionId; + for (let sample = 0; sample < COPY_SAMPLE_COUNT; sample += 1) { + targetSessionId = randomUUID(); + const startedAt = performance.now(); + await copySession({ + composition, + manager, + sourceSessionId: source.id, + sourceTurnId, + targetSessionId, + }); + copyDurations.push(performance.now() - startedAt); + } + const afterBytes = await directoryBytes(root); + const sourceMessages = await manager.getMessages(source.id); + const childMessages = await manager.getMessages(targetSessionId); + const sourceInvocations = await manager.listInvocations(source.id); + const childInvocations = await manager.listInvocations(targetSessionId); + + return { + historySize, + copyMsMedian: median(copyDurations).toFixed(1), + addedMiB: ((afterBytes - beforeBytes) / COPY_SAMPLE_COUNT / (1024 * 1024)).toFixed(3), + sourceMessages: sourceMessages.length, + childMessages: childMessages.length, + sourceRuns: sourceInvocations.length, + childRuns: childInvocations.length, + amplification: (childMessages.length / Math.max(1, sourceMessages.length)).toFixed(2), + }; + } finally { + SessionManager.prototype.recoverInterruptedSessionsStrict = originalRecover; + composition?.beginDrain(); + await composition?.close(); + await owner.close(); + await rm(root, { recursive: true, force: true }); + } +} + +async function copySession({ + composition, + manager, + sourceSessionId, + sourceTurnId, + targetSessionId = randomUUID(), +}) { + let lastOutcome; + for (let attempt = 0; attempt < COPY_RETRY_ATTEMPTS; attempt += 1) { + const sourceRecord = await manager.deps.store.readHeaderRecordSnapshot(sourceSessionId); + const copied = await composition.handlers['session.branch.create']( + { + sourceSessionId, + targetSessionId, + sourceTurnId, + expectedSourceRevision: sourceRecord.revision, + }, + operationContext, + ); + if (copied.ok && copied.result.kind === 'committed') return copied.result; + + lastOutcome = copied; + const retryable = + (copied.ok && copied.result.kind === 'source_revision_conflict') || + (!copied.ok && copied.error.code === 'session_busy'); + if (!retryable) break; + await sleep(COPY_RETRY_DELAY_MS); + } + throw new Error(`Session copy failed after retries: ${JSON.stringify(lastOutcome)}`); +} + +async function waitForCompletedTurn(manager, sessionId, turnId) { + const deadline = performance.now() + 10_000; + while (performance.now() < deadline) { + const messages = await manager.getMessages(sessionId); + const state = messages.findLast( + (message) => message.type === 'turn_state' && message.turnId === turnId, + ); + if (state?.status === 'completed') return; + if (state?.status === 'failed' || state?.status === 'aborted') { + throw new Error(`Benchmark turn ${turnId} ended as ${state.status}`); + } + await new Promise((resolve) => setTimeout(resolve, 5)); + } + throw new Error(`Timed out waiting for benchmark turn ${turnId}`); +} + +function median(values) { + const sorted = [...values].sort((left, right) => left - right); + return sorted[Math.floor(sorted.length / 2)]; +} + +function sleep(delayMs) { + return new Promise((resolve) => setTimeout(resolve, delayMs)); +} + +async function directoryBytes(root) { + const entries = await readdir(root, { withFileTypes: true }); + let total = 0; + for (const entry of entries) { + if (entry.name === '.maka-storage-root.json') continue; + const path = join(root, entry.name); + const info = await stat(path); + total += info.isDirectory() ? await directoryBytes(path) : info.size; + } + return total; +}