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/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/docs/IMPLEMENTATION.md b/docs/IMPLEMENTATION.md index 69798ff5c..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, @@ -415,18 +415,19 @@ 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 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. @@ -485,13 +486,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/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/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 5a8f55ae5..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, @@ -615,6 +618,7 @@ export async function runExec(config: Config): Promise { }, getProvider: () => config, getWorkdir: () => workdir, + getSessionId: () => sessionId, authorize: createReactorAuthorize(permissionGate), inferenceDeps, getSources: () => { @@ -637,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, @@ -681,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/session/assemble-runtime.test.ts b/src/session/assemble-runtime.test.ts index c4c7b8716..24751fd27 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(Object.is(agentAudits[0], agentStorages[0])).toBe(true); 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/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 new file mode 100644 index 000000000..2b9fbc804 --- /dev/null +++ b/src/session/commit-signer.ts @@ -0,0 +1,87 @@ +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; + } + 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}`); + } + 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 }); + try { + 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, flag: "wx" }, + ); + log.debug?.("wrote session commit signing key"); + keyPair = generated; + } catch (cause) { + if (cause instanceof Error && "code" in cause && cause.code === "EEXIST") { + keyPair = await loadPersistedKeyPair(filePath); + if (keyPair === null) { + throw new Error(`Commit signing key missing after EEXIST at ${filePath}`); + } + } else { + throw cause; + } + } + } + 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..a1762051c 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,49 @@ 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(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 245a685f0..7d8bdec83 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,13 @@ 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()}`); - } - return stdout.trimEnd(); +async function listIndexPaths(dir: string): Promise> { + return new Set(await git.listFiles({ fs, dir })); } -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 +322,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 +356,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 +367,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 +375,27 @@ async function reconcileSegmentStaging( } } +function extraCommitPaths(paths: readonly string[]): string[] { + return paths.filter((filepath) => !VENDOR_COMMIT_ROOT_FILES.has(filepath)); +} + +export interface 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 +441,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 +504,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 +523,60 @@ 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); + } +} diff --git a/src/subagent/run-audit-store.test.ts b/src/subagent/run-audit-store.test.ts new file mode 100644 index 000000000..62b93d70d --- /dev/null +++ b/src/subagent/run-audit-store.test.ts @@ -0,0 +1,77 @@ +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/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 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 05a19ac98..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: { @@ -471,6 +476,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]), @@ -535,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/runtime-channels.test.ts b/src/tui/runtime-channels.test.ts index 6e21ae1af..10a16679d 100644 --- a/src/tui/runtime-channels.test.ts +++ b/src/tui/runtime-channels.test.ts @@ -271,6 +271,74 @@ describe("agents chrome (live strip above the prompt)", () => { }); }); +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; +} 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/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"); 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); }); });