From 33697eefa73d7351c23a178770696a89f729dd85 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 8 Sep 2026 22:26:05 -0700 Subject: [PATCH 1/7] Commit session checkpoints through isomorphic git --- bun.lock | 2 + docs/IMPLEMENTATION.md | 8 +- package.json | 4 +- src/session/commit-signer.ts | 67 ++++++ src/session/optimized-context-store.test.ts | 164 +++++--------- src/session/optimized-context-store.ts | 233 +++++++++----------- src/session/session-dir-lock.ts | 31 +++ 7 files changed, 262 insertions(+), 247 deletions(-) create mode 100644 src/session/commit-signer.ts create mode 100644 src/session/session-dir-lock.ts diff --git a/bun.lock b/bun.lock index fe9d7ed87..3434dbde7 100644 --- a/bun.lock +++ b/bun.lock @@ -7,6 +7,7 @@ "dependencies": { "@intx/agent": "workspace:*", "@intx/authz": "workspace:*", + "@intx/crypto": "0.3.0", "@intx/harness": "workspace:*", "@intx/hub-sessions": "0.3.0", "@intx/inference": "workspace:*", @@ -23,6 +24,7 @@ "@opentui/core": "0.5.10", "arktype": "catalog:", "highlight.js": "^11.11.1", + "isomorphic-git": "catalog:", }, "devDependencies": { "@eslint/js": "^9.39.0", diff --git a/docs/IMPLEMENTATION.md b/docs/IMPLEMENTATION.md index 69798ff5c..3393e7911 100644 --- a/docs/IMPLEMENTATION.md +++ b/docs/IMPLEMENTATION.md @@ -415,10 +415,10 @@ error collector; there is no added retry subsystem or side-effect rollback. `createOptimizedContextStore` (`src/session/optimized-context-store.ts`) wraps the Interchange git store to keep per-checkpoint cost independent of session length. -Checkpoint commits go through system git and use the operator's global -`user.name` / `user.email` when both are set, so commit-author hooks see a real -identity; otherwise they fall back to Interchange's harness author -(`interchange-harness`, `harness@interchange.local`). +Checkpoint commits go through isomorphic-git (`base.commit()` after staging extra +segment files and blobs). Author and committer are Interchange's harness identity +(`interchange-harness`, `harness@interchange.local`); a `CommitSigner` from +`commit-signer.ts` signs each commit. The wrapper never shells out to system git. The append-only snapshots (`turns.jsonl`, `prompt.jsonl`) are written as rolling segments (`turns-0001.jsonl`, ...) that seal at 256KB, so `git add` re-hashes only the small active segment instead of the whole growing file. Segment zero keeps the diff --git a/package.json b/package.json index 6f16bc1e1..d3aa34638 100644 --- a/package.json +++ b/package.json @@ -82,6 +82,7 @@ "dependencies": { "@intx/agent": "workspace:*", "@intx/authz": "workspace:*", + "@intx/crypto": "0.3.0", "@intx/inference": "workspace:*", "@intx/log": "workspace:*", "@intx/storage-isogit": "workspace:*", @@ -97,7 +98,8 @@ "@modelcontextprotocol/sdk": "^1.29.0", "@opentui/core": "0.5.10", "arktype": "catalog:", - "highlight.js": "^11.11.1" + "highlight.js": "^11.11.1", + "isomorphic-git": "catalog:" }, "devDependencies": { "@eslint/js": "^9.39.0", diff --git a/src/session/commit-signer.ts b/src/session/commit-signer.ts new file mode 100644 index 000000000..eb9832d63 --- /dev/null +++ b/src/session/commit-signer.ts @@ -0,0 +1,67 @@ +import fs from "node:fs"; +import path from "node:path"; +import { type } from "arktype"; +import { createSSHSignature, generateKeyPair } from "@intx/crypto"; +import type { CommitSigner } from "@intx/storage-isogit/node"; +import { getLogger } from "@intx/log"; + +import { LOG_NAMESPACE_ROOT } from "../branding.js"; + +const KEY_DIR = "keys"; +const KEY_FILE = "commit-ed25519.json"; + +const log = getLogger([LOG_NAMESPACE_ROOT, "session", "commit-signer"]); + +const PersistedKeyPair = type({ + privateKey: "string", + publicKey: "string", +}); + +function decodeKey(label: string, value: string): Uint8Array { + const bytes = Buffer.from(value, "base64"); + if (bytes.length !== 32) { + throw new Error(`${label} must be 32 bytes, got ${bytes.length}`); + } + return new Uint8Array(bytes); +} + +async function loadPersistedKeyPair( + filePath: string, +): Promise<{ privateKey: Uint8Array; publicKey: Uint8Array } | null> { + let raw: string; + try { + raw = await fs.promises.readFile(filePath, "utf8"); + } catch (cause) { + if (cause instanceof Error && "code" in cause && cause.code === "ENOENT") return null; + throw cause; + } + const parsed = PersistedKeyPair(JSON.parse(raw) as unknown); + if (parsed instanceof type.errors) { + throw new Error(`Invalid commit signing key at ${filePath}: ${parsed.summary}`); + } + return { + privateKey: decodeKey("privateKey", parsed.privateKey), + publicKey: decodeKey("publicKey", parsed.publicKey), + }; +} + +export async function loadOrCreateCommitSigner(dir: string): Promise { + const keyDir = path.join(dir, KEY_DIR); + const filePath = path.join(keyDir, KEY_FILE); + let keyPair = await loadPersistedKeyPair(filePath); + if (keyPair === null) { + const generated = await generateKeyPair(); + await fs.promises.mkdir(keyDir, { recursive: true }); + await fs.promises.writeFile( + filePath, + JSON.stringify({ + privateKey: Buffer.from(generated.privateKey).toString("base64"), + publicKey: Buffer.from(generated.publicKey).toString("base64"), + }), + { encoding: "utf8", mode: 0o600 }, + ); + log.debug?.("wrote session commit signing key"); + keyPair = generated; + } + return (payload) => createSSHSignature(payload, keyPair.privateKey, keyPair.publicKey); +} diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index b888f536e..76dd7d824 100644 --- a/src/session/optimized-context-store.test.ts +++ b/src/session/optimized-context-store.test.ts @@ -2,8 +2,13 @@ import { describe, test, expect } from "bun:test"; import fs from "node:fs"; import os from "node:os"; import path from "node:path"; +import git, { type CommitObject } from "isomorphic-git"; import type { ConversationTurn } from "@intx/types/runtime"; -import { createOptimizedContextStore, loadRecentTurns } from "./optimized-context-store.js"; +import { + createOptimizedContextStore, + createSessionStores, + loadRecentTurns, +} from "./optimized-context-store.js"; import { segmentFileName, listSegmentFiles } from "./incremental-jsonl.js"; const TURNS_FILE = "turns.jsonl"; @@ -12,48 +17,11 @@ function tempDir(): string { return fs.mkdtempSync(path.join(os.tmpdir(), "opt-store-")); } -function isolatedGitEnv(gitconfig: string): NodeJS.ProcessEnv { - const dir = tempDir(); - const config = path.join(dir, "gitconfig"); - fs.writeFileSync(config, gitconfig); - return { - ...process.env, - GIT_CONFIG_GLOBAL: config, - GIT_CONFIG_SYSTEM: "/dev/null", - GIT_CONFIG_NOSYSTEM: "1", - HOME: dir, - XDG_CONFIG_HOME: dir, - }; -} - -async function headIdent(dir: string): Promise<{ - authorName: string; - authorEmail: string; - committerName: string; - committerEmail: string; -}> { - const proc = Bun.spawn(["git", "-C", dir, "log", "-1", "--format=%an%n%ae%n%cn%n%ce"], { - stdout: "pipe", - stderr: "pipe", - }); - const [exitCode, stdout, stderr] = await Promise.all([ - proc.exited, - new Response(proc.stdout).text(), - new Response(proc.stderr).text(), - ]); - if (exitCode !== 0) { - throw new Error(`git log failed: ${stderr.trim() || stdout.trim()}`); - } - const [authorName, authorEmail, committerName, committerEmail] = stdout.trimEnd().split("\n"); - if ( - authorName === undefined || - authorEmail === undefined || - committerName === undefined || - committerEmail === undefined - ) { - throw new Error(`unexpected git log identity output: ${JSON.stringify(stdout)}`); - } - return { authorName, authorEmail, committerName, committerEmail }; +async function headCommit(dir: string): Promise { + const [entry] = await git.log({ fs, dir, depth: 1 }); + if (entry === undefined) throw new Error("no commit"); + const { commit } = await git.readCommit({ fs, dir, oid: entry.oid }); + return commit; } const EMPTY_CHECKPOINT_METADATA = { @@ -61,22 +29,12 @@ const EMPTY_CHECKPOINT_METADATA = { tokenUsage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, thinking: 0 }, }; -async function commitEmptyCheckpoint( - dir: string, - opts?: Parameters[1], -): Promise { - const store = await createOptimizedContextStore(dir, opts); +async function commitEmptyCheckpoint(dir: string): Promise { + const store = await createOptimizedContextStore(dir); await store.writeMetadata(EMPTY_CHECKPOINT_METADATA); await store.commit({ message: "checkpoint: tool-execution" }); } -const HARNESS_IDENT = { - authorName: "interchange-harness", - authorEmail: "harness@interchange.local", - committerName: "interchange-harness", - committerEmail: "harness@interchange.local", -}; - function turn(text: string): ConversationTurn { return { role: "user", content: [{ type: "text", text }], timestamp: 1 }; } @@ -578,6 +536,9 @@ describe("createOptimizedContextStore checkpoint", () => { expect((await listSegmentFiles(dir, TURNS_FILE)).length).toBeGreaterThan(1); + const treeFiles = await git.listFiles({ fs, dir, ref: "HEAD" }); + expect(treeFiles.some((name) => /^turns-\d+\.jsonl$/.test(name))).toBe(true); + const reloaded = await createOptimizedContextStore(dir); const loaded = await reloaded.load(); expect(loaded.turns).toHaveLength(total); @@ -587,71 +548,46 @@ describe("createOptimizedContextStore checkpoint", () => { expect(atHead).toHaveLength(total); }, 20_000); - test("records the operator identity as author and committer from global git config", async () => { + test("signs checkpoints as the interchange harness author", async () => { const dir = tempDir(); - await commitEmptyCheckpoint(dir, { - env: isolatedGitEnv(`[user]\n\tname = Sawyer\n\temail = sawyer@dirtroad.dev\n`), - }); - - expect(await headIdent(dir)).toEqual({ - authorName: "Sawyer", - authorEmail: "sawyer@dirtroad.dev", - committerName: "Sawyer", - committerEmail: "sawyer@dirtroad.dev", - }); + await commitEmptyCheckpoint(dir); + const commit = await headCommit(dir); + expect(commit.author.name).toBe("interchange-harness"); + expect(commit.author.email).toBe("harness@interchange.local"); + expect(commit.committer.name).toBe("interchange-harness"); + expect(commit.committer.email).toBe("harness@interchange.local"); + expect(commit.gpgsig).toContain("BEGIN SSH SIGNATURE"); }); - test("records an injected author as both author and committer", async () => { + test("does not spawn system git while committing", async () => { const dir = tempDir(); - await commitEmptyCheckpoint(dir, { - author: { name: "Sawyer", email: "sawyer@dirtroad.dev" }, - }); - - expect(await headIdent(dir)).toEqual({ - authorName: "Sawyer", - authorEmail: "sawyer@dirtroad.dev", - committerName: "Sawyer", - committerEmail: "sawyer@dirtroad.dev", - }); + const gitSpawns: string[][] = []; + const original = Bun.spawn; + Bun.spawn = ((cmd: unknown, opts?: unknown) => { + const argv = Array.isArray(cmd) + ? cmd.map(String) + : typeof cmd === "object" && cmd !== null && "cmd" in cmd && Array.isArray(cmd.cmd) + ? cmd.cmd.map(String) + : []; + if (argv[0] === "git" || argv[0]?.endsWith("/git")) gitSpawns.push(argv); + return original(cmd as Parameters[0], opts as Parameters[1]); + }) as typeof Bun.spawn; + try { + await commitEmptyCheckpoint(dir); + } finally { + Bun.spawn = original; + } + expect(gitSpawns).toEqual([]); }); +}); - test("falls back to the harness identity when global config is missing", async () => { +describe("createSessionStores", () => { + test("exposes the same object as ContextStore and AuditStore", async () => { const dir = tempDir(); - await commitEmptyCheckpoint(dir, { env: isolatedGitEnv("") }); - expect(await headIdent(dir)).toEqual(HARNESS_IDENT); - }); - - test("falls back when only one of name or email is set", async () => { - const nameOnly = tempDir(); - await commitEmptyCheckpoint(nameOnly, { - env: isolatedGitEnv(`[user]\n\tname = Sawyer\n`), - }); - expect(await headIdent(nameOnly)).toEqual(HARNESS_IDENT); - - const emailOnly = tempDir(); - await commitEmptyCheckpoint(emailOnly, { - env: isolatedGitEnv(`[user]\n\temail = sawyer@dirtroad.dev\n`), - }); - expect(await headIdent(emailOnly)).toEqual(HARNESS_IDENT); - }); - - test("falls back when global name and email are empty or whitespace", async () => { - const bothEmpty = tempDir(); - await commitEmptyCheckpoint(bothEmpty, { - env: isolatedGitEnv(`[user]\n\tname =\n\temail =\n`), - }); - expect(await headIdent(bothEmpty)).toEqual(HARNESS_IDENT); - - const bothWhitespace = tempDir(); - await commitEmptyCheckpoint(bothWhitespace, { - env: isolatedGitEnv(`[user]\n\tname = \n\temail = \n`), - }); - expect(await headIdent(bothWhitespace)).toEqual(HARNESS_IDENT); - - const nameOnlyWhitespaceEmail = tempDir(); - await commitEmptyCheckpoint(nameOnlyWhitespaceEmail, { - env: isolatedGitEnv(`[user]\n\tname = Sawyer\n\temail = \n`), - }); - expect(await headIdent(nameOnlyWhitespaceEmail)).toEqual(HARNESS_IDENT); + const { storage, audit } = await createSessionStores(dir); + expect(storage).toBe(audit); + expect(typeof audit.commitAudit).toBe("function"); + expect(typeof audit.commitErrors).toBe("function"); + expect(typeof audit.loadAudit).toBe("function"); }); }); diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index 245a685f0..5ea3cc502 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -1,9 +1,11 @@ import fs from "node:fs"; import path from "node:path"; import { type } from "arktype"; -import { createIsogitStore } from "@intx/storage-isogit/node"; +import git from "isomorphic-git"; +import { createIsogitStore, type CommitSigner } from "@intx/storage-isogit/node"; import { ContentBlock, + type AuditStore, type ConnectorThreadState, type ConversationTurn, type PendingOperation, @@ -17,8 +19,10 @@ import { readExtraSegmentTexts, segmentFileName, } from "./incremental-jsonl.js"; -import type { ContextCommit, ContextStore } from "@intx/types/runtime"; +import type { ContextStore } from "@intx/types/runtime"; import { LOG_NAMESPACE_ROOT } from "../branding.js"; +import { loadOrCreateCommitSigner } from "./commit-signer.js"; +import { withResolvedDirLock } from "./session-dir-lock.js"; const TURNS_FILE = "turns.jsonl"; const PROMPT_FILE = "prompt.jsonl"; @@ -29,15 +33,13 @@ const TOOL_OUTPUT_DIR = "tool-output"; const log = getLogger([LOG_NAMESPACE_ROOT, "session", "context-store"]); -export interface CheckpointAuthor { - name: string; - email: string; -} - -const HARNESS_AUTHOR: CheckpointAuthor = { - name: "interchange-harness", - email: "harness@interchange.local", -}; +const VENDOR_COMMIT_ROOT_FILES = new Set([ + TURNS_FILE, + PROMPT_FILE, + RESPONSE_FILE, + MANIFEST_FILE, + METADATA_FILE, +]); const BLOB_EXTENSIONS: Readonly> = { "text/plain": ".txt", @@ -304,73 +306,17 @@ export async function loadRecentTurns(dir: string, minTurns: number): Promise { - const proc = Bun.spawn(["git", "-C", dir, ...args], { - stdout: "pipe", - stderr: "pipe", - env: { - ...env, - ...(author === undefined - ? {} - : { - GIT_AUTHOR_NAME: author.name, - GIT_AUTHOR_EMAIL: author.email, - GIT_COMMITTER_NAME: author.name, - GIT_COMMITTER_EMAIL: author.email, - }), - }, - }); - const [exitCode, stdout, stderr] = await Promise.all([ - proc.exited, - new Response(proc.stdout).text(), - new Response(proc.stderr).text(), - ]); - if (exitCode !== 0) { - throw new Error(`git ${args.join(" ")} failed: ${stderr.trim() || stdout.trim()}`); +async function listIndexPaths(dir: string): Promise> { + try { + return new Set(await git.listFiles({ fs, dir })); + } catch { + return new Set(); } - return stdout.trimEnd(); } -async function gitConfigGlobal(key: string, env: NodeJS.ProcessEnv): Promise { - const proc = Bun.spawn(["git", "config", "--global", "--get", key], { - stdout: "pipe", - stderr: "pipe", - env, - }); - const [exitCode, stdout] = await Promise.all([ - proc.exited, - new Response(proc.stdout).text(), - new Response(proc.stderr).text(), - ]); - if (exitCode !== 0) return null; - const value = stdout.trim(); - return value.length > 0 ? value : null; -} - -// Operator commit-author hooks see a real identity; machines without both -// global user.name and user.email still checkpoint via the harness fallback. -async function resolveCheckpointAuthor(env: NodeJS.ProcessEnv): Promise { - const [name, email] = await Promise.all([ - gitConfigGlobal("user.name", env), - gitConfigGlobal("user.email", env), - ]); - if (name === null || email === null) return HARNESS_AUTHOR; - return { name, email }; -} - -/** - * Names of the tail turn segments (`turns-0001.jsonl`, ...) present in a commit - * tree, in segment order. The base store reads the zeroth segment itself; these - * are the segments it does not know about. - */ async function extraSegmentNamesAtCommit(dir: string, hash: string): Promise { - const listing = await runGit(dir, ["ls-tree", "--name-only", hash]); - const present = new Set(listing.split("\n").filter((line) => line.length > 0)); + const listing = await git.listFiles({ fs, dir, ref: hash }); + const present = new Set(listing); const names: string[] = []; for (let index = 1; ; index++) { const name = segmentFileName(TURNS_FILE, index); @@ -380,20 +326,23 @@ async function extraSegmentNamesAtCommit(dir: string, hash: string): Promise { - const [hash, seconds, parents] = (await runGit(dir, ["log", "-1", "--format=%H%n%ct%n%P"])).split( - "\n", - ); - if (hash === undefined || hash.length === 0 || seconds === undefined) { - throw new Error("Unexpected log state after commit: no HEAD"); +async function blobTextAtCommit(dir: string, hash: string, filepath: string): Promise { + const { blob } = await git.readBlob({ fs, dir, oid: hash, filepath }); + return new TextDecoder().decode(blob); +} + +async function resetIndexPaths(dir: string, filepaths: readonly string[]): Promise { + for (const filepath of filepaths) { + try { + await git.resetIndex({ fs, dir, filepath }); + } catch { + // Not in the index; vendor restore already covers its own paths. + } } - const parentHash = parents?.split(" ")[0]; - const base = { hash, message: message.trimEnd(), timestamp: Number(seconds) * 1000 }; - return parentHash !== undefined && parentHash.length > 0 ? { ...base, parentHash } : base; } /** - * Stage every contiguous on-disk segment for `baseName` and `git rm` any + * Stage every contiguous on-disk segment for `baseName` and unstage any * higher-numbered or gapped segment still on disk or tracked after a rewrite * deleted it — even when the in-memory pending set was lost (process died * between heal unlink and commit). Gapped strays are unlinked, not re-added. @@ -411,6 +360,7 @@ async function reconcileSegmentStaging( // tails begin at 1. Empty contiguous (no base) still sweeps numbered files. const startIndex = Math.max(contiguous.length, 1); const highestDisk = await highestSegmentIndex(dir, baseName); + const tracked = await listIndexPaths(dir); for (let index = startIndex; ; index++) { const name = segmentFileName(baseName, index); @@ -421,8 +371,7 @@ async function reconcileSegmentStaging( toRemove.push(name); continue; } - const tracked = await runGit(dir, ["ls-files", "--", name]); - if (tracked.length === 0) { + if (!tracked.has(name)) { if (index > highestDisk) break; continue; } @@ -430,19 +379,27 @@ async function reconcileSegmentStaging( } } +function extraCommitPaths(paths: readonly string[]): string[] { + return paths.filter((filepath) => !VENDOR_COMMIT_ROOT_FILES.has(filepath)); +} + +export type SessionStores = { + storage: ContextStore; + audit: AuditStore; +}; + /** * Local wrapper around the Interchange git store that avoids O(session length) * work per reactor checkpoint. Turns and prompt snapshots are written as rolling - * segment files so `git add` re-hashes only the small active segment, and only - * spilled tool-output blobs that are new since the last commit are staged. + * segment files so only the small active extra segment is re-hashed, then + * `base.commit()` takes the vendor lock, durable commit, signing, and GC. */ -export async function createOptimizedContextStore( +export async function createSessionStores( dir: string, - opts?: { author?: CheckpointAuthor; env?: NodeJS.ProcessEnv }, -): Promise { - const gitEnv = opts?.env ?? process.env; - const author = opts?.author ?? (await resolveCheckpointAuthor(gitEnv)); - const base = await createIsogitStore(dir); + opts?: { signer?: CommitSigner }, +): Promise { + const signer = opts?.signer ?? (await loadOrCreateCommitSigner(dir)); + const base = await createIsogitStore(dir, signer); const pendingBlobFilepaths = new Set(); const pendingSegmentPaths = new Set(); const writeTurnsSegmented = createSegmentedJSONLWriter(dir, TURNS_FILE); @@ -488,7 +445,7 @@ export async function createOptimizedContextStore( return [...baseTurns, ...parsedExtras.slice(0, keepExtras).flat()]; } - return { + const store: ContextStore & AuditStore = { // Full-history read. Called by the reactor during initialization, where // the complete turn history is the actual live conversation state, not an // optional convenience — callers that only need a recent tail (e.g. TUI @@ -551,7 +508,7 @@ export async function createOptimizedContextStore( // malformed. Prefer the longest well-formed prefix; no on-disk side effects. const parsedExtras: ConversationTurn[][] = []; for (const name of extraNames) { - const text = await runGit(dir, ["show", `${hash}:${name}`]); + const text = await blobTextAtCommit(dir, hash, name); parsedExtras.push(parseSegmentTurns(text, false, name)); } const keepExtras = longestWellFormedExtraCount(baseTurns, parsedExtras); @@ -570,41 +527,61 @@ export async function createOptimizedContextStore( const filename = `${sanitizeCallId(key)}${blobExtensionFor(contentType)}`; pendingBlobFilepaths.add(`${TOOL_OUTPUT_DIR}/${filename}`); }, - async commit(options, _signal) { - const toAdd: string[] = []; - const toRemove: string[] = []; - - const rewrittenEachCycle = [RESPONSE_FILE, MANIFEST_FILE, METADATA_FILE]; - for (const filepath of rewrittenEachCycle) { - if (await pathExists(path.join(dir, filepath))) toAdd.push(filepath); - } - - for (const filepath of [...pendingSegmentPaths, ...pendingBlobFilepaths]) { - if (await pathExists(path.join(dir, filepath))) toAdd.push(filepath); - else toRemove.push(filepath); - } + async commit(options, signal) { + return withResolvedDirLock(dir, async () => { + const toAdd: string[] = []; + const toRemove: string[] = []; + + for (const filepath of [...pendingSegmentPaths, ...pendingBlobFilepaths]) { + if (await pathExists(path.join(dir, filepath))) toAdd.push(filepath); + else toRemove.push(filepath); + } - // Disk is source of truth for which turn/prompt segments should remain - // tracked after a rewrite or heal, even if pendingSegmentPaths was lost. - await reconcileSegmentStaging(dir, TURNS_FILE, toAdd, toRemove); - await reconcileSegmentStaging(dir, PROMPT_FILE, toAdd, toRemove); + // Disk is source of truth for which turn/prompt segments should remain + // tracked after a rewrite or heal, even if pendingSegmentPaths was lost. + await reconcileSegmentStaging(dir, TURNS_FILE, toAdd, toRemove); + await reconcileSegmentStaging(dir, PROMPT_FILE, toAdd, toRemove); - const add = [...new Set(toAdd)]; - const remove = [...new Set(toRemove)].filter((p) => !add.includes(p)); + const add = extraCommitPaths([...new Set(toAdd)]); + const remove = extraCommitPaths([...new Set(toRemove)]).filter((p) => !add.includes(p)); + const extraPaths = [...new Set([...add, ...remove])]; - if (add.length > 0) await runGit(dir, ["add", "--", ...add]); - if (remove.length > 0) { - await runGit(dir, ["rm", "--cached", "--ignore-unmatch", "--", ...remove]); - } - await runGit( - dir, - ["commit", "-m", options.message, `--author=${author.name} <${author.email}>`], - author, - gitEnv, - ); - pendingBlobFilepaths.clear(); - pendingSegmentPaths.clear(); - return describeHead(dir, options.message); + try { + for (const filepath of add) { + await git.add({ fs, dir, filepath }); + } + for (const filepath of remove) { + try { + await git.remove({ fs, dir, filepath }); + } catch { + // Already absent from the index. + } + } + const committed = await base.commit(options, signal); + pendingBlobFilepaths.clear(); + pendingSegmentPaths.clear(); + return committed; + } catch (cause) { + await resetIndexPaths(dir, extraPaths); + throw cause; + } + }); }, + commitAudit: (records, signal) => + withResolvedDirLock(dir, () => base.commitAudit(records, signal)), + commitErrors: (records, signal) => + withResolvedDirLock(dir, () => base.commitErrors(records, signal)), + loadAudit: (sessionId, signal) => base.loadAudit(sessionId, signal), }; + + return { storage: store, audit: store }; +} + +export async function createOptimizedContextStore( + dir: string, + opts?: { signer?: CommitSigner }, +): Promise { + const { storage } = await createSessionStores(dir, opts); + return storage; } + diff --git a/src/session/session-dir-lock.ts b/src/session/session-dir-lock.ts new file mode 100644 index 000000000..9bca81150 --- /dev/null +++ b/src/session/session-dir-lock.ts @@ -0,0 +1,31 @@ +import path from "node:path"; + +const locks = new Map>(); + +/** + * Process-wide mutex keyed by resolved directory. Wrapper staging and + * `base.commit()` / audit writes must not interleave on the same repo. + */ +export async function withResolvedDirLock(dir: string, fn: () => Promise): Promise { + const key = path.resolve(dir); + const previous = locks.get(key) ?? Promise.resolve(); + let release!: () => void; + const current = new Promise((resolve) => { + release = resolve; + }); + const tail = previous.then( + () => current, + () => current, + ); + locks.set(key, tail); + try { + await previous.then( + () => undefined, + () => undefined, + ); + return await fn(); + } finally { + release(); + if (locks.get(key) === tail) locks.delete(key); + } +} From 26ac5b4eee52a980c2e9579fd7db0ae89d0f4898 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 8 Sep 2026 22:28:48 -0700 Subject: [PATCH 2/7] Thread isogit audit methods into agent assembly --- docs/IMPLEMENTATION.md | 11 ++-- src/exec/runner.ts | 1 + src/session/assemble-runtime.test.ts | 37 +++++++++++--- src/session/assemble-runtime.ts | 10 ++-- src/subagent/run-audit-store.test.ts | 76 ++++++++++++++++++++++++++++ src/subagent/run.ts | 11 ++-- src/tui/runner/session.ts | 1 + 7 files changed, 126 insertions(+), 21 deletions(-) create mode 100644 src/subagent/run-audit-store.test.ts diff --git a/docs/IMPLEMENTATION.md b/docs/IMPLEMENTATION.md index 3393e7911..580f87269 100644 --- a/docs/IMPLEMENTATION.md +++ b/docs/IMPLEMENTATION.md @@ -485,13 +485,14 @@ Corbits Code v0.3 memory and stall hardening is implemented under `src/`, `tests ### Bounded audit collector retention between checkpoints -Agent-owned audit collectors buffer completed tool results until checkpoint or -shutdown flush, including when a noop store is supplied. Workers use the durable -store described under State Persistence; the parent's noop store does not make -collector retention inapplicable. Long, checkpoint-sparse runs can retain +Production chat and sub-agent assembly persist audit via the same isogit +object as context storage (`createSessionStores`), plus a stable `sessionId`. +Agent-owned audit collectors still buffer completed tool results until +checkpoint or shutdown flush. Long, checkpoint-sparse runs can retain unbounded results. Bounded retention remains owned by the `@intx/inference` audit collector: opportunistic flushing or capped result bodies must preserve -metadata. +metadata. If a collector is introduced in front of the isogit audit methods, +add a bounded wrapper in `src/` and re-run hardening tests. Other wave items (read bounds, shell truncation, process-group kill, grep caps, plugin spawn mitigation, per-tool watchdog, inference retry UX) are implemented or partially mitigated in `src/` with co-located tests; the two items above remain upstream-owned. diff --git a/src/exec/runner.ts b/src/exec/runner.ts index 5a8f55ae5..889bfa413 100644 --- a/src/exec/runner.ts +++ b/src/exec/runner.ts @@ -615,6 +615,7 @@ export async function runExec(config: Config): Promise { }, getProvider: () => config, getWorkdir: () => workdir, + getSessionId: () => sessionId, authorize: createReactorAuthorize(permissionGate), inferenceDeps, getSources: () => { diff --git a/src/session/assemble-runtime.test.ts b/src/session/assemble-runtime.test.ts index c4c7b8716..35bfe3c08 100644 --- a/src/session/assemble-runtime.test.ts +++ b/src/session/assemble-runtime.test.ts @@ -3,7 +3,7 @@ import { mkdtemp } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import type { Agent } from "@intx/agent"; -import type { Compactor, ContextStore, ToolDefinition } from "@intx/types/runtime"; +import type { AuditStore, Compactor, ContextStore, ToolDefinition } from "@intx/types/runtime"; import { withMockedModuleDuring } from "../../tests/helpers/mock-module.js"; import { @@ -109,22 +109,25 @@ function stubInferenceDeps(): ChatAgentWiring["inferenceDeps"] { } describe("assembleChatAgent", () => { - test("getWorkdir and getCompactor run at buildAgent time, not assemble time", async () => { + test("getWorkdir, getSessionId, and getCompactor run at buildAgent time", async () => { const storeDirs: string[] = []; const agentWorkdirs: string[] = []; + const agentSessionIds: string[] = []; + const agentAudits: AuditStore[] = []; + const agentStorages: ContextStore[] = []; const agentCompactors: Compactor[] = []; const fakeStorage = { readBlob: async () => new Uint8Array(), - } as unknown as ContextStore; + } as unknown as ContextStore & AuditStore; const fakeAgent = { close: async () => {} } as unknown as Agent; await withMockedModuleDuring( import.meta.resolve("./optimized-context-store.js"), (real: typeof import("./optimized-context-store.js")) => ({ ...real, - createOptimizedContextStore: async (dir: string) => { + createSessionStores: async (dir: string) => { storeDirs.push(dir); - return fakeStorage; + return { storage: fakeStorage, audit: fakeStorage }; }, }), async () => { @@ -134,9 +137,18 @@ describe("assembleChatAgent", () => { ...real, createAgentWithLiveToolDispatch: async ( _def: unknown, - env: { workdir: string; compactors: { "pruning-compactor": Compactor } }, + env: { + workdir: string; + sessionId?: string; + storage: ContextStore; + audit: AuditStore; + compactors: { "pruning-compactor": Compactor }; + }, ) => { agentWorkdirs.push(env.workdir); + if (env.sessionId !== undefined) agentSessionIds.push(env.sessionId); + agentStorages.push(env.storage); + agentAudits.push(env.audit); agentCompactors.push(env.compactors["pruning-compactor"]); return fakeAgent; }, @@ -144,8 +156,10 @@ describe("assembleChatAgent", () => { async () => { const { assembleChatAgent } = await import("./assemble-runtime.js"); const workdirCalls: string[] = []; + const sessionIdCalls: string[] = []; const compactorCalls: string[] = []; let liveDir = "/assemble-dir"; + let liveSessionId = "assemble-session"; let liveCompactor = stubCompactor("assemble"); const { buildAgent } = assembleChatAgent({ @@ -166,6 +180,10 @@ describe("assembleChatAgent", () => { workdirCalls.push(liveDir); return liveDir; }, + getSessionId: () => { + sessionIdCalls.push(liveSessionId); + return liveSessionId; + }, inferenceDeps: stubInferenceDeps(), getSources: () => [ { @@ -185,20 +203,27 @@ describe("assembleChatAgent", () => { }); expect(workdirCalls).toEqual([]); + expect(sessionIdCalls).toEqual([]); expect(compactorCalls).toEqual([]); expect(storeDirs).toEqual([]); expect(agentWorkdirs).toEqual([]); liveDir = "/build-dir"; + liveSessionId = "build-session"; liveCompactor = stubCompactor("build"); const builtCompactor = liveCompactor; await buildAgent(); expect(workdirCalls).toEqual(["/build-dir"]); + expect(sessionIdCalls).toEqual(["build-session"]); expect(compactorCalls).toEqual(["build"]); expect(storeDirs).toEqual(["/build-dir"]); expect(agentWorkdirs).toEqual(["/build-dir"]); + expect(agentSessionIds).toEqual(["build-session"]); + expect(agentStorages).toEqual([fakeStorage]); + expect(agentAudits).toEqual([fakeStorage]); + expect(agentAudits[0]).toBe(agentStorages[0]); expect(agentCompactors).toEqual([builtCompactor]); }, ); diff --git a/src/session/assemble-runtime.ts b/src/session/assemble-runtime.ts index 5cd9167f9..4bbf62906 100644 --- a/src/session/assemble-runtime.ts +++ b/src/session/assemble-runtime.ts @@ -22,7 +22,6 @@ import { type Agent, type AuthorizeFn, } from "@intx/agent"; -import { noopAuditStore } from "@intx/agent/testing"; import type { Compactor, ContextStore, InferenceSource, ToolDefinition } from "@intx/types/runtime"; import { type } from "arktype"; @@ -48,7 +47,7 @@ import { createChatDirector, type ChatDirector } from "../agent/director.js"; import type { Task } from "../agent/tasks.js"; import type { AgentToolset } from "../agent/tools.js"; import { createAgentWithLiveToolDispatch } from "../agent/live-tool-dispatch.js"; -import { createOptimizedContextStore } from "./optimized-context-store.js"; +import { createSessionStores } from "./optimized-context-store.js"; import { createAttachmentRehydrateTransform } from "./attachment-store.js"; import { loadProjectTrust, @@ -353,6 +352,8 @@ export interface ChatAgentWiring { authorize: AuthorizeFn; /** Read at each build so /clear and workdir rotation use the live store path. */ getWorkdir: () => string; + /** Read at each build so /clear and session rotation stamp the live session id. */ + getSessionId: () => string; inferenceDeps: Awaited>; getSources: () => InferenceSource[]; getDefaultSource: () => string; @@ -425,7 +426,7 @@ export function assembleChatAgent(wiring: ChatAgentWiring): AssembledChatAgent { const buildAgent = async (): Promise => { const workdir = wiring.getWorkdir(); - const storage = await createOptimizedContextStore(workdir); + const { storage, audit } = await createSessionStores(workdir); const agent = await createAgentWithLiveToolDispatch(agentDef, { sources: wiring.getSources(), defaultSource: wiring.getDefaultSource(), @@ -438,7 +439,8 @@ export function assembleChatAgent(wiring: ChatAgentWiring): AssembledChatAgent { ...wiring.inferenceDeps, contextTransforms: [createAttachmentRehydrateTransform((key) => storage.readBlob(key))], }, - audit: noopAuditStore(), + audit, + sessionId: wiring.getSessionId(), // Gate-backed reactor authorization: ask-tier calls suspend via the // vendored approval-suspend primitive instead of parking on a closure. authorize: wiring.authorize, diff --git a/src/subagent/run-audit-store.test.ts b/src/subagent/run-audit-store.test.ts new file mode 100644 index 000000000..6f3604669 --- /dev/null +++ b/src/subagent/run-audit-store.test.ts @@ -0,0 +1,76 @@ +import { expect, test } from "bun:test"; +import { mkdtemp } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { AuditStore, ContextStore } from "@intx/types/runtime"; + +import { withMockedModuleDuring } from "../../tests/helpers/mock-module.js"; +import { createPermissionGate } from "../permission/gate.js"; + +const permissionGate = createPermissionGate({ + approvals: [], + interactive: false, + skipPermissions: true, + reactorGated: false, +}); + +test("runSubAgent threads the isogit audit store and session id into createAgent", async () => { + const cwd = await mkdtemp(join(tmpdir(), "corbits-run-audit-")); + const fakeStore = { readBlob: async () => new Uint8Array() } as unknown as ContextStore & AuditStore; + let seen: { audit: AuditStore; sessionId?: string; storage: ContextStore } | undefined; + + await withMockedModuleDuring( + import.meta.resolve("../session/optimized-context-store.js"), + (real: typeof import("../session/optimized-context-store.js")) => ({ + ...real, + createSessionStores: async () => ({ storage: fakeStore, audit: fakeStore }), + }), + async () => { + await withMockedModuleDuring( + import.meta.resolve("../agent/live-tool-dispatch.js"), + (real: typeof import("../agent/live-tool-dispatch.js")) => ({ + ...real, + createAgentWithLiveToolDispatch: async ( + _def: unknown, + env: { storage: ContextStore; audit: AuditStore; sessionId?: string }, + ) => { + seen = env; + return { + send: async () => ({ + type: "reply" as const, + reply: "ok", + turn: { role: "assistant" as const, content: [] }, + }), + stream: () => (async function* () {})(), + deliver: () => {}, + close: async () => {}, + setSource: () => {}, + setSources: () => {}, + history: async () => [], + checkpoints: async () => [], + readAt: async () => [], + blobReader: {}, + }; + }, + }), + async () => { + const { runSubAgent } = await import("./run.js"); + await runSubAgent({ + cwd, + workdirBase: join(cwd, ".ctx"), + permissionGate, + provider: { providerName: "test", baseURL: "http://localhost", model: "test-model" }, + description: "audit wiring", + prompt: "noop", + id: "child-session-1", + }); + }, + ); + }, + ); + + expect(seen).toBeDefined(); + expect(seen!.storage).toBe(fakeStore); + expect(seen!.audit).toBe(fakeStore); + expect(seen!.sessionId).toBe("child-session-1"); +}); diff --git a/src/subagent/run.ts b/src/subagent/run.ts index 1a41a7737..426660387 100644 --- a/src/subagent/run.ts +++ b/src/subagent/run.ts @@ -17,9 +17,8 @@ import { type SendResult, } from "@intx/agent"; import type { AgentTool } from "@intx/agent"; -import { createIsogitStore } from "@intx/storage-isogit/node"; import { createWorkerAuthorize, workerPermissionGate } from "../permission/reactor-authorize.js"; -import { createOptimizedContextStore } from "../session/optimized-context-store.js"; +import { createSessionStores } from "../session/optimized-context-store.js"; import { createAgentWithLiveToolDispatch } from "../agent/live-tool-dispatch.js"; import { type } from "arktype"; import { createPosixTools } from "@intx/tools-posix"; @@ -873,7 +872,8 @@ async function runSubAgentInner( // scheme. const safeRequestedId = params.id !== undefined && /^[A-Za-z0-9_-]+$/.test(params.id) ? params.id : undefined; - const workdir = join(params.workdirBase, "subagents", safeRequestedId ?? generateSessionId()); + const sessionId = safeRequestedId ?? generateSessionId(); + const workdir = join(params.workdirBase, "subagents", sessionId); await mkdir(workdir, { recursive: true }); // One record per stop/nudge, with its measured value beside its // threshold, written into this leaf's own trace dir. @@ -896,9 +896,7 @@ async function runSubAgentInner( }, }); - const storage = await createOptimizedContextStore(workdir); - // Audit commits must not race the native context store's git index. - const audit = await createIsogitStore(join(workdir, "audit-store")); + const { storage, audit } = await createSessionStores(workdir); const authorize = createWorkerAuthorize(params.permissionGate); const head = { provider: params.provider.providerName, model: params.provider.model }; @@ -928,6 +926,7 @@ async function runSubAgentInner( contextTransforms: [createAttachmentRehydrateTransform((key) => storage.readBlob(key))], }, audit, + sessionId, authorize: (resource, action, context) => withWorkerIdentity(() => authorize(resource, action, context)), directors: createDirectorRegistry({ diff --git a/src/tui/runner/session.ts b/src/tui/runner/session.ts index 05a19ac98..7ee026c70 100644 --- a/src/tui/runner/session.ts +++ b/src/tui/runner/session.ts @@ -471,6 +471,7 @@ export async function assembleTUISession( state.reloadIfIdle?.(); }, getWorkdir: () => state.workdir, + getSessionId: () => state.sessionId, authorize: createReactorAuthorize(permissionGate), inferenceDeps: start.inferenceDeps, getSources: () => (state.liveSources.length > 0 ? state.liveSources : [state.liveSource]), From 58534b69ac733b64a47bc09a2955ff7c6ee73e44 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Wed, 9 Sep 2026 08:44:36 -0700 Subject: [PATCH 3/7] Move workflow lifecycle out of the terminal UI --- docs/ARCHITECTURE.md | 2 +- scripts/guard-real-projects-dir.ts | 2 +- src/exec/runner.ts | 16 ++- src/tui/runner/commands.ts | 4 +- src/tui/runner/exit.ts | 6 +- src/tui/runner/session.ts | 30 ++-- src/tui/runner/state.ts | 3 +- src/tui/runner/wiring.ts | 2 +- .../host.ts} | 67 ++++----- ...ntroller.test.ts => workflow-host.test.ts} | 136 ++++++++++-------- 10 files changed, 143 insertions(+), 125 deletions(-) rename src/{tui/workflow-controller.ts => workflows/host.ts} (82%) rename tests/unit/{workflow-controller.test.ts => workflow-host.test.ts} (54%) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 354600910..364ce82ce 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -198,7 +198,7 @@ Workflows are named, ordered recipes the agent follows step by step — a thin l - `coordinator.ts` — bridges runtime and director: produces the `[WORKFLOW STEP i/total: label]` directive injected into each turn's system prompt, and compare-and-advances the runtime when a `submit_output` tagged `{ step }` completes. Already-complete and not-current ids are acknowledged without moving the cursor. Shared by both directors. Fresh and resumed runs share one listener path. - The built-in recipes: the atomics `update-ticket`, `improve-docs`, `write-tests`, `triage-bug`, `code-review`, `scope-project`, and the `build-feature` composite that chains them. -Invocation: workflows are **not** top-level slash commands. Recipe definitions load into the `WORKFLOWS` registry from **enabled workflow/command plugins** at startup; command surfaces on those plugins (e.g. a workflow plugin's command prefix such as `/mywf scope`). Slash commands may also be authored as data-only markdown (`commands/*.md`, no `index.ts`); see PLUGINS.md. The model never suggests or auto-starts workflows from ordinary chat. Skills (bundled `corbits-skills`, enabled plugins, or `.agents/skills/`) load on demand via `skill_search` then `use_skill`, or as `/` slash commands when `user-invocable` is not `false` (see Skills below). The TUI surfaces state via `src/tui/workflow-controller.ts` (lifecycle, capability overrides, resume) — the header shows step progress (`⟳ name · step/total label`). +Invocation: workflows are **not** top-level slash commands. Recipe definitions load into the `WORKFLOWS` registry from **enabled workflow/command plugins** at startup; command surfaces on those plugins (e.g. a workflow plugin's command prefix such as `/mywf scope`). Slash commands may also be authored as data-only markdown (`commands/*.md`, no `index.ts`); see PLUGINS.md. The model never suggests or auto-starts workflows from ordinary chat. Skills (bundled `corbits-skills`, enabled plugins, or `.agents/skills/`) load on demand via `skill_search` then `use_skill`, or as `/` slash commands when `user-invocable` is not `false` (see Skills below). `WorkflowHost` (`src/workflows/host.ts`) owns lifecycle, capability overrides, and resume; the TUI only renders host status in the header (`⟳ name · step/total label`). ### Fleet agents (`src/subagent/`, `src/agent/agent-search.ts`) diff --git a/scripts/guard-real-projects-dir.ts b/scripts/guard-real-projects-dir.ts index 9d0b316d9..177bfcdce 100644 --- a/scripts/guard-real-projects-dir.ts +++ b/scripts/guard-real-projects-dir.ts @@ -86,7 +86,7 @@ async function main(): Promise { leaked.map((name) => ` ${name}`).join("\n") + "\n\nA test must pass an explicit `home` (mkdtemp'd) through to any " + "function that otherwise defaults to node:os homedir() — see " + - "tests/unit/workflow-controller.test.ts for the pattern.\n", + "tests/unit/workflow-host.test.ts for the pattern.\n", ); process.exit(1); } diff --git a/src/exec/runner.ts b/src/exec/runner.ts index 889bfa413..bedc98eb1 100644 --- a/src/exec/runner.ts +++ b/src/exec/runner.ts @@ -75,6 +75,7 @@ import { ID_PREFIX, LOG_NAMESPACE_ROOT } from "../branding.js"; import type { ReactorEmittedEvent } from "@intx/inference"; import { setAgentSourceUnlessClosed } from "../tui/agent-source-sync.js"; import { getToolApprovalBudget } from "../tui/tool-execution-watchdog.js"; +import { WorkflowHost } from "../workflows/host.js"; const logger = getLogger([LOG_NAMESPACE_ROOT, "exec"]); @@ -472,6 +473,7 @@ export async function runExec(config: Config): Promise { let currentStorage: ContextStore | null = null; const overlay = resolveExecDirectorOverlay(config.director); + const workflowHostHolder: { instance?: WorkflowHost } = {}; const agentToolset = await createAgentToolset({ cwd: config.cwd, @@ -494,8 +496,9 @@ export async function runExec(config: Config): Promise { } return currentAgent.blobReader; }, - // Exec has no workflow controller — intentional delta vs TUI. - isWorkflowActive: () => false, + isWorkflowActive: () => workflowHostHolder.instance?.isActive() === true, + completeWorkflowStep: (stepId) => + workflowHostHolder.instance?.complete(stepId) ?? "not-current", onOperatorGate: (question, options) => promptOperator(question, options, interactive), sessionMode, toolAvailability, @@ -638,6 +641,14 @@ export async function runExec(config: Config): Promise { }, }); + const workflowHost = new WorkflowHost({ + cwd: config.cwd, + getSessionId: () => sessionId, + getToolDefinitions: () => agentToolset.dynamicRunner.currentDefinitions(), + getDirector: () => directorHolder.instance, + }); + workflowHostHolder.instance = workflowHost; + const emitter = new EventEmitter(); const { hookManager, @@ -682,6 +693,7 @@ export async function runExec(config: Config): Promise { }); }); } + await workflowHost.resume(); const textChunks: string[] = []; // Cycles persist to the context store only on inference.done; the recorder diff --git a/src/tui/runner/commands.ts b/src/tui/runner/commands.ts index 78b4bd7a1..565ff9783 100644 --- a/src/tui/runner/commands.ts +++ b/src/tui/runner/commands.ts @@ -139,7 +139,7 @@ export function createCommandLayer(state: RunnerState, services: RunnerServices) }); return maskContextMeterWhenNoTurns(summary, services.runSink.getTurnCount()); }, - startWorkflow: (name) => services.workflowController.start(name), + startWorkflow: (name) => services.workflowHost.start(name), getFleetStatus: () => fleetDigest(services.subAgentSessions.list(), Date.now()), renameSession: (name) => { const trimmed = name.trim(); @@ -181,7 +181,7 @@ export function createCommandLayer(state: RunnerState, services: RunnerServices) void state.sendWithAttemptIdentity?.(userInboundMessage(result.text, [])); return; case "workflow": - state.systemNotice?.(services.workflowController.start(result.name)); + state.systemNotice?.(services.workflowHost.start(result.name)); return; case "noop": return; diff --git a/src/tui/runner/exit.ts b/src/tui/runner/exit.ts index e72e080a7..7797b7c8c 100644 --- a/src/tui/runner/exit.ts +++ b/src/tui/runner/exit.ts @@ -284,7 +284,7 @@ export async function createRunLifecycle( state.currentAgent = await services.buildAgent(); state.streamPromise = consumeStream(liveAgent(state).stream(), streamSink); // The rebuild made a fresh director; re-attach the active workflow. - services.workflowController.reattach(); + services.workflowHost.reattach(); } catch (err) { recordRunError(state, err); state.fatalBuildError = agentRebuildFailure(err); @@ -453,7 +453,7 @@ export async function createRunLifecycle( state.currentAgent = await services.buildAgent(); services.cycleRecorder.reset(); state.streamPromise = consumeStream(liveAgent(state).stream(), streamSink); - services.workflowController.reattach(); + services.workflowHost.reattach(); state.fatalBuildError = null; } catch (err) { recordRunError(state, err); @@ -533,7 +533,7 @@ export async function createRunLifecycle( state.streamPromise = consumeStream(liveAgent(state).stream(), streamSink); await persistRunSnapshot("running"); // A fresh session drops any active workflow. - services.workflowController.reset(); + services.workflowHost.reset(); state.fatalBuildError = null; // Sink and director are empty now — repaint so the meter stays hidden // rather than showing the pre-clear occupancy until the next turn. diff --git a/src/tui/runner/session.ts b/src/tui/runner/session.ts index 7ee026c70..0bc60f1a2 100644 --- a/src/tui/runner/session.ts +++ b/src/tui/runner/session.ts @@ -2,7 +2,7 @@ * Session assembly for the TUI runner: everything the old runTUI closure * built once inside its try block before the first agent build — the session * lifecycle (hooks sink, run sink, cycle recorder), the permission gate, - * plugin/tool resolution, the agent toolset, the workflow controller, and + * plugin/tool resolution, the agent toolset, the workflow host, and * the chat agent factory. Returns the const `RunnerServices` bag index.ts * threads through the other runner modules; mutable bindings live on * RunnerState. @@ -65,7 +65,7 @@ import { createAgentToolset, type MCPServerState, type OperatorResult } from ".. import type { ToolAvailability } from "../../agent/tool-search.js"; import { detectLanguageServerAvailable } from "../../agent/lsp-availability.js"; import type { SessionMode } from "../../config/session-mode.js"; -import { WorkflowController } from "../workflow-controller.js"; +import { WorkflowHost, type WorkflowHostState } from "../../workflows/host.js"; import type { ToolWatchdogConfig } from "../tool-execution-watchdog.js"; import { deliverAgentMessage } from "../deliver-agent-message.js"; import { createProviderFailureAttemptTracker } from "../provider/failure-attempt.js"; @@ -251,10 +251,10 @@ export async function assembleTUISession( const toolAvailability: ToolAvailability = { languageServerAvailable: detectLanguageServerAvailable(config.cwd), }; - // The workflow controller is built below, after the toolset; the holder lets + // The workflow host is built below, after the toolset; the holder lets // submit_output's handler complete the live workflow without a // construction-order cycle. - const workflowControllerHolder: { instance?: WorkflowController } = {}; + const workflowHostHolder: { instance?: WorkflowHost } = {}; const toolset = await createAgentToolset({ cwd: config.cwd, @@ -275,9 +275,9 @@ export async function assembleTUISession( liveAgent(state).deliver(buildShellBackgroundMessage(exit)), ); }, - isWorkflowActive: () => workflowControllerHolder.instance?.isActive() === true, + isWorkflowActive: () => workflowHostHolder.instance?.isActive() === true, completeWorkflowStep: (stepId) => - workflowControllerHolder.instance?.complete(stepId) ?? "not-current", + workflowHostHolder.instance?.complete(stepId) ?? "not-current", ...(extraToolPlugins.length > 0 ? { extraToolPlugins } : {}), onOperatorGate: (question, options) => new Promise((resolve) => { @@ -359,15 +359,20 @@ export async function assembleTUISession( const hostHolder: { instance?: RunnerHost } = {}; // Owns the workflow lifecycle: slash-command starts, capability overrides, - // resume, and publishing status to the App via the emitter. - const workflowController = new WorkflowController({ + // resume, and publishing status via the emitter. The TUI only renders. + const workflowHost = new WorkflowHost({ cwd: config.cwd, - emitter, getSessionId: () => state.sessionId, getToolDefinitions: () => toolset.dynamicRunner.currentDefinitions(), getDirector: () => directorHolder.instance, + onChange: () => { + emitter.emit("workflow", { + current: workflowHost.status(), + history: workflowHost.history(), + } satisfies WorkflowHostState); + }, }); - workflowControllerHolder.instance = workflowController; + workflowHostHolder.instance = workflowHost; // Dynamic tool discovery: only the fixed built-in prefix plus activated // tools reach the wire, so the provider cache prefix holds steady; MCP @@ -433,7 +438,7 @@ export async function assembleTUISession( deps: start.inferenceDeps, }); const summaryContext = (): SummaryContext | undefined => { - const status = workflowController.status(); + const status = workflowHost.status(); if (!status.active) return undefined; return { workflow: { @@ -536,8 +541,7 @@ export async function assembleTUISession( systemPrompt, directorHolder, hostHolder, - workflowControllerHolder, - workflowController, + workflowHost, activatedToolNames, computeAdvertised, buildAgent: chatAgent.buildAgent, diff --git a/src/tui/runner/state.ts b/src/tui/runner/state.ts index 662d62d7b..1c8164a32 100644 --- a/src/tui/runner/state.ts +++ b/src/tui/runner/state.ts @@ -107,8 +107,7 @@ export interface RunnerServices { instance?: ReturnType; }; hostHolder: { instance?: RunnerHost }; - workflowControllerHolder: { instance?: import("../workflow-controller.js").WorkflowController }; - workflowController: import("../workflow-controller.js").WorkflowController; + workflowHost: import("../../workflows/host.js").WorkflowHost; activatedToolNames: Awaited< ReturnType >["activated"]; diff --git a/src/tui/runner/wiring.ts b/src/tui/runner/wiring.ts index dc7f48d8a..ade4e9920 100644 --- a/src/tui/runner/wiring.ts +++ b/src/tui/runner/wiring.ts @@ -290,7 +290,7 @@ export function wirePostStartup( } // Now that the capability map reflects connected MCP servers, restore any // persisted workflow. New workflows are manual-only slash commands. - await services.workflowController.resume(); + await services.workflowHost.resume(); }) .catch((err: unknown) => { // Fire-and-forget: an aborted connect on exit is expected and ignored; diff --git a/src/tui/workflow-controller.ts b/src/workflows/host.ts similarity index 82% rename from src/tui/workflow-controller.ts rename to src/workflows/host.ts index 8e3f0bbeb..4ddce7007 100644 --- a/src/tui/workflow-controller.ts +++ b/src/workflows/host.ts @@ -1,24 +1,13 @@ -import type { EventEmitter } from "node:events"; import { join } from "node:path"; import type { ToolDefinition } from "@intx/types/runtime"; import { sessionDir } from "../session/index.js"; -import { CAPABILITIES, detectCapabilities, type CapabilityMap } from "../workflows/capabilities.js"; -import { WorkflowCoordinator } from "../workflows/coordinator.js"; -import { findWorkflow, WORKFLOWS } from "../workflows/index.js"; -import { WorkflowRuntime } from "../workflows/runtime.js"; -import { - loadWorkflowState, - saveWorkflowState, - warnWorkflowPersistenceFailure, -} from "../workflows/state.js"; -import type { - CapabilityName, - StepStatus, - Workflow, - WorkflowCompleteResult, -} from "../workflows/types.js"; -import type { WorkflowEvent } from "../workflows/runtime.js"; +import { CAPABILITIES, detectCapabilities, type CapabilityMap } from "./capabilities.js"; +import { WorkflowCoordinator } from "./coordinator.js"; +import { findWorkflow, WORKFLOWS } from "./index.js"; +import { WorkflowRuntime, type WorkflowEvent } from "./runtime.js"; +import { loadWorkflowState, saveWorkflowState, warnWorkflowPersistenceFailure } from "./state.js"; +import type { CapabilityName, StepStatus, Workflow, WorkflowCompleteResult } from "./types.js"; export interface CapabilityStatus { name: CapabilityName; @@ -45,16 +34,15 @@ export interface WorkflowStatus { completedAt?: number; } -export interface WorkflowControllerState { +export interface WorkflowHostState { current: WorkflowStatus; history: WorkflowStatus[]; } type SetCoordinator = (coordinator: WorkflowCoordinator | undefined) => void; -export interface WorkflowControllerArgs { +export interface WorkflowHostArgs { cwd: string; - emitter: EventEmitter; getSessionId: () => string; getToolDefinitions: () => ToolDefinition[]; // The live chat director; the workflow coordinator is attached to it when a @@ -63,12 +51,12 @@ export interface WorkflowControllerArgs { // Overrides the state-tree home (defaults to the real user home). Tests // pass a sandboxed dir here so persist()/resume() never touch ~/.corbits. home?: string; + onChange?: () => void; } -// Owns the workflow lifecycle for the TUI: starting, capability overrides, -// resume, and publishing status to the UI via the "workflow" emitter event. -// Framework-agnostic so it can be unit-tested without React. -export class WorkflowController { +// Owns workflow lifecycle: starting, capability overrides, resume, and +// persisting state. UI layers subscribe via onChange and render status(). +export class WorkflowHost { private runtime: WorkflowRuntime | undefined; private coordinator: WorkflowCoordinator | undefined; private overrides = new Set(); @@ -78,16 +66,15 @@ export class WorkflowController { // history on workflow-complete, where isActive() is already false. private lastActiveStatus: WorkflowStatus | undefined; - constructor(private readonly args: WorkflowControllerArgs) {} + constructor(private readonly args: WorkflowHostArgs) {} - // Re-attach the active coordinator to a freshly rebuilt director (the TUI - // rebuilds the agent when MCP servers connect). Safe to call with no active - // workflow — it just clears any stale coordinator. + // Re-attach the active coordinator to a freshly rebuilt director. Safe to + // call with no active workflow — it just clears any stale coordinator. reattach(): void { this.args.getDirector()?.setWorkflowCoordinator(this.coordinator); } - // Drop the active workflow and history (e.g. on /clear, which starts a fresh session). + // Drop the active workflow and history (e.g. on /clear). reset(): void { this.runtime = undefined; this.coordinator = undefined; @@ -95,7 +82,7 @@ export class WorkflowController { this.completedWorkflows = []; this.lastActiveStatus = undefined; this.args.getDirector()?.setWorkflowCoordinator(undefined); - this.publish(); + this.notify(); } history(): WorkflowStatus[] { @@ -118,10 +105,10 @@ export class WorkflowController { return WORKFLOWS.map((w) => ({ name: w.name, description: w.description })); } - private publish(): void { + private notify(): void { const current = this.status(); if (current.active) this.lastActiveStatus = current; - this.args.emitter.emit("workflow", { current, history: this.completedWorkflows }); + this.args.onChange?.(); } private persist(): void { @@ -145,7 +132,7 @@ export class WorkflowController { runtime, () => { this.persist(); - this.publish(); + this.notify(); }, workflow.stepThrough === true, ); @@ -156,7 +143,7 @@ export class WorkflowController { if (restore !== true) { runtime.start(workflow); this.persist(); - this.publish(); + this.notify(); } return runtime; } @@ -175,14 +162,10 @@ export class WorkflowController { } } this.persist(); - this.publish(); + this.notify(); }); } - private attach(workflow: Workflow): void { - this.attachRuntime(workflow); - } - // Start a workflow by name. If one is already active, the first call asks for // confirmation and a second call with the same name replaces it. start(name: string): string { @@ -198,7 +181,7 @@ export class WorkflowController { } } this.pendingReplace = undefined; - this.attach(workflow); + this.attachRuntime(workflow); return `Started ${name} workflow.`; } @@ -211,7 +194,7 @@ export class WorkflowController { if (workflow === undefined) return; const runtime = this.attachRuntime(workflow, true); runtime.restore(state); - this.publish(); + this.notify(); } // Toggle a capability off/on for this run. Affects not-yet-reached steps of an @@ -220,7 +203,7 @@ export class WorkflowController { if (this.overrides.has(name)) this.overrides.delete(name); else this.overrides.add(name); this.runtime?.setCapabilities(this.capabilityMap()); - this.publish(); + this.notify(); return this.overrides.has(name) ? `Disabled capability: ${name}.` : `Enabled capability: ${name}.`; diff --git a/tests/unit/workflow-controller.test.ts b/tests/unit/workflow-host.test.ts similarity index 54% rename from tests/unit/workflow-controller.test.ts rename to tests/unit/workflow-host.test.ts index 26ea932d4..7fd1363d2 100644 --- a/tests/unit/workflow-controller.test.ts +++ b/tests/unit/workflow-host.test.ts @@ -1,13 +1,12 @@ import { test, expect } from "bun:test"; import "../helpers/workflows.js"; -import { EventEmitter } from "node:events"; import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import type { ToolDefinition } from "@intx/types/runtime"; import { initSessionDir } from "../../src/session/index.js"; -import { WorkflowController } from "../../src/tui/workflow-controller.js"; import { WorkflowCoordinator } from "../../src/workflows/coordinator.js"; +import { WorkflowHost } from "../../src/workflows/host.js"; import { findWorkflow } from "../../src/workflows/index.js"; import { WorkflowRuntime } from "../../src/workflows/runtime.js"; import { flushWorkflowStateWrites, saveWorkflowState } from "../../src/workflows/state.js"; @@ -16,22 +15,33 @@ function tool(name: string): ToolDefinition { return { name, description: name, inputSchema: { type: "object", properties: {} } }; } -async function withController( +function drain( + host: WorkflowHost, + director: { coordinator: WorkflowCoordinator | undefined }, +): void { + while (host.isActive()) { + const stepId = director.coordinator?.currentStepId(); + expect(stepId).not.toBeNull(); + expect(host.complete(stepId!)).toBe("advanced"); + } +} + +async function withHost( tools: ToolDefinition[], fn: ( - c: WorkflowController, + host: WorkflowHost, director: { coordinator: WorkflowCoordinator | undefined }, cwd: string, home: string, ) => void | Promise, + onChange?: () => void, ): Promise { - const cwd = await mkdtemp(join(tmpdir(), "wf-controller-")); - const home = await mkdtemp(join(tmpdir(), "wf-controller-home-")); + const cwd = await mkdtemp(join(tmpdir(), "wf-host-")); + const home = await mkdtemp(join(tmpdir(), "wf-host-home-")); await initSessionDir(cwd, "session-1", home); const director = { coordinator: undefined as WorkflowCoordinator | undefined }; - const controller = new WorkflowController({ + const host = new WorkflowHost({ cwd, - emitter: new EventEmitter(), getSessionId: () => "session-1", getToolDefinitions: () => tools, getDirector: () => ({ @@ -40,9 +50,10 @@ async function withController( }, }), home, + ...(onChange !== undefined ? { onChange } : {}), }); try { - await fn(controller, director, cwd, home); + await fn(host, director, cwd, home); } finally { await flushWorkflowStateWrites(cwd, "session-1", home); await rm(cwd, { recursive: true, force: true }); @@ -51,56 +62,56 @@ async function withController( } test("starting a workflow attaches a coordinator to the director", async () => { - await withController([], async (controller, director, _cwd) => { - const msg = controller.start("review"); + await withHost([], async (host, director) => { + const msg = host.start("review"); expect(msg).toBe("Started review workflow."); - expect(controller.isActive()).toBe(true); + expect(host.isActive()).toBe(true); expect(director.coordinator).toBeInstanceOf(WorkflowCoordinator); }); }); test("starting an unknown workflow reports an error and stays inactive", async () => { - await withController([], async (controller, _director, _cwd) => { - expect(controller.start("nope")).toContain("No workflow"); - expect(controller.isActive()).toBe(false); + await withHost([], async (host) => { + expect(host.start("nope")).toContain("No workflow"); + expect(host.isActive()).toBe(false); }); }); test("replacing an active workflow requires a confirming second call", async () => { - await withController([], async (controller, _director, _cwd) => { - controller.start("review"); - const first = controller.start("build"); + await withHost([], async (host) => { + host.start("review"); + const first = host.start("build"); expect(first).toContain("again to replace"); - expect(controller.status().name).toBe("review"); - const second = controller.start("build"); + expect(host.status().name).toBe("review"); + const second = host.start("build"); expect(second).toBe("Started build workflow."); - expect(controller.status().name).toBe("build"); + expect(host.status().name).toBe("build"); }); }); test("status reports capability connection and override state", async () => { - await withController([tool("mcp__Linear__save_issue")], async (controller, _director, _cwd) => { - const before = controller.status().capabilities.find((c) => c.name === "ticket-tracker"); + await withHost([tool("mcp__Linear__save_issue")], async (host) => { + const before = host.status().capabilities.find((c) => c.name === "ticket-tracker"); expect(before?.connected).toBe(true); expect(before?.disabled).toBe(false); - controller.toggleCapability("ticket-tracker"); - const after = controller.status().capabilities.find((c) => c.name === "ticket-tracker"); + host.toggleCapability("ticket-tracker"); + const after = host.status().capabilities.find((c) => c.name === "ticket-tracker"); expect(after?.disabled).toBe(true); }); }); test("reset detaches the workflow", async () => { - await withController([], async (controller, director, _cwd) => { - controller.start("review"); - controller.reset(); - expect(controller.isActive()).toBe(false); + await withHost([], async (host, director) => { + host.start("review"); + host.reset(); + expect(host.isActive()).toBe(false); expect(director.coordinator).toBeUndefined(); }); }); test("directive uses submit_output with the current step id", async () => { - await withController([], async (controller, director, _cwd) => { - controller.start("build"); + await withHost([], async (host, director) => { + host.start("build"); const coordinator = director.coordinator!; expect(coordinator).toBeDefined(); const directive = coordinator.directive(); @@ -111,47 +122,56 @@ test("directive uses submit_output with the current step id", async () => { }); }); -test("history() entry after workflow completion contains the workflow name and steps", async () => { - await withController([], async (controller, _director, _cwd) => { - controller.start("review"); - const coordinator = (controller as unknown as { coordinator: WorkflowCoordinator }) - .coordinator!; - while (coordinator.isActive()) { - const stepId = coordinator.currentStepId(); - expect(stepId).not.toBeNull(); - coordinator.handleToolDone("submit_output", { step: stepId }, false); - } - expect(controller.isActive()).toBe(false); - const history = controller.history(); +test("complete() advances the current step and records history", async () => { + await withHost([], async (host, director) => { + host.start("review"); + drain(host, director); + expect(host.isActive()).toBe(false); + const history = host.history(); expect(history).toHaveLength(1); expect(history[0]!.name).toBe("review"); expect(history[0]!.steps.length).toBeGreaterThan(0); }); }); +test("complete() is not-current when no workflow is active", async () => { + await withHost([], async (host) => { + expect(host.complete("any")).toBe("not-current"); + }); +}); + +test("start notifies onChange", async () => { + let changes = 0; + await withHost( + [], + async (host) => { + host.start("review"); + expect(changes).toBeGreaterThan(0); + }, + () => { + changes += 1; + }, + ); +}); + test("resume() uses the same completion listener as a fresh start", async () => { - await withController([], async (controller, director, cwd, home) => { + await withHost([], async (host, director, cwd, home) => { const workflow = findWorkflow("review"); expect(workflow).toBeDefined(); const runtime = new WorkflowRuntime(new Map()); runtime.start(workflow!); await saveWorkflowState(cwd, "session-1", runtime.state(), home); - await controller.resume(); - expect(controller.isActive()).toBe(true); - const coordinator = director.coordinator!; - while (coordinator.isActive()) { - const stepId = coordinator.currentStepId(); - expect(stepId).not.toBeNull(); - coordinator.handleToolDone("submit_output", { step: stepId }, false); - } - expect(controller.history()).toHaveLength(1); - expect(controller.history()[0]!.name).toBe("review"); + await host.resume(); + expect(host.isActive()).toBe(true); + drain(host, director); + expect(host.history()).toHaveLength(1); + expect(host.history()[0]!.name).toBe("review"); }); }); test("resume() restores an on-disk workflow snapshot for the session", async () => { - await withController([], async (controller, director, cwd, home) => { + await withHost([], async (host, director, cwd, home) => { const workflow = findWorkflow("review"); expect(workflow).toBeDefined(); const runtime = new WorkflowRuntime(new Map()); @@ -159,9 +179,9 @@ test("resume() restores an on-disk workflow snapshot for the session", async () runtime.advance(); await saveWorkflowState(cwd, "session-1", runtime.state(), home); - await controller.resume(); - expect(controller.isActive()).toBe(true); - expect(controller.status().name).toBe("review"); + await host.resume(); + expect(host.isActive()).toBe(true); + expect(host.status().name).toBe("review"); expect(director.coordinator).toBeInstanceOf(WorkflowCoordinator); }); }); From 552a37b448b69ef01cab3a8f313c3799366fa832 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Wed, 9 Sep 2026 09:22:53 -0700 Subject: [PATCH 4/7] Format session store and documentation --- src/session/optimized-context-store.test.ts | 5 ++++- src/session/optimized-context-store.ts | 1 - src/subagent/run-audit-store.test.ts | 3 ++- 3 files changed, 6 insertions(+), 3 deletions(-) diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index 76dd7d824..4f55c8217 100644 --- a/src/session/optimized-context-store.test.ts +++ b/src/session/optimized-context-store.test.ts @@ -570,7 +570,10 @@ describe("createOptimizedContextStore checkpoint", () => { ? cmd.cmd.map(String) : []; if (argv[0] === "git" || argv[0]?.endsWith("/git")) gitSpawns.push(argv); - return original(cmd as Parameters[0], opts as Parameters[1]); + return original( + cmd as Parameters[0], + opts as Parameters[1], + ); }) as typeof Bun.spawn; try { await commitEmptyCheckpoint(dir); diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index 5ea3cc502..16e48c2a0 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -584,4 +584,3 @@ export async function createOptimizedContextStore( const { storage } = await createSessionStores(dir, opts); return storage; } - diff --git a/src/subagent/run-audit-store.test.ts b/src/subagent/run-audit-store.test.ts index 6f3604669..62b93d70d 100644 --- a/src/subagent/run-audit-store.test.ts +++ b/src/subagent/run-audit-store.test.ts @@ -16,7 +16,8 @@ const permissionGate = createPermissionGate({ test("runSubAgent threads the isogit audit store and session id into createAgent", async () => { const cwd = await mkdtemp(join(tmpdir(), "corbits-run-audit-")); - const fakeStore = { readBlob: async () => new Uint8Array() } as unknown as ContextStore & AuditStore; + const fakeStore = { readBlob: async () => new Uint8Array() } as unknown as ContextStore & + AuditStore; let seen: { audit: AuditStore; sessionId?: string; storage: ContextStore } | undefined; await withMockedModuleDuring( From f7bbda0530e6c4383c512806a4e7428f5594c72f Mon Sep 17 00:00:00 2001 From: Sawyer Date: Wed, 9 Sep 2026 09:51:00 -0700 Subject: [PATCH 5/7] Persist session commit keys with exclusive create Write first-time commit keys with wx and reload on EEXIST. Wrap JSON.parse failures as Invalid commit signing key. Stop listIndexPaths from swallowing every error. Treat SessionStores as an interface. Drop the IMPLEMENTATION.md claim that exclusive-delta blob staging is in place. --- docs/IMPLEMENTATION.md | 7 +- src/session/assemble-runtime.test.ts | 2 +- src/session/commit-signer.test.ts | 75 +++++++++++++++++++++ src/session/commit-signer.ts | 42 +++++++++--- src/session/optimized-context-store.test.ts | 2 +- src/session/optimized-context-store.ts | 10 +-- 6 files changed, 115 insertions(+), 23 deletions(-) create mode 100644 src/session/commit-signer.test.ts diff --git a/docs/IMPLEMENTATION.md b/docs/IMPLEMENTATION.md index 580f87269..0052175e2 100644 --- a/docs/IMPLEMENTATION.md +++ b/docs/IMPLEMENTATION.md @@ -424,9 +424,10 @@ segments (`turns-0001.jsonl`, ...) that seal at 256KB, so `git add` re-hashes on the small active segment instead of the whole growing file. Segment zero keeps the original filename, so a legacy monolithic `turns.jsonl` reads back as its own first segment. `load` and `readAt` concatenate every segment in order; a torn final line -in the active segment (from a crash mid-write) is dropped on resume. Only tool-output -blobs new since the last commit are staged, and stale segments deleted by a -history rewrite (compaction) are removed from the tree on the next commit. The +in the active segment (from a crash mid-write) is dropped on resume. The wrapper +stages extra tool-output blobs it wrote since the last commit; vendor +`base.commit()` also stages the whole `tool-output/` tree. Stale segments deleted +by a history rewrite (compaction) are removed from the tree on the next commit. The per-commit git tree still grows one entry per spilled tool-output blob across the session; that tree re-write is inherent to git and left as residual cost. diff --git a/src/session/assemble-runtime.test.ts b/src/session/assemble-runtime.test.ts index 35bfe3c08..24751fd27 100644 --- a/src/session/assemble-runtime.test.ts +++ b/src/session/assemble-runtime.test.ts @@ -223,7 +223,7 @@ describe("assembleChatAgent", () => { expect(agentSessionIds).toEqual(["build-session"]); expect(agentStorages).toEqual([fakeStorage]); expect(agentAudits).toEqual([fakeStorage]); - expect(agentAudits[0]).toBe(agentStorages[0]); + expect(Object.is(agentAudits[0], agentStorages[0])).toBe(true); expect(agentCompactors).toEqual([builtCompactor]); }, ); diff --git a/src/session/commit-signer.test.ts b/src/session/commit-signer.test.ts new file mode 100644 index 000000000..0cc553914 --- /dev/null +++ b/src/session/commit-signer.test.ts @@ -0,0 +1,75 @@ +import { describe, test, expect } from "bun:test"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import { loadOrCreateCommitSigner } from "./commit-signer.js"; + +const KEY_FILE = path.join("keys", "commit-ed25519.json"); + +function tempDir(): string { + return fs.mkdtempSync(path.join(os.tmpdir(), "commit-signer-")); +} + +function keyPath(dir: string): string { + return path.join(dir, KEY_FILE); +} + +function publicKeyInFile(dir: string): string { + const parsed = JSON.parse(fs.readFileSync(keyPath(dir), "utf8")) as { publicKey: string }; + return parsed.publicKey; +} + +describe("loadOrCreateCommitSigner", () => { + test("first call creates a signed-able signer and a 0600 key file", async () => { + const dir = tempDir(); + const signer = await loadOrCreateCommitSigner(dir); + const signature = await signer("payload"); + expect(typeof signature).toBe("string"); + expect(signature.length).toBeGreaterThan(0); + + const st = fs.statSync(keyPath(dir)); + expect(st.mode & 0o777).toBe(0o600); + }); + + test("second call reloads the same key (same publicKey bytes in the file)", async () => { + const dir = tempDir(); + await loadOrCreateCommitSigner(dir); + const firstPublic = publicKeyInFile(dir); + await loadOrCreateCommitSigner(dir); + expect(publicKeyInFile(dir)).toBe(firstPublic); + }); + + test("concurrent first-time create: both succeed, only one key file, both signers work", async () => { + const dir = tempDir(); + const [a, b] = await Promise.all([ + loadOrCreateCommitSigner(dir), + loadOrCreateCommitSigner(dir), + ]); + const sigA = await a("a"); + const sigB = await b("b"); + expect(typeof sigA).toBe("string"); + expect(typeof sigB).toBe("string"); + expect(sigA.length).toBeGreaterThan(0); + expect(sigB.length).toBeGreaterThan(0); + expect(fs.readdirSync(path.join(dir, "keys"))).toEqual(["commit-ed25519.json"]); + }); + + test("corrupt JSON throws Invalid commit signing key", async () => { + const dir = tempDir(); + fs.mkdirSync(path.join(dir, "keys")); + fs.writeFileSync(keyPath(dir), "{not-json"); + await expect(loadOrCreateCommitSigner(dir)).rejects.toThrow( + `Invalid commit signing key at ${keyPath(dir)}`, + ); + }); + + test("invalid arktype shape throws Invalid commit signing key", async () => { + const dir = tempDir(); + fs.mkdirSync(path.join(dir, "keys")); + fs.writeFileSync(keyPath(dir), JSON.stringify({ privateKey: 1 })); + await expect(loadOrCreateCommitSigner(dir)).rejects.toThrow( + `Invalid commit signing key at ${keyPath(dir)}`, + ); + }); +}); diff --git a/src/session/commit-signer.ts b/src/session/commit-signer.ts index eb9832d63..2b9fbc804 100644 --- a/src/session/commit-signer.ts +++ b/src/session/commit-signer.ts @@ -35,7 +35,16 @@ async function loadPersistedKeyPair( if (cause instanceof Error && "code" in cause && cause.code === "ENOENT") return null; throw cause; } - const parsed = PersistedKeyPair(JSON.parse(raw) as unknown); + let parsedJson: unknown; + try { + parsedJson = JSON.parse(raw); + } catch (cause) { + if (cause instanceof SyntaxError) { + throw new Error(`Invalid commit signing key at ${filePath}`, { cause }); + } + throw cause; + } + const parsed = PersistedKeyPair(parsedJson); if (parsed instanceof type.errors) { throw new Error(`Invalid commit signing key at ${filePath}: ${parsed.summary}`); } @@ -52,16 +61,27 @@ export async function loadOrCreateCommitSigner(dir: string): Promise createSSHSignature(payload, keyPair.privateKey, keyPair.publicKey); } diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index 4f55c8217..a1762051c 100644 --- a/src/session/optimized-context-store.test.ts +++ b/src/session/optimized-context-store.test.ts @@ -588,7 +588,7 @@ describe("createSessionStores", () => { test("exposes the same object as ContextStore and AuditStore", async () => { const dir = tempDir(); const { storage, audit } = await createSessionStores(dir); - expect(storage).toBe(audit); + expect(Object.is(storage, audit)).toBe(true); expect(typeof audit.commitAudit).toBe("function"); expect(typeof audit.commitErrors).toBe("function"); expect(typeof audit.loadAudit).toBe("function"); diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index 16e48c2a0..7d8bdec83 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -307,11 +307,7 @@ export async function loadRecentTurns(dir: string, minTurns: number): Promise> { - try { - return new Set(await git.listFiles({ fs, dir })); - } catch { - return new Set(); - } + return new Set(await git.listFiles({ fs, dir })); } async function extraSegmentNamesAtCommit(dir: string, hash: string): Promise { @@ -383,10 +379,10 @@ function extraCommitPaths(paths: readonly string[]): string[] { return paths.filter((filepath) => !VENDOR_COMMIT_ROOT_FILES.has(filepath)); } -export type SessionStores = { +export interface SessionStores { storage: ContextStore; audit: AuditStore; -}; +} /** * Local wrapper around the Interchange git store that avoids O(session length) From 30d5ea8f3d3913eeceeee6c6673e619d0c332878 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Wed, 9 Sep 2026 09:51:03 -0700 Subject: [PATCH 6/7] Flash workflow status from the product host The runner now emits workflow. The product host subscribes and flashes the active step or complete via existing notices. --- src/tui/product-host.ts | 14 +++++ src/tui/runtime-channels.test.ts | 70 ++++++++++++++++++++- src/tui/runtime-notices.test.ts | 104 +++++++++++++++++++++++++++++++ src/tui/runtime-notices.ts | 54 ++++++++++++++++ 4 files changed, 241 insertions(+), 1 deletion(-) diff --git a/src/tui/product-host.ts b/src/tui/product-host.ts index 6cf65a1fe..f19bb4fbc 100644 --- a/src/tui/product-host.ts +++ b/src/tui/product-host.ts @@ -34,6 +34,8 @@ import { mcpServerState, RUNTIME_FLASH_MS, type RuntimeNotice, + workflowNotice, + workflowPayloadInfo, } from "./runtime-notices.js"; import type { PaletteCommand } from "./command-catalog.js"; import { @@ -388,6 +390,7 @@ export async function mountProductHost(config: ProductHostConfig): Promise { }); }); +describe("workflow channel", () => { + const liveIdle = { + current: { + active: false, + name: undefined as string | undefined, + stepIndex: 0, + total: 0, + label: "", + steps: [] as unknown[], + capabilities: [] as unknown[], + }, + history: [{ name: "ship" }], + }; + + test("an active live emit paints the step and holds no transcript row", async () => { + const { host, emitter, frame, cleanup } = await mountHeadless(); + try { + emitter.emit("workflow", { + current: { active: true, name: "ship", stepIndex: 0, total: 2, label: "build" }, + history: [], + }); + expect(await frame()).toContain("workflow ship · step 1/2: build"); + expect(host.shell.streamLog).toEqual([]); + } finally { + cleanup(); + } + }); + + test("active then idle paints complete once and streamLog stays empty", async () => { + const { host, emitter, frame, cleanup } = await mountHeadless(); + try { + emitter.emit("workflow", { + current: { active: true, name: "ship", stepIndex: 0, total: 2, label: "build" }, + history: [], + }); + expect(await frame()).toContain("workflow ship · step 1/2: build"); + emitter.emit("workflow", liveIdle); + expect(await frame()).toContain("workflow ship complete"); + expect(host.shell.streamLog).toEqual([]); + emitter.emit("workflow", liveIdle); + expect(host.shell.streamLog).toEqual([]); + } finally { + cleanup(); + } + }); + + test("first-emit idle with history does not complete-flash", async () => { + const { emitter, frame, cleanup } = await mountHeadless(); + try { + emitter.emit("workflow", liveIdle); + expect(await frame()).not.toContain("workflow ship complete"); + } finally { + cleanup(); + } + }); + + test("a bad payload paints nothing", async () => { + const { host, emitter, frame, cleanup } = await mountHeadless(); + try { + emitter.emit("workflow", { current: { active: true } }); + expect(await frame()).not.toContain("workflow ship"); + expect(host.shell.streamLog).toEqual([]); + } finally { + cleanup(); + } + }); +}); + /** * Static guard for the whole bug class: an emitted channel with no `.on` * anywhere is a feature nobody can see, and it fails silently. Static because @@ -300,7 +368,7 @@ describe("every emitted runtime channel has a subscriber", () => { emitted.delete("subagent.progress"); test("the runner still emits the channels this suite knows about", () => { - for (const channel of ["hook", "mcp.status", "permission.grant", "compaction"]) { + for (const channel of ["hook", "mcp.status", "permission.grant", "compaction", "workflow"]) { expect([...emitted]).toContain(channel); } }); diff --git a/src/tui/runtime-notices.test.ts b/src/tui/runtime-notices.test.ts index 9feadba60..3b5592bd2 100644 --- a/src/tui/runtime-notices.test.ts +++ b/src/tui/runtime-notices.test.ts @@ -13,6 +13,8 @@ import { mcpNotice, mcpServerState, subAgentProgress, + workflowNotice, + workflowPayloadInfo, } from "./runtime-notices.js"; const hook = { @@ -166,4 +168,106 @@ describe("payload validation", () => { expect(compactionFoldInfo(null)).toBeNull(); expect(compactionFoldInfo("nope")).toBeNull(); }); + + test("workflow payloads require current + history and reject junk", () => { + expect( + workflowPayloadInfo({ + current: { active: true, name: "ship", stepIndex: 0, total: 2, label: "build" }, + history: [], + }), + ).toEqual({ + current: { active: true, name: "ship", stepIndex: 0, total: 2, label: "build" }, + history: [], + }); + expect(workflowPayloadInfo({ current: { active: true } })).toBeNull(); + expect(workflowPayloadInfo(null)).toBeNull(); + expect(workflowPayloadInfo("nope")).toBeNull(); + }); + + test("live inactive payload with name: undefined and extra status keys parses", () => { + const parsed = workflowPayloadInfo({ + current: { + active: false, + name: undefined, + stepIndex: 0, + total: 0, + label: "", + steps: [{ id: "a" }], + capabilities: ["x"], + }, + history: [{ name: "ship", extra: true }], + }); + expect(parsed).not.toBeNull(); + expect(parsed?.current.active).toBe(false); + expect(parsed?.current.name).toBeUndefined(); + expect(parsed?.history.at(-1)?.name).toBe("ship"); + }); +}); + +describe("workflowNotice", () => { + test("active named step flashes index+1 and label", () => { + expect( + workflowNotice({ + current: { active: true, name: "ship", stepIndex: 0, total: 2, label: "build" }, + history: [], + }), + ).toEqual({ + kind: "flash", + text: "workflow ship · step 1/2: build", + }); + }); + + test("inactive with last history name flashes complete only when wasActive", () => { + const payload = { + current: { + active: false, + name: undefined as string | undefined, + stepIndex: 1, + total: 2, + label: "done", + }, + history: [{ name: "ship" }], + }; + expect(workflowNotice(payload, { wasActive: true })).toEqual({ + kind: "flash", + text: "workflow ship complete", + }); + expect(workflowNotice(payload, { wasActive: false })).toBeNull(); + expect(workflowNotice(payload)).toBeNull(); + }); + + test("idle snapshots say nothing", () => { + expect( + workflowNotice({ + current: { active: false, stepIndex: 0, total: 0, label: "" }, + history: [], + }), + ).toBeNull(); + expect( + workflowNotice({ + current: { active: true, stepIndex: 0, total: 1, label: "x" }, + history: [], + }), + ).toBeNull(); + }); + + test("active live payload with extra keys still flashes the step", () => { + const parsed = workflowPayloadInfo({ + current: { + active: true, + name: "ship", + stepIndex: 0, + total: 2, + label: "build", + steps: [], + capabilities: [], + }, + history: [], + }); + expect(parsed).not.toBeNull(); + expect(workflowNotice(parsed!)).toEqual({ + kind: "flash", + text: "workflow ship · step 1/2: build", + }); + }); }); diff --git a/src/tui/runtime-notices.ts b/src/tui/runtime-notices.ts index f6a7c6eea..ba2447903 100644 --- a/src/tui/runtime-notices.ts +++ b/src/tui/runtime-notices.ts @@ -208,3 +208,57 @@ export function subAgentProgress(raw: unknown): SubAgentProgress | null { if (parsed instanceof type.errors) return null; return parsed; } + +export interface WorkflowNoticePayload { + readonly current: { + readonly active: boolean; + readonly name?: string | undefined; + readonly stepIndex: number; + readonly total: number; + readonly label: string; + }; + readonly history: readonly { readonly name?: string | undefined }[]; +} + +const workflowHistoryEntry = type({ "name?": "string | undefined" }); + +const workflowPayload = type({ + current: { + active: "boolean", + "name?": "string | undefined", + stepIndex: "number", + total: "number", + label: "string", + }, + history: workflowHistoryEntry.array(), +}); + +export function workflowPayloadInfo(raw: unknown): WorkflowNoticePayload | null { + const parsed = workflowPayload(raw); + if (parsed instanceof type.errors) return null; + return parsed; +} + +/** + * Live workflow projection. Active named steps flash the current step; + * complete flashes only on the active→idle transition when last history has a name. + */ +export function workflowNotice( + payload: WorkflowNoticePayload, + opts?: { wasActive?: boolean }, +): RuntimeNotice | null { + const { current, history } = payload; + if (current.active && current.name !== undefined && current.name.length > 0) { + return { + kind: "flash", + text: `workflow ${current.name} · step ${current.stepIndex + 1}/${current.total}: ${current.label}`, + }; + } + if (!current.active && opts?.wasActive === true) { + const last = history.at(-1); + if (last?.name !== undefined && last.name.length > 0) { + return { kind: "flash", text: `workflow ${last.name} complete` }; + } + } + return null; +} From ab7a6dcd2c1f306424689ce3f7cfef997748e957 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Wed, 9 Sep 2026 22:58:09 -0700 Subject: [PATCH 7/7] Read worker audit from the shared session store --- docs/IMPLEMENTATION.md | 6 +++--- tests/integration/subagent-permission.test.ts | 4 ++-- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/docs/IMPLEMENTATION.md b/docs/IMPLEMENTATION.md index 0052175e2..fecbe76e4 100644 --- a/docs/IMPLEMENTATION.md +++ b/docs/IMPLEMENTATION.md @@ -400,9 +400,9 @@ Session runtime state lives under the global projects tree (not in the repo): - Migration: if a session exists only under in-repo `.agent-state//`, it is moved into the global tree on open/list - Atomic JSON writes with schema validation on load -**Worker audit persistence.** Workers initialize a real `@intx/storage-isogit` -`AuditStore` at `/audit-store` (`src/subagent/run.ts`), separate -from the native context store's Git index. Initialization failure prevents worker +**Worker audit persistence.** Workers get context and audit from one +`createSessionStores` call on the worker workdir: the same isomorphic-git +repo, one index. Initialization failure prevents worker execution. The existing agent-owned audit and error collectors persist at checkpoint and shutdown; retained worker sessions flush at checkpoint/resume and close. The parent still supplies `noopAuditStore()`: collectors exist there too, diff --git a/tests/integration/subagent-permission.test.ts b/tests/integration/subagent-permission.test.ts index 269d0e43c..b366ef993 100644 --- a/tests/integration/subagent-permission.test.ts +++ b/tests/integration/subagent-permission.test.ts @@ -42,7 +42,7 @@ async function withWorker( const cwd = await mkdtemp(join(tmpdir(), "worker-permission-")); const harness = setupHarness(); const workdirBase = join(cwd, "state"); - const auditPath = join(workdirBase, "subagents", "worker", "audit-store"); + const auditPath = join(workdirBase, "subagents", "worker"); const params: RunSubAgentParams = { id: "worker", cwd, @@ -703,7 +703,7 @@ test.serial( if (nested === undefined) throw new Error("missing nested worker"); expect(nested.finishedAt).toBeDefined(); expect(await Bun.file(join(cwd, "probe.txt")).exists()).toBe(false); - const auditPath = join(params.workdirBase, "subagents", nested.id, "audit-store"); + const auditPath = join(params.workdirBase, "subagents", nested.id); const store = await createIsogitStore(auditPath); const [sessionId] = await readdir(join(auditPath, "state", "audit")); if (sessionId === undefined) throw new Error("missing nested audit");