+24

docs/exec-plans/active/vertical-slices.md

reviewed

··· 89 89 - `bun test packages/cli/src` 90 90 - `bun test packages/driver-pi/src` 91 91 92 92 + **Status (2026-02-23)** 93 93 + 94 94 + - ✅ Added Tier-1 event discriminated union schemas in `packages/core/src/domain/event.schema.ts` with persisted decode helpers (`Schema.parseJson` + `Schema.decodeUnknown*`). 95 95 + - ✅ Expanded run/spawn schema contracts with typed decode utilities for persisted artifacts (`run.json`, `result.json`) and runtime validation of `SpawnResult` (`sessionRef` non-empty). 96 96 + - ✅ Implemented `packages/core/src/internal/run-store.effect.ts` for run directory creation and append-only `events.ndjson` persistence. 97 97 + - ✅ Implemented sync lifecycle orchestration in `packages/core/src/internal/engine.effect.ts` (run start/status, spawn mapping, run terminal persistence). 98 98 + - ✅ Implemented `run --sync` and `status` CLI command handlers in `packages/cli/src/public/index.api.ts`, with JSON mode contract preserved on stdout. 99 99 + - ✅ Implemented process-backed pi driver runtime in `packages/driver-pi` with codec decoding and command invocation via `Command.make(cmd, ...args)`. 100 100 + - ✅ Added unit/integration/e2e coverage for schemas, run store, engine↔driver flow, CLI `run --sync`, CLI `status`, and persisted artifact verification. 101 101 + - ✅ Re-ran targeted slice suites (`core/domain`, `core/internal`, `cli/src`, `driver-pi/src`) with green results. 102 102 + 92 103 --- 93 104 94 105 ## S3 — Wait Semantics + Terminal Single-Shot Invariants ··· 121 132 - `bun test packages/core/src/internal` 122 133 - `bun test packages/cli/src/public` 123 134 - `bun test packages/driver-pi/src` 135 135 + 136 136 + **Status (2026-02-23)** 137 137 + 138 138 + - ✅ Added core lifecycle transition guards in `packages/core/src/internal/lifecycle-guard.effect.ts` and unit coverage for terminal single-shot / terminal→non-terminal rejection paths. 139 139 + - ✅ Hardened run status transitions in `RunStore` so terminal statuses are immutable (`complete|failed|cancelled` cannot transition further). 140 140 + - ✅ Implemented `MillEngine.wait(runId, timeout)` with deterministic polling over persisted events plus typed timeout error (`WaitTimeoutError`). 141 141 + - ✅ `wait` now validates persisted event streams with lifecycle guards and tracks terminal observation without allowing post-terminal transitions. 142 142 + - ✅ Implemented deterministic terminal policy: duplicate terminal emissions are rejected via `LifecycleInvariantError` (not ignored). 143 143 + - ✅ Added CLI `wait` command (`mill wait <runId> --timeout <seconds> [--json]`) with JSON/human output parity. 144 144 + - ✅ Added typed JSON timeout contract for CLI (`{ ok: false, error: { _tag: "WaitTimeoutError", runId, timeoutSeconds, message } }`) and non-zero timeout exit code. 145 145 + - ✅ Added driver-pi malformed fixture coverage for duplicate/invalid terminal output ordering in codec + runtime integration tests. 146 146 + - ✅ Added integration/e2e coverage for persisted/live wait behavior and timeout behavior. 147 147 + - ✅ Re-ran full workspace `bun test`; suite is green. 124 148 125 149 --- 126 150

+318

packages/cli/src/public/index.api.test.ts

reviewed

··· 1 1 import { describe, expect, it } from "bun:test"; 2 2 + import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; 3 3 + import { tmpdir } from "node:os"; 4 4 + import { join } from "node:path"; 2 5 import * as Schema from "@effect/schema/Schema"; 3 6 import { runCli } from "./index.api"; 4 7 ··· 29 32 }), 30 33 ); 31 34 35 35 + const RunSyncEnvelope = Schema.parseJson( 36 36 + Schema.Struct({ 37 37 + run: Schema.Struct({ 38 38 + id: Schema.String, 39 39 + status: Schema.String, 40 40 + paths: Schema.Struct({ 41 41 + runDir: Schema.String, 42 42 + runFile: Schema.String, 43 43 + eventsFile: Schema.String, 44 44 + resultFile: Schema.String, 45 45 + }), 46 46 + }), 47 47 + result: Schema.Struct({ 48 48 + runId: Schema.String, 49 49 + status: Schema.String, 50 50 + spawns: Schema.Array( 51 51 + Schema.Struct({ 52 52 + text: Schema.String, 53 53 + sessionRef: Schema.String, 54 54 + agent: Schema.String, 55 55 + model: Schema.String, 56 56 + driver: Schema.String, 57 57 + exitCode: Schema.Number, 58 58 + }), 59 59 + ), 60 60 + }), 61 61 + }), 62 62 + ); 63 63 + 64 64 + const StatusEnvelope = Schema.parseJson( 65 65 + Schema.Struct({ 66 66 + id: Schema.String, 67 67 + status: Schema.String, 68 68 + paths: Schema.Struct({ 69 69 + runDir: Schema.String, 70 70 + runFile: Schema.String, 71 71 + eventsFile: Schema.String, 72 72 + resultFile: Schema.String, 73 73 + }), 74 74 + }), 75 75 + ); 76 76 + 77 77 + const WaitTimeoutEnvelope = Schema.parseJson( 78 78 + Schema.Struct({ 79 79 + ok: Schema.Literal(false), 80 80 + error: Schema.Struct({ 81 81 + _tag: Schema.Literal("WaitTimeoutError"), 82 82 + runId: Schema.String, 83 83 + timeoutSeconds: Schema.Number, 84 84 + message: Schema.String, 85 85 + }), 86 86 + }), 87 87 + ); 88 88 + 32 89 describe("runCli", () => { 33 90 it("writes machine payload to stdout only in --json mode", async () => { 34 91 const stdout: Array<string> = []; ··· 83 140 expect(stdout).toHaveLength(1); 84 141 expect(stderr).toHaveLength(0); 85 142 expect(stdout[0]).toContain("mill — Effect-first orchestration runtime"); 143 143 + }); 144 144 + 145 145 + it("executes run --sync and resolves status for persisted runId", async () => { 146 146 + const tempDirectory = await mkdtemp(join(tmpdir(), "mill-cli-run-")); 147 147 + const homeDirectory = join(tempDirectory, "home"); 148 148 + const programPath = join(tempDirectory, "program.ts"); 149 149 + 150 150 + await writeFile( 151 151 + programPath, 152 152 + [ 153 153 + "const scan = await mill.spawn({", 154 154 + ' agent: "scout",', 155 155 + ' systemPrompt: "You are concise.",', 156 156 + ' prompt: "Say hello",', 157 157 + "});", 158 158 + "globalThis.__millLastText = scan.text;", 159 159 + ].join("\n"), 160 160 + "utf-8", 161 161 + ); 162 162 + 163 163 + const runStdout: Array<string> = []; 164 164 + const runStderr: Array<string> = []; 165 165 + 166 166 + try { 167 167 + const runCode = await runCli(["run", programPath, "--sync", "--json"], { 168 168 + cwd: tempDirectory, 169 169 + homeDirectory, 170 170 + pathExists: async () => false, 171 171 + io: { 172 172 + stdout: (line) => { 173 173 + runStdout.push(line); 174 174 + }, 175 175 + stderr: (line) => { 176 176 + runStderr.push(line); 177 177 + }, 178 178 + }, 179 179 + }); 180 180 + 181 181 + expect(runCode).toBe(0); 182 182 + expect(runStderr).toHaveLength(0); 183 183 + expect(runStdout).toHaveLength(1); 184 184 + 185 185 + const runPayload = Schema.decodeUnknownSync(RunSyncEnvelope)(runStdout[0]); 186 186 + expect(runPayload.run.status).toBe("complete"); 187 187 + expect(runPayload.result.status).toBe("complete"); 188 188 + expect(runPayload.result.spawns).toHaveLength(1); 189 189 + 190 190 + const statusStdout: Array<string> = []; 191 191 + const statusStderr: Array<string> = []; 192 192 + 193 193 + const statusCode = await runCli(["status", runPayload.run.id, "--json"], { 194 194 + cwd: tempDirectory, 195 195 + homeDirectory, 196 196 + pathExists: async () => false, 197 197 + io: { 198 198 + stdout: (line) => { 199 199 + statusStdout.push(line); 200 200 + }, 201 201 + stderr: (line) => { 202 202 + statusStderr.push(line); 203 203 + }, 204 204 + }, 205 205 + }); 206 206 + 207 207 + expect(statusCode).toBe(0); 208 208 + expect(statusStderr).toHaveLength(0); 209 209 + expect(statusStdout).toHaveLength(1); 210 210 + 211 211 + const statusPayload = Schema.decodeUnknownSync(StatusEnvelope)(statusStdout[0]); 212 212 + expect(statusPayload.id).toBe(runPayload.run.id); 213 213 + expect(statusPayload.status).toBe("complete"); 214 214 + } finally { 215 215 + await rm(tempDirectory, { recursive: true, force: true }); 216 216 + } 217 217 + }); 218 218 + 219 219 + it("wait returns terminal run payload with --json and human output parity", async () => { 220 220 + const tempDirectory = await mkdtemp(join(tmpdir(), "mill-cli-wait-")); 221 221 + const homeDirectory = join(tempDirectory, "home"); 222 222 + const programPath = join(tempDirectory, "program.ts"); 223 223 + 224 224 + await writeFile( 225 225 + programPath, 226 226 + [ 227 227 + "const scan = await mill.spawn({", 228 228 + ' agent: "scout",', 229 229 + ' systemPrompt: "You are concise.",', 230 230 + ' prompt: "Say hello",', 231 231 + "});", 232 232 + "globalThis.__millWaitText = scan.text;", 233 233 + ].join("\n"), 234 234 + "utf-8", 235 235 + ); 236 236 + 237 237 + try { 238 238 + const runStdout: Array<string> = []; 239 239 + const runCode = await runCli(["run", programPath, "--sync", "--json"], { 240 240 + cwd: tempDirectory, 241 241 + homeDirectory, 242 242 + pathExists: async () => false, 243 243 + io: { 244 244 + stdout: (line) => { 245 245 + runStdout.push(line); 246 246 + }, 247 247 + stderr: () => undefined, 248 248 + }, 249 249 + }); 250 250 + 251 251 + expect(runCode).toBe(0); 252 252 + const runPayload = Schema.decodeUnknownSync(RunSyncEnvelope)(runStdout[0]); 253 253 + 254 254 + const waitJsonStdout: Array<string> = []; 255 255 + const waitJsonStderr: Array<string> = []; 256 256 + const waitJsonCode = await runCli( 257 257 + ["wait", runPayload.run.id, "--timeout", "2", "--json"], 258 258 + { 259 259 + cwd: tempDirectory, 260 260 + homeDirectory, 261 261 + pathExists: async () => false, 262 262 + io: { 263 263 + stdout: (line) => { 264 264 + waitJsonStdout.push(line); 265 265 + }, 266 266 + stderr: (line) => { 267 267 + waitJsonStderr.push(line); 268 268 + }, 269 269 + }, 270 270 + }, 271 271 + ); 272 272 + 273 273 + expect(waitJsonCode).toBe(0); 274 274 + expect(waitJsonStderr).toHaveLength(0); 275 275 + const waitJsonPayload = Schema.decodeUnknownSync(StatusEnvelope)(waitJsonStdout[0]); 276 276 + expect(waitJsonPayload.id).toBe(runPayload.run.id); 277 277 + expect(waitJsonPayload.status).toBe("complete"); 278 278 + 279 279 + const waitHumanStdout: Array<string> = []; 280 280 + const waitHumanStderr: Array<string> = []; 281 281 + const waitHumanCode = await runCli(["wait", runPayload.run.id, "--timeout", "2"], { 282 282 + cwd: tempDirectory, 283 283 + homeDirectory, 284 284 + pathExists: async () => false, 285 285 + io: { 286 286 + stdout: (line) => { 287 287 + waitHumanStdout.push(line); 288 288 + }, 289 289 + stderr: (line) => { 290 290 + waitHumanStderr.push(line); 291 291 + }, 292 292 + }, 293 293 + }); 294 294 + 295 295 + expect(waitHumanCode).toBe(0); 296 296 + expect(waitHumanStderr).toHaveLength(0); 297 297 + expect(waitHumanStdout[0]).toContain(`run ${runPayload.run.id}`); 298 298 + expect(waitHumanStdout[0]).toContain("status=complete"); 299 299 + } finally { 300 300 + await rm(tempDirectory, { recursive: true, force: true }); 301 301 + } 302 302 + }); 303 303 + 304 304 + it("wait timeout is deterministic with typed JSON error contract", async () => { 305 305 + const tempDirectory = await mkdtemp(join(tmpdir(), "mill-cli-wait-timeout-")); 306 306 + const runsDirectory = join(tempDirectory, "runs"); 307 307 + const runId = `run_timeout_${crypto.randomUUID()}`; 308 308 + const runDir = join(runsDirectory, runId); 309 309 + const runFile = join(runDir, "run.json"); 310 310 + const eventsFile = join(runDir, "events.ndjson"); 311 311 + const resultFile = join(runDir, "result.json"); 312 312 + 313 313 + await mkdir(runDir, { recursive: true }); 314 314 + await writeFile( 315 315 + runFile, 316 316 + `${JSON.stringify( 317 317 + { 318 318 + id: runId, 319 319 + status: "running", 320 320 + programPath: "/tmp/program.ts", 321 321 + driver: "default", 322 322 + createdAt: "2026-02-23T20:00:00.000Z", 323 323 + updatedAt: "2026-02-23T20:00:00.000Z", 324 324 + paths: { 325 325 + runDir, 326 326 + runFile, 327 327 + eventsFile, 328 328 + resultFile, 329 329 + }, 330 330 + }, 331 331 + null, 332 332 + 2, 333 333 + )}\n`, 334 334 + "utf-8", 335 335 + ); 336 336 + await writeFile( 337 337 + eventsFile, 338 338 + `${JSON.stringify({ 339 339 + schemaVersion: 1, 340 340 + runId, 341 341 + sequence: 1, 342 342 + timestamp: "2026-02-23T20:00:00.000Z", 343 343 + type: "run:start", 344 344 + payload: { 345 345 + programPath: "/tmp/program.ts", 346 346 + }, 347 347 + })}\n`, 348 348 + "utf-8", 349 349 + ); 350 350 + 351 351 + try { 352 352 + const jsonStdout: Array<string> = []; 353 353 + const jsonStderr: Array<string> = []; 354 354 + 355 355 + const jsonCode = await runCli( 356 356 + ["wait", runId, "--timeout", "1", "--json", "--runs-dir", runsDirectory], 357 357 + { 358 358 + cwd: tempDirectory, 359 359 + homeDirectory: join(tempDirectory, "home"), 360 360 + pathExists: async () => false, 361 361 + io: { 362 362 + stdout: (line) => { 363 363 + jsonStdout.push(line); 364 364 + }, 365 365 + stderr: (line) => { 366 366 + jsonStderr.push(line); 367 367 + }, 368 368 + }, 369 369 + }, 370 370 + ); 371 371 + 372 372 + expect(jsonCode).toBe(2); 373 373 + expect(jsonStderr).toHaveLength(0); 374 374 + const timeoutPayload = Schema.decodeUnknownSync(WaitTimeoutEnvelope)(jsonStdout[0]); 375 375 + expect(timeoutPayload.error.runId).toBe(runId); 376 376 + expect(timeoutPayload.error.timeoutSeconds).toBe(1); 377 377 + 378 378 + const humanStdout: Array<string> = []; 379 379 + const humanStderr: Array<string> = []; 380 380 + 381 381 + const humanCode = await runCli( 382 382 + ["wait", runId, "--timeout", "1", "--runs-dir", runsDirectory], 383 383 + { 384 384 + cwd: tempDirectory, 385 385 + homeDirectory: join(tempDirectory, "home"), 386 386 + pathExists: async () => false, 387 387 + io: { 388 388 + stdout: (line) => { 389 389 + humanStdout.push(line); 390 390 + }, 391 391 + stderr: (line) => { 392 392 + humanStderr.push(line); 393 393 + }, 394 394 + }, 395 395 + }, 396 396 + ); 397 397 + 398 398 + expect(humanCode).toBe(2); 399 399 + expect(humanStdout).toHaveLength(0); 400 400 + expect(humanStderr[0]).toContain(`Timeout waiting for run ${runId}`); 401 401 + } finally { 402 402 + await rm(tempDirectory, { recursive: true, force: true }); 403 403 + } 86 404 }); 87 405 });

+199 -2

packages/cli/src/public/index.api.ts

reviewed

··· 1 1 import { 2 2 createDiscoveryPayload, 3 3 defineConfig, 4 4 + getRunStatus, 4 5 processDriver, 6 6 + runProgramSync, 7 7 + waitForRun, 5 8 type ConfigOverrides, 6 9 } from "@mill/core"; 7 10 import { createPiDriverRegistration } from "@mill/driver-pi"; ··· 14 17 interface RunCliOptions { 15 18 readonly cwd?: string; 16 19 readonly homeDirectory?: string; 20 20 + readonly runsDirectory?: string; 17 21 readonly pathExists?: (path: string) => Promise<boolean>; 18 22 readonly loadConfigOverrides?: (path: string) => Promise<ConfigOverrides>; 19 23 readonly io?: CliIo; ··· 40 44 }, 41 45 }); 42 46 47 47 + const readFlagValue = (argv: ReadonlyArray<string>, flag: string): string | undefined => { 48 48 + const index = argv.indexOf(flag); 49 49 + 50 50 + if (index < 0) { 51 51 + return undefined; 52 52 + } 53 53 + 54 54 + return argv[index + 1]; 55 55 + }; 56 56 + 57 57 + const parseTimeoutSeconds = (argv: ReadonlyArray<string>): number | undefined => { 58 58 + const value = readFlagValue(argv, "--timeout"); 59 59 + 60 60 + if (value === undefined) { 61 61 + return undefined; 62 62 + } 63 63 + 64 64 + const parsed = Number.parseFloat(value); 65 65 + 66 66 + if (!Number.isFinite(parsed) || parsed <= 0) { 67 67 + return undefined; 68 68 + } 69 69 + 70 70 + return parsed; 71 71 + }; 72 72 + 73 73 + const runSyncCommand = async ( 74 74 + argv: ReadonlyArray<string>, 75 75 + options: RunCliOptions, 76 76 + io: CliIo, 77 77 + ): Promise<number> => { 78 78 + const programPath = argv[0]; 79 79 + 80 80 + if (programPath === undefined) { 81 81 + io.stderr("Usage: mill run <program.ts> --sync [--json]"); 82 82 + return 1; 83 83 + } 84 84 + 85 85 + if (!argv.includes("--sync")) { 86 86 + io.stderr("v0 currently supports `mill run` only with --sync."); 87 87 + return 1; 88 88 + } 89 89 + 90 90 + const output = await runProgramSync({ 91 91 + defaults: defaultConfig, 92 92 + programPath, 93 93 + cwd: options.cwd, 94 94 + homeDirectory: options.homeDirectory, 95 95 + runsDirectory: readFlagValue(argv, "--runs-dir") ?? options.runsDirectory, 96 96 + driverName: readFlagValue(argv, "--driver"), 97 97 + pathExists: options.pathExists, 98 98 + loadConfigOverrides: options.loadConfigOverrides, 99 99 + }); 100 100 + 101 101 + if (argv.includes("--json")) { 102 102 + io.stdout(JSON.stringify(output)); 103 103 + return 0; 104 104 + } 105 105 + 106 106 + io.stdout(`run ${output.run.id} -> ${output.run.status}`); 107 107 + return 0; 108 108 + }; 109 109 + 110 110 + const statusCommand = async ( 111 111 + argv: ReadonlyArray<string>, 112 112 + options: RunCliOptions, 113 113 + io: CliIo, 114 114 + ): Promise<number> => { 115 115 + const runId = argv[0]; 116 116 + 117 117 + if (runId === undefined) { 118 118 + io.stderr("Usage: mill status <runId> [--json]"); 119 119 + return 1; 120 120 + } 121 121 + 122 122 + const output = await getRunStatus({ 123 123 + defaults: defaultConfig, 124 124 + runId, 125 125 + cwd: options.cwd, 126 126 + homeDirectory: options.homeDirectory, 127 127 + runsDirectory: readFlagValue(argv, "--runs-dir") ?? options.runsDirectory, 128 128 + driverName: readFlagValue(argv, "--driver"), 129 129 + pathExists: options.pathExists, 130 130 + loadConfigOverrides: options.loadConfigOverrides, 131 131 + }); 132 132 + 133 133 + if (argv.includes("--json")) { 134 134 + io.stdout(JSON.stringify(output)); 135 135 + return 0; 136 136 + } 137 137 + 138 138 + io.stdout(`run ${output.id} status=${output.status}`); 139 139 + return 0; 140 140 + }; 141 141 + 142 142 + const waitCommand = async ( 143 143 + argv: ReadonlyArray<string>, 144 144 + options: RunCliOptions, 145 145 + io: CliIo, 146 146 + ): Promise<number> => { 147 147 + const runId = argv[0]; 148 148 + const timeoutSeconds = parseTimeoutSeconds(argv); 149 149 + const isJson = argv.includes("--json"); 150 150 + 151 151 + if (runId === undefined || timeoutSeconds === undefined) { 152 152 + io.stderr("Usage: mill wait <runId> --timeout <seconds> [--json]"); 153 153 + return 1; 154 154 + } 155 155 + 156 156 + const [waitResult] = await Promise.allSettled([ 157 157 + waitForRun({ 158 158 + defaults: defaultConfig, 159 159 + runId, 160 160 + timeoutSeconds, 161 161 + cwd: options.cwd, 162 162 + homeDirectory: options.homeDirectory, 163 163 + runsDirectory: readFlagValue(argv, "--runs-dir") ?? options.runsDirectory, 164 164 + driverName: readFlagValue(argv, "--driver"), 165 165 + pathExists: options.pathExists, 166 166 + loadConfigOverrides: options.loadConfigOverrides, 167 167 + }), 168 168 + ]); 169 169 + 170 170 + if (waitResult.status === "fulfilled") { 171 171 + if (isJson) { 172 172 + io.stdout(JSON.stringify(waitResult.value)); 173 173 + return 0; 174 174 + } 175 175 + 176 176 + io.stdout(`run ${waitResult.value.id} status=${waitResult.value.status}`); 177 177 + return 0; 178 178 + } 179 179 + 180 180 + const waitError = waitResult.reason as { 181 181 + readonly _tag?: string; 182 182 + readonly message?: string; 183 183 + }; 184 184 + 185 185 + if (waitError._tag === "WaitTimeoutError") { 186 186 + const message = `Timeout waiting for run ${runId} after ${timeoutSeconds}s.`; 187 187 + 188 188 + if (isJson) { 189 189 + io.stdout( 190 190 + JSON.stringify({ 191 191 + ok: false, 192 192 + error: { 193 193 + _tag: "WaitTimeoutError", 194 194 + runId, 195 195 + timeoutSeconds, 196 196 + message, 197 197 + }, 198 198 + }), 199 199 + ); 200 200 + return 2; 201 201 + } 202 202 + 203 203 + io.stderr(message); 204 204 + return 2; 205 205 + } 206 206 + 207 207 + const fallbackMessage = waitError.message ?? String(waitResult.reason); 208 208 + 209 209 + if (isJson) { 210 210 + io.stdout( 211 211 + JSON.stringify({ 212 212 + ok: false, 213 213 + error: { 214 214 + _tag: "WaitError", 215 215 + runId, 216 216 + timeoutSeconds, 217 217 + message: fallbackMessage, 218 218 + }, 219 219 + }), 220 220 + ); 221 221 + return 1; 222 222 + } 223 223 + 224 224 + io.stderr(fallbackMessage); 225 225 + return 1; 226 226 + }; 227 227 + 43 228 export const runCli = async ( 44 229 argv: ReadonlyArray<string>, 45 230 options?: RunCliOptions, ··· 73 258 return 0; 74 259 } 75 260 76 76 - io.stderr("v0 scaffold: only help/discovery is wired in this foundation stage."); 77 77 - return 0; 261 261 + if (argv[0] === "run") { 262 262 + return runSyncCommand(argv.slice(1), options ?? {}, io); 263 263 + } 264 264 + 265 265 + if (argv[0] === "status") { 266 266 + return statusCommand(argv.slice(1), options ?? {}, io); 267 267 + } 268 268 + 269 269 + if (argv[0] === "wait") { 270 270 + return waitCommand(argv.slice(1), options ?? {}, io); 271 271 + } 272 272 + 273 273 + io.stderr(`Unknown command: ${argv[0]}`); 274 274 + return 1; 78 275 };

+209 -7

packages/cli/src/public/index.e2e.test.ts

reviewed

··· 1 1 import { describe, expect, it } from "bun:test"; 2 2 + import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; 3 3 + import { tmpdir } from "node:os"; 4 4 + import { join } from "node:path"; 2 5 import * as Command from "@effect/platform/Command"; 3 6 import * as Schema from "@effect/schema/Schema"; 4 7 import * as BunContext from "@effect/platform-bun/BunContext"; ··· 33 36 }), 34 37 ); 35 38 39 39 + const RunSyncEnvelope = Schema.parseJson( 40 40 + Schema.Struct({ 41 41 + run: Schema.Struct({ 42 42 + id: Schema.String, 43 43 + status: Schema.String, 44 44 + paths: Schema.Struct({ 45 45 + runDir: Schema.String, 46 46 + runFile: Schema.String, 47 47 + eventsFile: Schema.String, 48 48 + resultFile: Schema.String, 49 49 + }), 50 50 + }), 51 51 + result: Schema.Struct({ 52 52 + runId: Schema.String, 53 53 + status: Schema.String, 54 54 + spawns: Schema.Array( 55 55 + Schema.Struct({ 56 56 + text: Schema.String, 57 57 + sessionRef: Schema.String, 58 58 + agent: Schema.String, 59 59 + model: Schema.String, 60 60 + driver: Schema.String, 61 61 + exitCode: Schema.Number, 62 62 + }), 63 63 + ), 64 64 + }), 65 65 + }), 66 66 + ); 67 67 + 68 68 + const StatusEnvelope = Schema.parseJson( 69 69 + Schema.Struct({ 70 70 + id: Schema.String, 71 71 + status: Schema.String, 72 72 + }), 73 73 + ); 74 74 + 75 75 + const commandOutput = (command: Command.Command): Promise<string> => 76 76 + Runtime.runPromise(runtime)(Effect.provide(Command.string(command), BunContext.layer)); 77 77 + 78 78 + const commandExitCode = (command: Command.Command): Promise<number> => 79 79 + Runtime.runPromise(runtime)(Effect.provide(Command.exitCode(command), BunContext.layer)); 80 80 + 36 81 describe("mill --help --json (e2e)", () => { 37 82 it("returns discovery contract payload on stdout", async () => { 38 38 - const output = await Runtime.runPromise(runtime)( 39 39 - Effect.provide( 40 40 - Command.string( 41 41 - Command.make("bun", "run", "packages/cli/src/bin/mill.ts", "--help", "--json"), 42 42 - ), 43 43 - BunContext.layer, 44 44 - ), 83 83 + const output = await commandOutput( 84 84 + Command.make("bun", "run", "packages/cli/src/bin/mill.ts", "--help", "--json"), 45 85 ); 46 86 47 87 const payload = Schema.decodeUnknownSync(DiscoveryEnvelope)(output); ··· 55 95 expect(payload.async.submit).toBe("mill run <program.ts> --json"); 56 96 }); 57 97 }); 98 98 + 99 99 + describe("mill run/status/wait (e2e)", () => { 100 100 + it("executes run --sync and wait --timeout returns terminal result", async () => { 101 101 + const tempDirectory = await mkdtemp(join(tmpdir(), "mill-cli-e2e-")); 102 102 + const runsDirectory = join(tempDirectory, "runs"); 103 103 + const programPath = join(tempDirectory, "program.ts"); 104 104 + 105 105 + await writeFile( 106 106 + programPath, 107 107 + [ 108 108 + "const first = await mill.spawn({", 109 109 + ' agent: "scout",', 110 110 + ' systemPrompt: "You are concise.",', 111 111 + ' prompt: "Inspect repository layout.",', 112 112 + "});", 113 113 + "const second = await mill.spawn({", 114 114 + ' agent: "synth",', 115 115 + ' systemPrompt: "You summarize findings.",', 116 116 + " prompt: first.text,", 117 117 + "});", 118 118 + "globalThis.__millSecondText = second.text;", 119 119 + ].join("\n"), 120 120 + "utf-8", 121 121 + ); 122 122 + 123 123 + try { 124 124 + const runOutput = await commandOutput( 125 125 + Command.make( 126 126 + "bun", 127 127 + "run", 128 128 + "packages/cli/src/bin/mill.ts", 129 129 + "run", 130 130 + programPath, 131 131 + "--sync", 132 132 + "--json", 133 133 + "--runs-dir", 134 134 + runsDirectory, 135 135 + ), 136 136 + ); 137 137 + 138 138 + const runPayload = Schema.decodeUnknownSync(RunSyncEnvelope)(runOutput); 139 139 + expect(runPayload.run.status).toBe("complete"); 140 140 + expect(runPayload.result.status).toBe("complete"); 141 141 + expect(runPayload.result.spawns).toHaveLength(2); 142 142 + 143 143 + const statusOutput = await commandOutput( 144 144 + Command.make( 145 145 + "bun", 146 146 + "run", 147 147 + "packages/cli/src/bin/mill.ts", 148 148 + "status", 149 149 + runPayload.run.id, 150 150 + "--json", 151 151 + "--runs-dir", 152 152 + runsDirectory, 153 153 + ), 154 154 + ); 155 155 + 156 156 + const statusPayload = Schema.decodeUnknownSync(StatusEnvelope)(statusOutput); 157 157 + expect(statusPayload.id).toBe(runPayload.run.id); 158 158 + expect(statusPayload.status).toBe("complete"); 159 159 + 160 160 + const waitOutput = await commandOutput( 161 161 + Command.make( 162 162 + "bun", 163 163 + "run", 164 164 + "packages/cli/src/bin/mill.ts", 165 165 + "wait", 166 166 + runPayload.run.id, 167 167 + "--timeout", 168 168 + "2", 169 169 + "--json", 170 170 + "--runs-dir", 171 171 + runsDirectory, 172 172 + ), 173 173 + ); 174 174 + 175 175 + const waitPayload = Schema.decodeUnknownSync(StatusEnvelope)(waitOutput); 176 176 + expect(waitPayload.id).toBe(runPayload.run.id); 177 177 + expect(waitPayload.status).toBe("complete"); 178 178 + 179 179 + const runFile = await readFile(runPayload.run.paths.runFile, "utf-8"); 180 180 + const eventsFile = await readFile(runPayload.run.paths.eventsFile, "utf-8"); 181 181 + const resultFile = await readFile(runPayload.run.paths.resultFile, "utf-8"); 182 182 + 183 183 + expect(runFile.length).toBeGreaterThan(0); 184 184 + expect(eventsFile.length).toBeGreaterThan(0); 185 185 + expect(resultFile.length).toBeGreaterThan(0); 186 186 + } finally { 187 187 + await rm(tempDirectory, { recursive: true, force: true }); 188 188 + } 189 189 + }); 190 190 + 191 191 + it("wait timeout exits non-zero", async () => { 192 192 + const tempDirectory = await mkdtemp(join(tmpdir(), "mill-cli-wait-timeout-e2e-")); 193 193 + const runsDirectory = join(tempDirectory, "runs"); 194 194 + const runId = `run_timeout_e2e_${crypto.randomUUID()}`; 195 195 + const runDir = join(runsDirectory, runId); 196 196 + const runFile = join(runDir, "run.json"); 197 197 + const eventsFile = join(runDir, "events.ndjson"); 198 198 + const resultFile = join(runDir, "result.json"); 199 199 + 200 200 + await mkdir(runDir, { recursive: true }); 201 201 + await writeFile( 202 202 + runFile, 203 203 + `${JSON.stringify( 204 204 + { 205 205 + id: runId, 206 206 + status: "running", 207 207 + programPath: "/tmp/program.ts", 208 208 + driver: "default", 209 209 + createdAt: "2026-02-23T20:00:00.000Z", 210 210 + updatedAt: "2026-02-23T20:00:00.000Z", 211 211 + paths: { 212 212 + runDir, 213 213 + runFile, 214 214 + eventsFile, 215 215 + resultFile, 216 216 + }, 217 217 + }, 218 218 + null, 219 219 + 2, 220 220 + )}\n`, 221 221 + "utf-8", 222 222 + ); 223 223 + await writeFile( 224 224 + eventsFile, 225 225 + `${JSON.stringify({ 226 226 + schemaVersion: 1, 227 227 + runId, 228 228 + sequence: 1, 229 229 + timestamp: "2026-02-23T20:00:00.000Z", 230 230 + type: "run:start", 231 231 + payload: { 232 232 + programPath: "/tmp/program.ts", 233 233 + }, 234 234 + })}\n`, 235 235 + "utf-8", 236 236 + ); 237 237 + 238 238 + try { 239 239 + const exitCode = await commandExitCode( 240 240 + Command.make( 241 241 + "bun", 242 242 + "run", 243 243 + "packages/cli/src/bin/mill.ts", 244 244 + "wait", 245 245 + runId, 246 246 + "--timeout", 247 247 + "1", 248 248 + "--json", 249 249 + "--runs-dir", 250 250 + runsDirectory, 251 251 + ), 252 252 + ); 253 253 + 254 254 + expect(exitCode).toBe(2); 255 255 + } finally { 256 256 + await rm(tempDirectory, { recursive: true, force: true }); 257 257 + } 258 258 + }); 259 259 + });

+54

packages/core/src/domain/event.schema.test.ts

reviewed

··· 1 1 + import { describe, expect, it } from "bun:test"; 2 2 + import { decodeMillEventJsonSync } from "./event.schema"; 3 3 + 4 4 + describe("MillEvent schema union", () => { 5 5 + it("decodes persisted spawn:complete events with required envelope fields", () => { 6 6 + const event = decodeMillEventJsonSync( 7 7 + JSON.stringify({ 8 8 + schemaVersion: 1, 9 9 + runId: "run_test_01", 10 10 + sequence: 5, 11 11 + timestamp: "2026-02-23T20:00:00.000Z", 12 12 + type: "spawn:complete", 13 13 + payload: { 14 14 + spawnId: "spawn_01", 15 15 + result: { 16 16 + text: "done", 17 17 + sessionRef: "session/pi/test", 18 18 + agent: "scout", 19 19 + model: "openai/gpt-5.3-codex", 20 20 + driver: "pi", 21 21 + exitCode: 0, 22 22 + }, 23 23 + }, 24 24 + }), 25 25 + ); 26 26 + 27 27 + expect(event.schemaVersion).toBe(1); 28 28 + expect(event.runId).toBe("run_test_01"); 29 29 + expect(event.sequence).toBe(5); 30 30 + expect(event.timestamp).toBe("2026-02-23T20:00:00.000Z"); 31 31 + expect(event.type).toBe("spawn:complete"); 32 32 + 33 33 + if (event.type === "spawn:complete") { 34 34 + expect(event.payload.result.sessionRef.length).toBeGreaterThan(0); 35 35 + } 36 36 + }); 37 37 + 38 38 + it("fails decoding unknown schemaVersion values", () => { 39 39 + expect(() => 40 40 + decodeMillEventJsonSync( 41 41 + JSON.stringify({ 42 42 + schemaVersion: 2, 43 43 + runId: "run_test_01", 44 44 + sequence: 1, 45 45 + timestamp: "2026-02-23T20:00:00.000Z", 46 46 + type: "run:start", 47 47 + payload: { 48 48 + programPath: "/tmp/program.ts", 49 49 + }, 50 50 + }), 51 51 + ), 52 52 + ).toThrow(); 53 53 + }); 54 54 + });

+155

packages/core/src/domain/event.schema.ts

reviewed

··· 1 1 + import * as Schema from "@effect/schema/Schema"; 2 2 + import { 3 3 + RunId, 4 4 + RunResult, 5 5 + RunStatus, 6 6 + SchemaVersion, 7 7 + SpawnId, 8 8 + type RunId as RunIdType, 9 9 + } from "./run.schema"; 10 10 + import { SpawnOptions, SpawnResult } from "./spawn.schema"; 11 11 + 12 12 + const EventEnvelope = { 13 13 + schemaVersion: SchemaVersion, 14 14 + runId: RunId, 15 15 + sequence: Schema.Int, 16 16 + timestamp: Schema.String, 17 17 + } as const; 18 18 + 19 19 + export const RunStartEvent = Schema.Struct({ 20 20 + ...EventEnvelope, 21 21 + type: Schema.Literal("run:start"), 22 22 + payload: Schema.Struct({ 23 23 + programPath: Schema.NonEmptyString, 24 24 + }), 25 25 + }); 26 26 + export type RunStartEvent = Schema.Schema.Type<typeof RunStartEvent>; 27 27 + 28 28 + export const RunStatusEvent = Schema.Struct({ 29 29 + ...EventEnvelope, 30 30 + type: Schema.Literal("run:status"), 31 31 + payload: Schema.Struct({ 32 32 + status: RunStatus, 33 33 + }), 34 34 + }); 35 35 + export type RunStatusEvent = Schema.Schema.Type<typeof RunStatusEvent>; 36 36 + 37 37 + export const RunCompleteEvent = Schema.Struct({ 38 38 + ...EventEnvelope, 39 39 + type: Schema.Literal("run:complete"), 40 40 + payload: Schema.Struct({ 41 41 + result: RunResult, 42 42 + }), 43 43 + }); 44 44 + export type RunCompleteEvent = Schema.Schema.Type<typeof RunCompleteEvent>; 45 45 + 46 46 + export const RunFailedEvent = Schema.Struct({ 47 47 + ...EventEnvelope, 48 48 + type: Schema.Literal("run:failed"), 49 49 + payload: Schema.Struct({ 50 50 + message: Schema.String, 51 51 + }), 52 52 + }); 53 53 + export type RunFailedEvent = Schema.Schema.Type<typeof RunFailedEvent>; 54 54 + 55 55 + export const RunCancelledEvent = Schema.Struct({ 56 56 + ...EventEnvelope, 57 57 + type: Schema.Literal("run:cancelled"), 58 58 + payload: Schema.Struct({ 59 59 + reason: Schema.optional(Schema.String), 60 60 + }), 61 61 + }); 62 62 + export type RunCancelledEvent = Schema.Schema.Type<typeof RunCancelledEvent>; 63 63 + 64 64 + export const SpawnStartEvent = Schema.Struct({ 65 65 + ...EventEnvelope, 66 66 + type: Schema.Literal("spawn:start"), 67 67 + payload: Schema.Struct({ 68 68 + spawnId: SpawnId, 69 69 + input: SpawnOptions, 70 70 + }), 71 71 + }); 72 72 + export type SpawnStartEvent = Schema.Schema.Type<typeof SpawnStartEvent>; 73 73 + 74 74 + export const SpawnMilestoneEvent = Schema.Struct({ 75 75 + ...EventEnvelope, 76 76 + type: Schema.Literal("spawn:milestone"), 77 77 + payload: Schema.Struct({ 78 78 + spawnId: SpawnId, 79 79 + message: Schema.NonEmptyString, 80 80 + }), 81 81 + }); 82 82 + export type SpawnMilestoneEvent = Schema.Schema.Type<typeof SpawnMilestoneEvent>; 83 83 + 84 84 + export const SpawnToolCallEvent = Schema.Struct({ 85 85 + ...EventEnvelope, 86 86 + type: Schema.Literal("spawn:tool_call"), 87 87 + payload: Schema.Struct({ 88 88 + spawnId: SpawnId, 89 89 + toolName: Schema.NonEmptyString, 90 90 + }), 91 91 + }); 92 92 + export type SpawnToolCallEvent = Schema.Schema.Type<typeof SpawnToolCallEvent>; 93 93 + 94 94 + export const SpawnErrorEvent = Schema.Struct({ 95 95 + ...EventEnvelope, 96 96 + type: Schema.Literal("spawn:error"), 97 97 + payload: Schema.Struct({ 98 98 + spawnId: SpawnId, 99 99 + message: Schema.String, 100 100 + }), 101 101 + }); 102 102 + export type SpawnErrorEvent = Schema.Schema.Type<typeof SpawnErrorEvent>; 103 103 + 104 104 + export const SpawnCompleteEvent = Schema.Struct({ 105 105 + ...EventEnvelope, 106 106 + type: Schema.Literal("spawn:complete"), 107 107 + payload: Schema.Struct({ 108 108 + spawnId: SpawnId, 109 109 + result: SpawnResult, 110 110 + }), 111 111 + }); 112 112 + export type SpawnCompleteEvent = Schema.Schema.Type<typeof SpawnCompleteEvent>; 113 113 + 114 114 + export const SpawnCancelledEvent = Schema.Struct({ 115 115 + ...EventEnvelope, 116 116 + type: Schema.Literal("spawn:cancelled"), 117 117 + payload: Schema.Struct({ 118 118 + spawnId: SpawnId, 119 119 + reason: Schema.optional(Schema.String), 120 120 + }), 121 121 + }); 122 122 + export type SpawnCancelledEvent = Schema.Schema.Type<typeof SpawnCancelledEvent>; 123 123 + 124 124 + export const MillEvent = Schema.Union( 125 125 + RunStartEvent, 126 126 + RunStatusEvent, 127 127 + RunCompleteEvent, 128 128 + RunFailedEvent, 129 129 + RunCancelledEvent, 130 130 + SpawnStartEvent, 131 131 + SpawnMilestoneEvent, 132 132 + SpawnToolCallEvent, 133 133 + SpawnErrorEvent, 134 134 + SpawnCompleteEvent, 135 135 + SpawnCancelledEvent, 136 136 + ); 137 137 + export type MillEvent = Schema.Schema.Type<typeof MillEvent>; 138 138 + 139 139 + export const MillEventJson = Schema.parseJson(MillEvent); 140 140 + 141 141 + export const decodeMillEventJson = Schema.decodeUnknown(MillEventJson); 142 142 + export const decodeMillEventJsonSync = Schema.decodeUnknownSync(MillEventJson); 143 143 + 144 144 + export const encodeMillEventJson = (event: MillEvent): string => JSON.stringify(event); 145 145 + 146 146 + export const makeEventEnvelope = ( 147 147 + runId: RunIdType, 148 148 + sequence: number, 149 149 + timestamp: string, 150 150 + ): Pick<MillEvent, "schemaVersion" | "runId" | "sequence" | "timestamp"> => ({ 151 151 + schemaVersion: 1, 152 152 + runId, 153 153 + sequence, 154 154 + timestamp, 155 155 + });

+52

packages/core/src/domain/run.schema.ts

reviewed

··· 1 1 import * as Schema from "@effect/schema/Schema"; 2 2 + import { SpawnResult } from "./spawn.schema"; 3 3 + 4 4 + export const SchemaVersion = Schema.Literal(1); 5 5 + export type SchemaVersion = Schema.Schema.Type<typeof SchemaVersion>; 2 6 3 7 export const RunId = Schema.String.pipe(Schema.brand("RunId")); 4 8 export type RunId = Schema.Schema.Type<typeof RunId>; ··· 9 13 export const RunStatus = Schema.Literal("pending", "running", "complete", "failed", "cancelled"); 10 14 export type RunStatus = Schema.Schema.Type<typeof RunStatus>; 11 15 16 16 + export const RunTerminalStatus = Schema.Literal("complete", "failed", "cancelled"); 17 17 + export type RunTerminalStatus = Schema.Schema.Type<typeof RunTerminalStatus>; 18 18 + 19 19 + export const RunPaths = Schema.Struct({ 20 20 + runDir: Schema.NonEmptyString, 21 21 + runFile: Schema.NonEmptyString, 22 22 + eventsFile: Schema.NonEmptyString, 23 23 + resultFile: Schema.NonEmptyString, 24 24 + }); 25 25 + export type RunPaths = Schema.Schema.Type<typeof RunPaths>; 26 26 + 12 27 export const RunRecord = Schema.Struct({ 13 28 id: RunId, 14 29 status: RunStatus, 30 30 + programPath: Schema.NonEmptyString, 31 31 + driver: Schema.NonEmptyString, 32 32 + createdAt: Schema.String, 33 33 + updatedAt: Schema.String, 34 34 + paths: RunPaths, 15 35 }); 16 36 export type RunRecord = Schema.Schema.Type<typeof RunRecord>; 37 37 + 38 38 + export const RunResult = Schema.Struct({ 39 39 + runId: RunId, 40 40 + status: RunTerminalStatus, 41 41 + startedAt: Schema.String, 42 42 + completedAt: Schema.String, 43 43 + spawns: Schema.Array(SpawnResult), 44 44 + programResult: Schema.optional(Schema.String), 45 45 + errorMessage: Schema.optional(Schema.String), 46 46 + }); 47 47 + export type RunResult = Schema.Schema.Type<typeof RunResult>; 48 48 + 49 49 + export const RunSyncOutput = Schema.Struct({ 50 50 + run: RunRecord, 51 51 + result: RunResult, 52 52 + }); 53 53 + export type RunSyncOutput = Schema.Schema.Type<typeof RunSyncOutput>; 54 54 + 55 55 + export const RunRecordJson = Schema.parseJson(RunRecord); 56 56 + export const RunResultJson = Schema.parseJson(RunResult); 57 57 + export const RunSyncOutputJson = Schema.parseJson(RunSyncOutput); 58 58 + 59 59 + export const decodeRunId = Schema.decodeUnknown(RunId); 60 60 + export const decodeRunIdSync = Schema.decodeUnknownSync(RunId); 61 61 + export const decodeSpawnId = Schema.decodeUnknown(SpawnId); 62 62 + export const decodeSpawnIdSync = Schema.decodeUnknownSync(SpawnId); 63 63 + export const decodeRunRecordJson = Schema.decodeUnknown(RunRecordJson); 64 64 + export const decodeRunRecordJsonSync = Schema.decodeUnknownSync(RunRecordJson); 65 65 + export const decodeRunResultJson = Schema.decodeUnknown(RunResultJson); 66 66 + export const decodeRunResultJsonSync = Schema.decodeUnknownSync(RunResultJson); 67 67 + export const decodeRunSyncOutputJson = Schema.decodeUnknown(RunSyncOutputJson); 68 68 + export const decodeRunSyncOutputJsonSync = Schema.decodeUnknownSync(RunSyncOutputJson);

+5

packages/core/src/domain/spawn.schema.ts

reviewed

··· 21 21 }); 22 22 23 23 export type SpawnResult = Schema.Schema.Type<typeof SpawnResult>; 24 24 + 25 25 + export const decodeSpawnOptions = Schema.decodeUnknown(SpawnOptions); 26 26 + export const decodeSpawnOptionsSync = Schema.decodeUnknownSync(SpawnOptions); 27 27 + export const decodeSpawnResult = Schema.decodeUnknown(SpawnResult); 28 28 + export const decodeSpawnResultSync = Schema.decodeUnknownSync(SpawnResult);

+1

packages/core/src/engine.effect.ts

reviewed

··· 1 1 + export * from "./internal/engine.effect";

+1

packages/core/src/event.schema.ts

reviewed

··· 1 1 + export * from "./domain/event.schema";

+277

packages/core/src/internal/engine.effect.test.ts

reviewed

··· 1 1 + import { describe, expect, it } from "bun:test"; 2 2 + import { mkdtemp, readFile, rm } from "node:fs/promises"; 3 3 + import { tmpdir } from "node:os"; 4 4 + import { join } from "node:path"; 5 5 + import { setTimeout as delay } from "node:timers/promises"; 6 6 + import { Effect } from "effect"; 7 7 + import { decodeMillEventJsonSync, type MillEvent } from "../domain/event.schema"; 8 8 + import { decodeRunIdSync } from "../domain/run.schema"; 9 9 + import { runWithBunContext } from "../public/test-runtime.api"; 10 10 + import type { DriverRuntime } from "../public/types"; 11 11 + import { makeMillEngine } from "./engine.effect"; 12 12 + import { makeRunStore } from "./run-store.effect"; 13 13 + 14 14 + const testDriver: DriverRuntime = { 15 15 + name: "test-driver", 16 16 + spawn: (input) => 17 17 + Effect.succeed({ 18 18 + events: [ 19 19 + { 20 20 + type: "milestone", 21 21 + message: `spawned:${input.agent}`, 22 22 + }, 23 23 + ], 24 24 + result: { 25 25 + text: `driver:${input.prompt}`, 26 26 + sessionRef: `session/${input.agent}`, 27 27 + agent: input.agent, 28 28 + model: input.model, 29 29 + driver: "test-driver", 30 30 + exitCode: 0, 31 31 + }, 32 32 + }), 33 33 + }; 34 34 + 35 35 + const parseEvents = (content: string): ReadonlyArray<MillEvent> => 36 36 + content 37 37 + .split("\n") 38 38 + .map((line) => line.trim()) 39 39 + .filter((line) => line.length > 0) 40 40 + .map((line) => decodeMillEventJsonSync(line)); 41 41 + 42 42 + const runTerminalTypes = new Set(["run:complete", "run:failed", "run:cancelled"]); 43 43 + const spawnTerminalTypes = new Set(["spawn:complete", "spawn:error", "spawn:cancelled"]); 44 44 + 45 45 + describe("MillEngine sync lifecycle", () => { 46 46 + it("persists deterministic run/start/spawn/complete lifecycle", async () => { 47 47 + const runsDirectory = await mkdtemp(join(tmpdir(), "mill-engine-")); 48 48 + const runId = decodeRunIdSync(`run_${crypto.randomUUID()}`); 49 49 + 50 50 + const engine = makeMillEngine({ 51 51 + runsDirectory, 52 52 + defaultModel: "openai/gpt-5.3-codex", 53 53 + driverName: "default", 54 54 + driver: testDriver, 55 55 + }); 56 56 + 57 57 + try { 58 58 + const output = await runWithBunContext( 59 59 + engine.runSync({ 60 60 + runId, 61 61 + programPath: "/tmp/program.ts", 62 62 + executeProgram: (spawn) => 63 63 + Effect.gen(function* () { 64 64 + const result = yield* spawn({ 65 65 + agent: "scout", 66 66 + systemPrompt: "You are concise.", 67 67 + prompt: "Summarize this file.", 68 68 + }); 69 69 + 70 70 + expect(result.sessionRef.length).toBeGreaterThan(0); 71 71 + }), 72 72 + }), 73 73 + ); 74 74 + 75 75 + expect(output.result.status).toBe("complete"); 76 76 + expect(output.result.spawns).toHaveLength(1); 77 77 + expect(output.run.status).toBe("complete"); 78 78 + 79 79 + const status = await runWithBunContext(engine.status(runId)); 80 80 + expect(status.status).toBe("complete"); 81 81 + 82 82 + const eventsContent = await readFile(output.run.paths.eventsFile, "utf-8"); 83 83 + const events = parseEvents(eventsContent); 84 84 + 85 85 + expect(events.length).toBeGreaterThan(0); 86 86 + 87 87 + const spawnComplete = events.find((event) => event.type === "spawn:complete"); 88 88 + expect(spawnComplete).toBeDefined(); 89 89 + 90 90 + if (spawnComplete !== undefined && spawnComplete.type === "spawn:complete") { 91 91 + expect(spawnComplete.payload.result.sessionRef.length).toBeGreaterThan(0); 92 92 + } 93 93 + 94 94 + for (const event of events) { 95 95 + expect(event.schemaVersion).toBe(1); 96 96 + expect(event.runId).toBe(runId); 97 97 + expect(event.sequence).toBeGreaterThan(0); 98 98 + expect(event.timestamp.length).toBeGreaterThan(0); 99 99 + } 100 100 + 101 101 + const runTerminalCount = events.filter((event) => runTerminalTypes.has(event.type)).length; 102 102 + expect(runTerminalCount).toBe(1); 103 103 + 104 104 + const spawnIds = events 105 105 + .filter((event): event is Extract<MillEvent, { type: "spawn:start" }> => 106 106 + event.type === "spawn:start", 107 107 + ) 108 108 + .map((event) => event.payload.spawnId); 109 109 + 110 110 + for (const spawnId of spawnIds) { 111 111 + const terminalCount = events.filter((event) => { 112 112 + if (!spawnTerminalTypes.has(event.type)) { 113 113 + return false; 114 114 + } 115 115 + 116 116 + if (event.type === "spawn:complete") { 117 117 + return event.payload.spawnId === spawnId; 118 118 + } 119 119 + 120 120 + if (event.type === "spawn:error") { 121 121 + return event.payload.spawnId === spawnId; 122 122 + } 123 123 + 124 124 + if (event.type === "spawn:cancelled") { 125 125 + return event.payload.spawnId === spawnId; 126 126 + } 127 127 + 128 128 + return false; 129 129 + }).length; 130 130 + 131 131 + expect(terminalCount).toBe(1); 132 132 + } 133 133 + } finally { 134 134 + await rm(runsDirectory, { recursive: true, force: true }); 135 135 + } 136 136 + }); 137 137 + 138 138 + it("wait resolves when terminal event arrives after wait subscription starts", async () => { 139 139 + const runsDirectory = await mkdtemp(join(tmpdir(), "mill-engine-wait-live-")); 140 140 + const runId = decodeRunIdSync(`run_${crypto.randomUUID()}`); 141 141 + 142 142 + const store = makeRunStore({ runsDirectory }); 143 143 + const engine = makeMillEngine({ 144 144 + runsDirectory, 145 145 + defaultModel: "openai/gpt-5.3-codex", 146 146 + driverName: "default", 147 147 + driver: testDriver, 148 148 + }); 149 149 + 150 150 + try { 151 151 + await runWithBunContext( 152 152 + store.create({ 153 153 + runId, 154 154 + programPath: "/tmp/program.ts", 155 155 + driver: "default", 156 156 + timestamp: "2026-02-23T20:00:00.000Z", 157 157 + }), 158 158 + ); 159 159 + 160 160 + await runWithBunContext( 161 161 + store.appendEvent(runId, { 162 162 + schemaVersion: 1, 163 163 + runId, 164 164 + sequence: 1, 165 165 + timestamp: "2026-02-23T20:00:00.000Z", 166 166 + type: "run:start", 167 167 + payload: { 168 168 + programPath: "/tmp/program.ts", 169 169 + }, 170 170 + }), 171 171 + ); 172 172 + 173 173 + await runWithBunContext( 174 174 + store.appendEvent(runId, { 175 175 + schemaVersion: 1, 176 176 + runId, 177 177 + sequence: 2, 178 178 + timestamp: "2026-02-23T20:00:01.000Z", 179 179 + type: "run:status", 180 180 + payload: { 181 181 + status: "running", 182 182 + }, 183 183 + }), 184 184 + ); 185 185 + 186 186 + const appendTerminal = (async () => { 187 187 + await delay(50); 188 188 + 189 189 + await runWithBunContext( 190 190 + store.appendEvent(runId, { 191 191 + schemaVersion: 1, 192 192 + runId, 193 193 + sequence: 3, 194 194 + timestamp: "2026-02-23T20:00:02.000Z", 195 195 + type: "run:complete", 196 196 + payload: { 197 197 + result: { 198 198 + runId, 199 199 + status: "complete", 200 200 + startedAt: "2026-02-23T20:00:00.000Z", 201 201 + completedAt: "2026-02-23T20:00:02.000Z", 202 202 + spawns: [], 203 203 + }, 204 204 + }, 205 205 + }), 206 206 + ); 207 207 + 208 208 + await runWithBunContext( 209 209 + store.setResult( 210 210 + runId, 211 211 + { 212 212 + runId, 213 213 + status: "complete", 214 214 + startedAt: "2026-02-23T20:00:00.000Z", 215 215 + completedAt: "2026-02-23T20:00:02.000Z", 216 216 + spawns: [], 217 217 + }, 218 218 + "2026-02-23T20:00:02.000Z", 219 219 + ), 220 220 + ); 221 221 + })(); 222 222 + 223 223 + const waitedRun = await runWithBunContext(engine.wait(runId, "2 seconds")); 224 224 + 225 225 + expect(waitedRun.status).toBe("complete"); 226 226 + await appendTerminal; 227 227 + } finally { 228 228 + await rm(runsDirectory, { recursive: true, force: true }); 229 229 + } 230 230 + }); 231 231 + 232 232 + it("wait fails with typed timeout error when no terminal event arrives", async () => { 233 233 + const runsDirectory = await mkdtemp(join(tmpdir(), "mill-engine-wait-timeout-")); 234 234 + const runId = decodeRunIdSync(`run_${crypto.randomUUID()}`); 235 235 + 236 236 + const store = makeRunStore({ runsDirectory }); 237 237 + const engine = makeMillEngine({ 238 238 + runsDirectory, 239 239 + defaultModel: "openai/gpt-5.3-codex", 240 240 + driverName: "default", 241 241 + driver: testDriver, 242 242 + }); 243 243 + 244 244 + try { 245 245 + await runWithBunContext( 246 246 + store.create({ 247 247 + runId, 248 248 + programPath: "/tmp/program.ts", 249 249 + driver: "default", 250 250 + timestamp: "2026-02-23T20:00:00.000Z", 251 251 + }), 252 252 + ); 253 253 + 254 254 + await runWithBunContext( 255 255 + store.appendEvent(runId, { 256 256 + schemaVersion: 1, 257 257 + runId, 258 258 + sequence: 1, 259 259 + timestamp: "2026-02-23T20:00:00.000Z", 260 260 + type: "run:start", 261 261 + payload: { 262 262 + programPath: "/tmp/program.ts", 263 263 + }, 264 264 + }), 265 265 + ); 266 266 + 267 267 + const waitError = await runWithBunContext(Effect.flip(engine.wait(runId, 40))); 268 268 + 269 269 + expect(waitError).toMatchObject({ 270 270 + _tag: "WaitTimeoutError", 271 271 + runId, 272 272 + }); 273 273 + } finally { 274 274 + await rm(runsDirectory, { recursive: true, force: true }); 275 275 + } 276 276 + }); 277 277 + });

+499 -23

packages/core/src/internal/engine.effect.ts

reviewed

··· 1 1 - import { Data, Effect } from "effect"; 2 2 - import type { RunId, RunRecord } from "../domain/run.schema"; 3 3 - import type { SpawnOptions, SpawnResult } from "../domain/spawn.schema"; 1 1 + import { Cause, Clock, Data, Effect, Exit, Ref } from "effect"; 2 2 + import { 3 3 + makeEventEnvelope, 4 4 + type MillEvent, 5 5 + type SpawnCompleteEvent, 6 6 + type SpawnErrorEvent, 7 7 + type SpawnMilestoneEvent, 8 8 + type SpawnStartEvent, 9 9 + type SpawnToolCallEvent, 10 10 + } from "../domain/event.schema"; 11 11 + import { 12 12 + decodeSpawnIdSync, 13 13 + type RunId, 14 14 + type RunResult, 15 15 + type RunSyncOutput, 16 16 + } from "../domain/run.schema"; 17 17 + import { decodeSpawnResult, type SpawnOptions, type SpawnResult } from "../domain/spawn.schema"; 18 18 + import type { DriverRuntime } from "../public/types"; 19 19 + import { 20 20 + LifecycleInvariantError, 21 21 + applyLifecycleTransition, 22 22 + initialLifecycleGuardState, 23 23 + isRunTerminalEvent, 24 24 + type LifecycleGuardState, 25 25 + } from "./lifecycle-guard.effect"; 26 26 + import { 27 27 + PersistenceError, 28 28 + RunNotFoundError, 29 29 + makeRunStore, 30 30 + type RunStore, 31 31 + } from "./run-store.effect"; 4 32 5 33 export class ConfigError extends Data.TaggedError("ConfigError")<{ message: string }> {} 6 6 - 7 7 - export class RunNotFoundError extends Data.TaggedError("RunNotFoundError")<{ runId: string }> {} 8 34 9 35 export class ProgramExecutionError extends Data.TaggedError("ProgramExecutionError")<{ 10 36 runId: string; 11 37 message: string; 12 38 }> {} 13 39 14 14 - export class PersistenceError extends Data.TaggedError("PersistenceError")<{ 15 15 - path: string; 40 40 + export class WaitTimeoutError extends Data.TaggedError("WaitTimeoutError")<{ 41 41 + runId: string; 42 42 + timeoutMillis: number; 16 43 message: string; 17 44 }> {} 18 45 46 46 + export interface RunSyncInput { 47 47 + readonly runId: RunId; 48 48 + readonly programPath: string; 49 49 + readonly executeProgram: ( 50 50 + spawn: ( 51 51 + input: SpawnOptions, 52 52 + ) => Effect.Effect< 53 53 + SpawnResult, 54 54 + ProgramExecutionError | PersistenceError | LifecycleInvariantError 55 55 + >, 56 56 + ) => Effect.Effect<unknown, ProgramExecutionError>; 57 57 + } 58 58 + 19 59 export interface MillEngine { 20 20 - readonly submit: ( 21 21 - input: SpawnOptions, 22 22 - ) => Effect.Effect<SpawnResult, ConfigError | PersistenceError | ProgramExecutionError>; 23 23 - readonly status: (runId: RunId) => Effect.Effect<RunRecord, RunNotFoundError | PersistenceError>; 60 60 + readonly runSync: ( 61 61 + input: RunSyncInput, 62 62 + ) => Effect.Effect< 63 63 + RunSyncOutput, 64 64 + ConfigError | PersistenceError | ProgramExecutionError | LifecycleInvariantError 65 65 + >; 66 66 + readonly status: ( 67 67 + runId: RunId, 68 68 + ) => Effect.Effect<RunSyncOutput["run"], RunNotFoundError | PersistenceError>; 69 69 + readonly wait: ( 70 70 + runId: RunId, 71 71 + timeout: number | string, 72 72 + ) => Effect.Effect< 73 73 + RunSyncOutput["run"], 74 74 + RunNotFoundError | PersistenceError | LifecycleInvariantError | WaitTimeoutError 75 75 + >; 76 76 + } 77 77 + 78 78 + export interface MakeMillEngineInput { 79 79 + readonly runsDirectory: string; 80 80 + readonly driverName: string; 81 81 + readonly defaultModel: string; 82 82 + readonly driver: DriverRuntime; 24 83 } 25 84 26 26 - const buildNoopResult = (input: SpawnOptions): SpawnResult => ({ 27 27 - text: `noop response for ${input.agent}`, 28 28 - sessionRef: "session/noop", 29 29 - agent: input.agent, 30 30 - model: input.model ?? "openai/gpt-5.3-codex", 31 31 - driver: "default", 32 32 - exitCode: 0, 33 33 - }); 85 85 + const toIsoTimestamp = Effect.map(Clock.currentTimeMillis, (millis) => 86 86 + new Date(millis).toISOString(), 87 87 + ); 88 88 + 89 89 + const toMessage = (error: unknown): string => { 90 90 + if (error instanceof Error) { 91 91 + return error.message; 92 92 + } 93 93 + 94 94 + return String(error); 95 95 + }; 96 96 + 97 97 + const nextSequence = (sequenceRef: Ref.Ref<number>): Effect.Effect<number> => 98 98 + Ref.updateAndGet(sequenceRef, (current) => current + 1); 99 99 + 100 100 + const appendTier1Event = ( 101 101 + lifecycleStateRef: Ref.Ref<LifecycleGuardState>, 102 102 + sequenceRef: Ref.Ref<number>, 103 103 + runStore: RunStore, 104 104 + runId: RunId, 105 105 + eventBuilder: (sequence: number, timestamp: string) => MillEvent, 106 106 + ): Effect.Effect<void, PersistenceError | LifecycleInvariantError> => 107 107 + Effect.gen(function* () { 108 108 + const sequence = yield* nextSequence(sequenceRef); 109 109 + const timestamp = yield* toIsoTimestamp; 110 110 + const event = eventBuilder(sequence, timestamp); 111 111 + const lifecycleState = yield* Ref.get(lifecycleStateRef); 112 112 + const nextState = yield* applyLifecycleTransition(lifecycleState, event); 113 113 + 114 114 + yield* Ref.set(lifecycleStateRef, nextState); 115 115 + yield* runStore.appendEvent(runId, event); 116 116 + }); 117 117 + 118 118 + const toTimeoutMillis = (timeout: number | string): number => { 119 119 + if (typeof timeout === "number") { 120 120 + return timeout; 121 121 + } 122 122 + 123 123 + const parsed = Number.parseFloat(timeout); 124 124 + 125 125 + if (Number.isNaN(parsed)) { 126 126 + return 0; 127 127 + } 128 128 + 129 129 + if (timeout.includes("second")) { 130 130 + return Math.max(0, Math.round(parsed * 1000)); 131 131 + } 132 132 + 133 133 + return Math.max(0, Math.round(parsed)); 134 134 + }; 135 135 + 136 136 + const isRunTerminalStatus = (status: RunSyncOutput["run"]["status"]): boolean => 137 137 + status === "complete" || status === "failed" || status === "cancelled"; 138 138 + 139 139 + const waitForRunTerminal = ( 140 140 + runStore: RunStore, 141 141 + runId: RunId, 142 142 + ): Effect.Effect<RunSyncOutput["run"], RunNotFoundError | PersistenceError | LifecycleInvariantError> => 143 143 + Effect.gen(function* () { 144 144 + let observedEvents = 0; 145 145 + let terminalObserved = false; 146 146 + let lifecycleState = initialLifecycleGuardState; 147 147 + 148 148 + while (true) { 149 149 + const events = yield* runStore.readEvents(runId); 150 150 + 151 151 + for (let index = observedEvents; index < events.length; index += 1) { 152 152 + const event = events[index]; 153 153 + 154 154 + lifecycleState = yield* applyLifecycleTransition(lifecycleState, event); 155 155 + observedEvents = index + 1; 156 156 + 157 157 + if (isRunTerminalEvent(event)) { 158 158 + terminalObserved = true; 159 159 + } 160 160 + } 161 161 + 162 162 + if (terminalObserved) { 163 163 + const currentRun = yield* runStore.getRun(runId); 164 164 + 165 165 + if (isRunTerminalStatus(currentRun.status)) { 166 166 + return currentRun; 167 167 + } 168 168 + } 34 169 35 35 - export const makeNoopMillEngine: Effect.Effect<MillEngine> = Effect.succeed({ 36 36 - submit: (input) => Effect.succeed(buildNoopResult(input)), 37 37 - status: () => Effect.fail(new RunNotFoundError({ runId: "missing" })), 38 38 - }); 170 170 + yield* Effect.sleep("25 millis"); 171 171 + } 172 172 + }); 173 173 + 174 174 + const appendSpawnErrorEvent = ( 175 175 + lifecycleStateRef: Ref.Ref<LifecycleGuardState>, 176 176 + sequenceRef: Ref.Ref<number>, 177 177 + runStore: RunStore, 178 178 + runId: RunId, 179 179 + spawnId: string, 180 180 + message: string, 181 181 + ): Effect.Effect<void, PersistenceError | LifecycleInvariantError> => 182 182 + appendTier1Event(lifecycleStateRef, sequenceRef, runStore, runId, (sequence, timestamp) => ({ 183 183 + ...makeEventEnvelope(runId, sequence, timestamp), 184 184 + type: "spawn:error", 185 185 + payload: { 186 186 + spawnId: decodeSpawnIdSync(spawnId), 187 187 + message, 188 188 + }, 189 189 + })); 190 190 + 191 191 + export const makeMillEngine = (input: MakeMillEngineInput): MillEngine => { 192 192 + const runStore = makeRunStore({ 193 193 + runsDirectory: input.runsDirectory, 194 194 + }); 195 195 + 196 196 + return { 197 197 + runSync: (runInput) => 198 198 + Effect.gen(function* () { 199 199 + const lifecycleStateRef = yield* Ref.make(initialLifecycleGuardState); 200 200 + const sequenceRef = yield* Ref.make(0); 201 201 + const spawnCounterRef = yield* Ref.make(0); 202 202 + const spawnResultsRef = yield* Ref.make<ReadonlyArray<SpawnResult>>([]); 203 203 + 204 204 + const startedAt = yield* toIsoTimestamp; 205 205 + 206 206 + yield* runStore.create({ 207 207 + runId: runInput.runId, 208 208 + programPath: runInput.programPath, 209 209 + driver: input.driverName, 210 210 + timestamp: startedAt, 211 211 + }); 212 212 + 213 213 + yield* appendTier1Event( 214 214 + lifecycleStateRef, 215 215 + sequenceRef, 216 216 + runStore, 217 217 + runInput.runId, 218 218 + (sequence, timestamp) => ({ 219 219 + ...makeEventEnvelope(runInput.runId, sequence, timestamp), 220 220 + type: "run:start", 221 221 + payload: { 222 222 + programPath: runInput.programPath, 223 223 + }, 224 224 + }), 225 225 + ); 226 226 + 227 227 + yield* appendTier1Event( 228 228 + lifecycleStateRef, 229 229 + sequenceRef, 230 230 + runStore, 231 231 + runInput.runId, 232 232 + (sequence, timestamp) => ({ 233 233 + ...makeEventEnvelope(runInput.runId, sequence, timestamp), 234 234 + type: "run:status", 235 235 + payload: { 236 236 + status: "running", 237 237 + }, 238 238 + }), 239 239 + ); 240 240 + 241 241 + const spawn = ( 242 242 + spawnInput: SpawnOptions, 243 243 + ): Effect.Effect< 244 244 + SpawnResult, 245 245 + ProgramExecutionError | PersistenceError | LifecycleInvariantError 246 246 + > => 247 247 + Effect.gen(function* () { 248 248 + const nextSpawnCounter = yield* Ref.updateAndGet( 249 249 + spawnCounterRef, 250 250 + (counter) => counter + 1, 251 251 + ); 252 252 + const spawnId = decodeSpawnIdSync(`spawn_${nextSpawnCounter}`); 253 253 + 254 254 + const spawnStartEvent: Omit< 255 255 + SpawnStartEvent, 256 256 + "schemaVersion" | "runId" | "sequence" | "timestamp" 257 257 + > = { 258 258 + type: "spawn:start", 259 259 + payload: { 260 260 + spawnId, 261 261 + input: spawnInput, 262 262 + }, 263 263 + }; 264 264 + 265 265 + yield* appendTier1Event( 266 266 + lifecycleStateRef, 267 267 + sequenceRef, 268 268 + runStore, 269 269 + runInput.runId, 270 270 + (sequence, timestamp) => ({ 271 271 + ...makeEventEnvelope(runInput.runId, sequence, timestamp), 272 272 + ...spawnStartEvent, 273 273 + }), 274 274 + ); 275 275 + 276 276 + const driverOutputExit = yield* Effect.exit( 277 277 + Effect.mapError( 278 278 + input.driver.spawn({ 279 279 + runId: runInput.runId, 280 280 + spawnId, 281 281 + agent: spawnInput.agent, 282 282 + systemPrompt: spawnInput.systemPrompt, 283 283 + prompt: spawnInput.prompt, 284 284 + model: spawnInput.model ?? input.defaultModel, 285 285 + }), 286 286 + (error) => 287 287 + new ProgramExecutionError({ 288 288 + runId: runInput.runId, 289 289 + message: `Driver ${input.driver.name} failed: ${toMessage(error)}`, 290 290 + }), 291 291 + ), 292 292 + ); 293 293 + 294 294 + if (Exit.isFailure(driverOutputExit)) { 295 295 + const failureMessage = Cause.pretty(driverOutputExit.cause); 296 296 + 297 297 + yield* appendSpawnErrorEvent( 298 298 + lifecycleStateRef, 299 299 + sequenceRef, 300 300 + runStore, 301 301 + runInput.runId, 302 302 + spawnId, 303 303 + failureMessage, 304 304 + ); 305 305 + 306 306 + return yield* Effect.fail( 307 307 + new ProgramExecutionError({ 308 308 + runId: runInput.runId, 309 309 + message: failureMessage, 310 310 + }), 311 311 + ); 312 312 + } 313 313 + 314 314 + for (const driverEvent of driverOutputExit.value.events) { 315 315 + if (driverEvent.type === "milestone") { 316 316 + const milestoneEvent: Omit< 317 317 + SpawnMilestoneEvent, 318 318 + "schemaVersion" | "runId" | "sequence" | "timestamp" 319 319 + > = { 320 320 + type: "spawn:milestone", 321 321 + payload: { 322 322 + spawnId, 323 323 + message: driverEvent.message, 324 324 + }, 325 325 + }; 326 326 + 327 327 + yield* appendTier1Event( 328 328 + lifecycleStateRef, 329 329 + sequenceRef, 330 330 + runStore, 331 331 + runInput.runId, 332 332 + (sequence, timestamp) => ({ 333 333 + ...makeEventEnvelope(runInput.runId, sequence, timestamp), 334 334 + ...milestoneEvent, 335 335 + }), 336 336 + ); 337 337 + } 338 338 + 339 339 + if (driverEvent.type === "tool_call") { 340 340 + const toolCallEvent: Omit< 341 341 + SpawnToolCallEvent, 342 342 + "schemaVersion" | "runId" | "sequence" | "timestamp" 343 343 + > = { 344 344 + type: "spawn:tool_call", 345 345 + payload: { 346 346 + spawnId, 347 347 + toolName: driverEvent.toolName, 348 348 + }, 349 349 + }; 350 350 + 351 351 + yield* appendTier1Event( 352 352 + lifecycleStateRef, 353 353 + sequenceRef, 354 354 + runStore, 355 355 + runInput.runId, 356 356 + (sequence, timestamp) => ({ 357 357 + ...makeEventEnvelope(runInput.runId, sequence, timestamp), 358 358 + ...toolCallEvent, 359 359 + }), 360 360 + ); 361 361 + } 362 362 + } 363 363 + 364 364 + const spawnResultExit = yield* Effect.exit( 365 365 + Effect.mapError(decodeSpawnResult(driverOutputExit.value.result), (error) => 366 366 + new ProgramExecutionError({ 367 367 + runId: runInput.runId, 368 368 + message: `Spawn result decode failed: ${toMessage(error)}`, 369 369 + }), 370 370 + ), 371 371 + ); 372 372 + 373 373 + if (Exit.isFailure(spawnResultExit)) { 374 374 + const failureMessage = Cause.pretty(spawnResultExit.cause); 375 375 + 376 376 + yield* appendSpawnErrorEvent( 377 377 + lifecycleStateRef, 378 378 + sequenceRef, 379 379 + runStore, 380 380 + runInput.runId, 381 381 + spawnId, 382 382 + failureMessage, 383 383 + ); 384 384 + 385 385 + return yield* Effect.fail( 386 386 + new ProgramExecutionError({ 387 387 + runId: runInput.runId, 388 388 + message: failureMessage, 389 389 + }), 390 390 + ); 391 391 + } 392 392 + 393 393 + const spawnResult = spawnResultExit.value; 394 394 + const spawnCompleteEvent: Omit< 395 395 + SpawnCompleteEvent, 396 396 + "schemaVersion" | "runId" | "sequence" | "timestamp" 397 397 + > = { 398 398 + type: "spawn:complete", 399 399 + payload: { 400 400 + spawnId, 401 401 + result: spawnResult, 402 402 + }, 403 403 + }; 404 404 + 405 405 + yield* appendTier1Event( 406 406 + lifecycleStateRef, 407 407 + sequenceRef, 408 408 + runStore, 409 409 + runInput.runId, 410 410 + (sequence, timestamp) => ({ 411 411 + ...makeEventEnvelope(runInput.runId, sequence, timestamp), 412 412 + ...spawnCompleteEvent, 413 413 + }), 414 414 + ); 415 415 + 416 416 + yield* Ref.update(spawnResultsRef, (items) => [...items, spawnResult]); 417 417 + 418 418 + return spawnResult; 419 419 + }); 420 420 + 421 421 + const executionExit = yield* Effect.exit(runInput.executeProgram(spawn)); 422 422 + const completedAt = yield* toIsoTimestamp; 423 423 + const spawnResults = yield* Ref.get(spawnResultsRef); 424 424 + 425 425 + if (Exit.isSuccess(executionExit)) { 426 426 + const runResult: RunResult = { 427 427 + runId: runInput.runId, 428 428 + status: "complete", 429 429 + startedAt, 430 430 + completedAt, 431 431 + spawns: spawnResults, 432 432 + programResult: 433 433 + typeof executionExit.value === "string" 434 434 + ? executionExit.value 435 435 + : JSON.stringify(executionExit.value), 436 436 + }; 437 437 + 438 438 + yield* appendTier1Event( 439 439 + lifecycleStateRef, 440 440 + sequenceRef, 441 441 + runStore, 442 442 + runInput.runId, 443 443 + (sequence, timestamp) => ({ 444 444 + ...makeEventEnvelope(runInput.runId, sequence, timestamp), 445 445 + type: "run:complete", 446 446 + payload: { 447 447 + result: runResult, 448 448 + }, 449 449 + }), 450 450 + ); 451 451 + 452 452 + yield* runStore.setResult(runInput.runId, runResult, completedAt); 453 453 + 454 454 + const completedRun = yield* runStore.getRun(runInput.runId); 455 455 + 456 456 + return { 457 457 + run: completedRun, 458 458 + result: runResult, 459 459 + } satisfies RunSyncOutput; 460 460 + } 461 461 + 462 462 + const failureMessage = Cause.pretty(executionExit.cause); 463 463 + const failedResult: RunResult = { 464 464 + runId: runInput.runId, 465 465 + status: "failed", 466 466 + startedAt, 467 467 + completedAt, 468 468 + spawns: spawnResults, 469 469 + errorMessage: failureMessage, 470 470 + }; 471 471 + 472 472 + yield* appendTier1Event( 473 473 + lifecycleStateRef, 474 474 + sequenceRef, 475 475 + runStore, 476 476 + runInput.runId, 477 477 + (sequence, timestamp) => ({ 478 478 + ...makeEventEnvelope(runInput.runId, sequence, timestamp), 479 479 + type: "run:failed", 480 480 + payload: { 481 481 + message: failureMessage, 482 482 + }, 483 483 + }), 484 484 + ); 485 485 + 486 486 + yield* runStore.setResult(runInput.runId, failedResult, completedAt); 487 487 + 488 488 + return yield* Effect.fail( 489 489 + new ProgramExecutionError({ 490 490 + runId: runInput.runId, 491 491 + message: failureMessage, 492 492 + }), 493 493 + ); 494 494 + }), 495 495 + 496 496 + status: (runId) => runStore.getRun(runId), 497 497 + 498 498 + wait: (runId, timeout) => { 499 499 + const timeoutMillis = toTimeoutMillis(timeout); 500 500 + 501 501 + return waitForRunTerminal(runStore, runId).pipe( 502 502 + Effect.timeoutFail({ 503 503 + duration: timeoutMillis, 504 504 + onTimeout: () => 505 505 + new WaitTimeoutError({ 506 506 + runId, 507 507 + timeoutMillis, 508 508 + message: `Timed out waiting for terminal event for run ${runId} after ${timeoutMillis}ms.`, 509 509 + }), 510 510 + }), 511 511 + ); 512 512 + }, 513 513 + }; 514 514 + };

+244

packages/core/src/internal/lifecycle-guard.effect.test.ts

reviewed

··· 1 1 + import { describe, expect, it } from "bun:test"; 2 2 + import { Effect, Runtime, type Effect as EffectType } from "effect"; 3 3 + import { decodeRunIdSync, decodeSpawnIdSync } from "../domain/run.schema"; 4 4 + import { 5 5 + applyLifecycleTransition, 6 6 + initialLifecycleGuardState, 7 7 + type LifecycleGuardState, 8 8 + } from "./lifecycle-guard.effect"; 9 9 + 10 10 + const runtime = Runtime.defaultRuntime; 11 11 + const runEffect = <A, E>(effect: EffectType<A, E>): Promise<A> => 12 12 + Runtime.runPromise(runtime)(effect); 13 13 + 14 14 + const runId = decodeRunIdSync("run_lifecycle_guard"); 15 15 + const spawnId = decodeSpawnIdSync("spawn_lifecycle_guard"); 16 16 + 17 17 + const runStartEvent = { 18 18 + schemaVersion: 1 as const, 19 19 + runId, 20 20 + sequence: 1, 21 21 + timestamp: "2026-02-23T20:00:00.000Z", 22 22 + type: "run:start" as const, 23 23 + payload: { 24 24 + programPath: "/tmp/program.ts", 25 25 + }, 26 26 + }; 27 27 + 28 28 + describe("lifecycle guard transitions", () => { 29 29 + it("accepts a valid non-terminal to terminal progression", async () => { 30 30 + let state: LifecycleGuardState = initialLifecycleGuardState; 31 31 + 32 32 + state = await runEffect(applyLifecycleTransition(state, runStartEvent)); 33 33 + state = await runEffect( 34 34 + applyLifecycleTransition(state, { 35 35 + schemaVersion: 1, 36 36 + runId, 37 37 + sequence: 2, 38 38 + timestamp: "2026-02-23T20:00:01.000Z", 39 39 + type: "run:status", 40 40 + payload: { 41 41 + status: "running", 42 42 + }, 43 43 + }), 44 44 + ); 45 45 + state = await runEffect( 46 46 + applyLifecycleTransition(state, { 47 47 + schemaVersion: 1, 48 48 + runId, 49 49 + sequence: 3, 50 50 + timestamp: "2026-02-23T20:00:02.000Z", 51 51 + type: "spawn:start", 52 52 + payload: { 53 53 + spawnId, 54 54 + input: { 55 55 + agent: "scout", 56 56 + systemPrompt: "You are concise.", 57 57 + prompt: "summarize", 58 58 + }, 59 59 + }, 60 60 + }), 61 61 + ); 62 62 + state = await runEffect( 63 63 + applyLifecycleTransition(state, { 64 64 + schemaVersion: 1, 65 65 + runId, 66 66 + sequence: 4, 67 67 + timestamp: "2026-02-23T20:00:03.000Z", 68 68 + type: "spawn:complete", 69 69 + payload: { 70 70 + spawnId, 71 71 + result: { 72 72 + text: "done", 73 73 + sessionRef: "session/scout", 74 74 + agent: "scout", 75 75 + model: "openai/gpt-5.3-codex", 76 76 + driver: "pi", 77 77 + exitCode: 0, 78 78 + }, 79 79 + }, 80 80 + }), 81 81 + ); 82 82 + 83 83 + const terminalState = await runEffect( 84 84 + applyLifecycleTransition(state, { 85 85 + schemaVersion: 1, 86 86 + runId, 87 87 + sequence: 5, 88 88 + timestamp: "2026-02-23T20:00:04.000Z", 89 89 + type: "run:complete", 90 90 + payload: { 91 91 + result: { 92 92 + runId, 93 93 + status: "complete", 94 94 + startedAt: "2026-02-23T20:00:00.000Z", 95 95 + completedAt: "2026-02-23T20:00:04.000Z", 96 96 + spawns: [ 97 97 + { 98 98 + text: "done", 99 99 + sessionRef: "session/scout", 100 100 + agent: "scout", 101 101 + model: "openai/gpt-5.3-codex", 102 102 + driver: "pi", 103 103 + exitCode: 0, 104 104 + }, 105 105 + ], 106 106 + }, 107 107 + }, 108 108 + }), 109 109 + ); 110 110 + 111 111 + expect(terminalState.runTerminal).toBe("run:complete"); 112 112 + }); 113 113 + 114 114 + it("rejects terminal -> non-terminal transitions", async () => { 115 115 + const terminalState = await runEffect( 116 116 + applyLifecycleTransition(initialLifecycleGuardState, { 117 117 + ...runStartEvent, 118 118 + type: "run:complete", 119 119 + sequence: 2, 120 120 + payload: { 121 121 + result: { 122 122 + runId, 123 123 + status: "complete", 124 124 + startedAt: "2026-02-23T20:00:00.000Z", 125 125 + completedAt: "2026-02-23T20:00:01.000Z", 126 126 + spawns: [], 127 127 + }, 128 128 + }, 129 129 + }), 130 130 + ); 131 131 + 132 132 + const failure = await runEffect( 133 133 + Effect.flip( 134 134 + applyLifecycleTransition(terminalState, { 135 135 + schemaVersion: 1, 136 136 + runId, 137 137 + sequence: 3, 138 138 + timestamp: "2026-02-23T20:00:02.000Z", 139 139 + type: "run:status", 140 140 + payload: { 141 141 + status: "running", 142 142 + }, 143 143 + }), 144 144 + ), 145 145 + ); 146 146 + 147 147 + expect(failure).toMatchObject({ 148 148 + _tag: "LifecycleInvariantError", 149 149 + runId, 150 150 + }); 151 151 + }); 152 152 + 153 153 + it("rejects duplicate run terminal emissions deterministically", async () => { 154 154 + const terminalState = await runEffect( 155 155 + applyLifecycleTransition(initialLifecycleGuardState, { 156 156 + ...runStartEvent, 157 157 + type: "run:failed", 158 158 + sequence: 2, 159 159 + payload: { 160 160 + message: "boom", 161 161 + }, 162 162 + }), 163 163 + ); 164 164 + 165 165 + const failure = await runEffect( 166 166 + Effect.flip( 167 167 + applyLifecycleTransition(terminalState, { 168 168 + schemaVersion: 1, 169 169 + runId, 170 170 + sequence: 3, 171 171 + timestamp: "2026-02-23T20:00:02.000Z", 172 172 + type: "run:cancelled", 173 173 + payload: {}, 174 174 + }), 175 175 + ), 176 176 + ); 177 177 + 178 178 + expect(failure).toMatchObject({ 179 179 + _tag: "LifecycleInvariantError", 180 180 + runId, 181 181 + }); 182 182 + }); 183 183 + 184 184 + it("rejects duplicate spawn terminal emissions for a single spawn", async () => { 185 185 + let state = await runEffect( 186 186 + applyLifecycleTransition(initialLifecycleGuardState, { 187 187 + ...runStartEvent, 188 188 + sequence: 1, 189 189 + }), 190 190 + ); 191 191 + 192 192 + state = await runEffect( 193 193 + applyLifecycleTransition(state, { 194 194 + schemaVersion: 1, 195 195 + runId, 196 196 + sequence: 2, 197 197 + timestamp: "2026-02-23T20:00:01.000Z", 198 198 + type: "spawn:start", 199 199 + payload: { 200 200 + spawnId, 201 201 + input: { 202 202 + agent: "scout", 203 203 + systemPrompt: "You are concise.", 204 204 + prompt: "summarize", 205 205 + }, 206 206 + }, 207 207 + }), 208 208 + ); 209 209 + 210 210 + state = await runEffect( 211 211 + applyLifecycleTransition(state, { 212 212 + schemaVersion: 1, 213 213 + runId, 214 214 + sequence: 3, 215 215 + timestamp: "2026-02-23T20:00:02.000Z", 216 216 + type: "spawn:error", 217 217 + payload: { 218 218 + spawnId, 219 219 + message: "driver failed", 220 220 + }, 221 221 + }), 222 222 + ); 223 223 + 224 224 + const failure = await runEffect( 225 225 + Effect.flip( 226 226 + applyLifecycleTransition(state, { 227 227 + schemaVersion: 1, 228 228 + runId, 229 229 + sequence: 4, 230 230 + timestamp: "2026-02-23T20:00:03.000Z", 231 231 + type: "spawn:cancelled", 232 232 + payload: { 233 233 + spawnId, 234 234 + }, 235 235 + }), 236 236 + ), 237 237 + ); 238 238 + 239 239 + expect(failure).toMatchObject({ 240 240 + _tag: "LifecycleInvariantError", 241 241 + runId, 242 242 + }); 243 243 + }); 244 244 + });

+140

packages/core/src/internal/lifecycle-guard.effect.ts

reviewed

··· 1 1 + import { Data, Effect } from "effect"; 2 2 + import type { MillEvent } from "../domain/event.schema"; 3 3 + import type { RunStatus } from "../domain/run.schema"; 4 4 + 5 5 + type RunTerminalEventType = Extract<MillEvent["type"], "run:complete" | "run:failed" | "run:cancelled">; 6 6 + type SpawnTerminalEventType = Extract< 7 7 + MillEvent["type"], 8 8 + "spawn:complete" | "spawn:error" | "spawn:cancelled" 9 9 + >; 10 10 + 11 11 + export class LifecycleInvariantError extends Data.TaggedError("LifecycleInvariantError")<{ 12 12 + runId: string; 13 13 + message: string; 14 14 + }> {} 15 15 + 16 16 + export type LifecycleGuardState = { 17 17 + readonly runTerminal?: RunTerminalEventType; 18 18 + readonly spawnTerminals: Readonly<Record<string, SpawnTerminalEventType>>; 19 19 + }; 20 20 + 21 21 + export const initialLifecycleGuardState: LifecycleGuardState = { 22 22 + spawnTerminals: {}, 23 23 + }; 24 24 + 25 25 + const isRunTerminalType = (eventType: MillEvent["type"]): eventType is RunTerminalEventType => 26 26 + eventType === "run:complete" || eventType === "run:failed" || eventType === "run:cancelled"; 27 27 + 28 28 + const isSpawnTerminalType = (eventType: MillEvent["type"]): eventType is SpawnTerminalEventType => 29 29 + eventType === "spawn:complete" || eventType === "spawn:error" || eventType === "spawn:cancelled"; 30 30 + 31 31 + const spawnIdForEvent = (event: MillEvent): string | undefined => { 32 32 + if (event.type === "spawn:start") { 33 33 + return event.payload.spawnId; 34 34 + } 35 35 + 36 36 + if (event.type === "spawn:milestone") { 37 37 + return event.payload.spawnId; 38 38 + } 39 39 + 40 40 + if (event.type === "spawn:tool_call") { 41 41 + return event.payload.spawnId; 42 42 + } 43 43 + 44 44 + if (event.type === "spawn:complete") { 45 45 + return event.payload.spawnId; 46 46 + } 47 47 + 48 48 + if (event.type === "spawn:error") { 49 49 + return event.payload.spawnId; 50 50 + } 51 51 + 52 52 + if (event.type === "spawn:cancelled") { 53 53 + return event.payload.spawnId; 54 54 + } 55 55 + 56 56 + return undefined; 57 57 + }; 58 58 + 59 59 + export const applyLifecycleTransition = ( 60 60 + state: LifecycleGuardState, 61 61 + event: MillEvent, 62 62 + ): Effect.Effect<LifecycleGuardState, LifecycleInvariantError> => 63 63 + Effect.gen(function* () { 64 64 + if (state.runTerminal !== undefined) { 65 65 + return yield* Effect.fail( 66 66 + new LifecycleInvariantError({ 67 67 + runId: event.runId, 68 68 + message: `Event ${event.type} violates terminal single-shot policy: run already terminal with ${state.runTerminal}.`, 69 69 + }), 70 70 + ); 71 71 + } 72 72 + 73 73 + const spawnId = spawnIdForEvent(event); 74 74 + 75 75 + if (spawnId !== undefined && state.spawnTerminals[spawnId] !== undefined) { 76 76 + return yield* Effect.fail( 77 77 + new LifecycleInvariantError({ 78 78 + runId: event.runId, 79 79 + message: `Event ${event.type} violates terminal single-shot policy for spawn ${spawnId}: terminal already set to ${state.spawnTerminals[spawnId]}.`, 80 80 + }), 81 81 + ); 82 82 + } 83 83 + 84 84 + const nextRunTerminal = isRunTerminalType(event.type) ? event.type : state.runTerminal; 85 85 + 86 86 + if (spawnId === undefined || !isSpawnTerminalType(event.type)) { 87 87 + return { 88 88 + ...state, 89 89 + runTerminal: nextRunTerminal, 90 90 + }; 91 91 + } 92 92 + 93 93 + return { 94 94 + ...state, 95 95 + runTerminal: nextRunTerminal, 96 96 + spawnTerminals: { 97 97 + ...state.spawnTerminals, 98 98 + [spawnId]: event.type, 99 99 + }, 100 100 + }; 101 101 + }); 102 102 + 103 103 + const isTerminalStatus = (status: RunStatus): boolean => 104 104 + status === "complete" || status === "failed" || status === "cancelled"; 105 105 + 106 106 + export const ensureRunStatusTransition = ( 107 107 + runId: string, 108 108 + current: RunStatus, 109 109 + next: RunStatus, 110 110 + ): Effect.Effect<void, LifecycleInvariantError> => { 111 111 + if (isTerminalStatus(current)) { 112 112 + return Effect.fail( 113 113 + new LifecycleInvariantError({ 114 114 + runId, 115 115 + message: `Run status transition ${current} -> ${next} is invalid: terminal statuses are immutable.`, 116 116 + }), 117 117 + ); 118 118 + } 119 119 + 120 120 + if (current === "pending" && next === "running") { 121 121 + return Effect.void; 122 122 + } 123 123 + 124 124 + if (current === "pending" && next === "pending") { 125 125 + return Effect.void; 126 126 + } 127 127 + 128 128 + if (current === "running" && (next === "running" || isTerminalStatus(next))) { 129 129 + return Effect.void; 130 130 + } 131 131 + 132 132 + return Effect.fail( 133 133 + new LifecycleInvariantError({ 134 134 + runId, 135 135 + message: `Run status transition ${current} -> ${next} violates lifecycle transition guards.`, 136 136 + }), 137 137 + ); 138 138 + }; 139 139 + 140 140 + export const isRunTerminalEvent = (event: MillEvent): boolean => isRunTerminalType(event.type);

+118

packages/core/src/internal/run-store.effect.test.ts

reviewed

··· 1 1 + import { describe, expect, it } from "bun:test"; 2 2 + import { mkdtemp, readFile, rm } from "node:fs/promises"; 3 3 + import { tmpdir } from "node:os"; 4 4 + import { join } from "node:path"; 5 5 + import { Effect } from "effect"; 6 6 + import { decodeMillEventJsonSync } from "../domain/event.schema"; 7 7 + import { decodeRunIdSync } from "../domain/run.schema"; 8 8 + import { runWithBunContext } from "../public/test-runtime.api"; 9 9 + import { makeRunStore } from "./run-store.effect"; 10 10 + 11 11 + describe("RunStore", () => { 12 12 + it("creates run artifacts and appends tier-1 events as NDJSON", async () => { 13 13 + const runsDirectory = await mkdtemp(join(tmpdir(), "mill-run-store-")); 14 14 + const runId = decodeRunIdSync(`run_${crypto.randomUUID()}`); 15 15 + 16 16 + const store = makeRunStore({ runsDirectory }); 17 17 + 18 18 + try { 19 19 + const runRecord = await runWithBunContext( 20 20 + store.create({ 21 21 + runId, 22 22 + programPath: "/tmp/program.ts", 23 23 + driver: "default", 24 24 + timestamp: "2026-02-23T20:00:00.000Z", 25 25 + }), 26 26 + ); 27 27 + 28 28 + expect(runRecord.paths.runFile.endsWith("run.json")).toBe(true); 29 29 + expect(runRecord.paths.eventsFile.endsWith("events.ndjson")).toBe(true); 30 30 + expect(runRecord.paths.resultFile.endsWith("result.json")).toBe(true); 31 31 + 32 32 + await runWithBunContext( 33 33 + store.appendEvent(runId, { 34 34 + schemaVersion: 1, 35 35 + runId, 36 36 + sequence: 1, 37 37 + timestamp: "2026-02-23T20:00:01.000Z", 38 38 + type: "run:start", 39 39 + payload: { 40 40 + programPath: "/tmp/program.ts", 41 41 + }, 42 42 + }), 43 43 + ); 44 44 + 45 45 + await runWithBunContext( 46 46 + store.appendEvent(runId, { 47 47 + schemaVersion: 1, 48 48 + runId, 49 49 + sequence: 2, 50 50 + timestamp: "2026-02-23T20:00:02.000Z", 51 51 + type: "run:status", 52 52 + payload: { 53 53 + status: "running", 54 54 + }, 55 55 + }), 56 56 + ); 57 57 + 58 58 + const eventsFile = await readFile(runRecord.paths.eventsFile, "utf-8"); 59 59 + const lines = eventsFile 60 60 + .split("\n") 61 61 + .map((line) => line.trim()) 62 62 + .filter((line) => line.length > 0); 63 63 + 64 64 + expect(lines).toHaveLength(2); 65 65 + 66 66 + const firstEvent = decodeMillEventJsonSync(lines[0]); 67 67 + const secondEvent = decodeMillEventJsonSync(lines[1]); 68 68 + 69 69 + expect(firstEvent.type).toBe("run:start"); 70 70 + expect(secondEvent.type).toBe("run:status"); 71 71 + } finally { 72 72 + await rm(runsDirectory, { recursive: true, force: true }); 73 73 + } 74 74 + }); 75 75 + 76 76 + it("rejects terminal to non-terminal status transitions", async () => { 77 77 + const runsDirectory = await mkdtemp(join(tmpdir(), "mill-run-store-transition-")); 78 78 + const runId = decodeRunIdSync(`run_${crypto.randomUUID()}`); 79 79 + 80 80 + const store = makeRunStore({ runsDirectory }); 81 81 + 82 82 + try { 83 83 + await runWithBunContext( 84 84 + store.create({ 85 85 + runId, 86 86 + programPath: "/tmp/program.ts", 87 87 + driver: "default", 88 88 + timestamp: "2026-02-23T20:00:00.000Z", 89 89 + }), 90 90 + ); 91 91 + 92 92 + await runWithBunContext( 93 93 + store.setResult( 94 94 + runId, 95 95 + { 96 96 + runId, 97 97 + status: "complete", 98 98 + startedAt: "2026-02-23T20:00:00.000Z", 99 99 + completedAt: "2026-02-23T20:00:02.000Z", 100 100 + spawns: [], 101 101 + }, 102 102 + "2026-02-23T20:00:02.000Z", 103 103 + ), 104 104 + ); 105 105 + 106 106 + const transitionError = await runWithBunContext( 107 107 + Effect.flip(store.setStatus(runId, "running", "2026-02-23T20:00:03.000Z")), 108 108 + ); 109 109 + 110 110 + expect(transitionError).toMatchObject({ 111 111 + _tag: "LifecycleInvariantError", 112 112 + runId, 113 113 + }); 114 114 + } finally { 115 115 + await rm(runsDirectory, { recursive: true, force: true }); 116 116 + } 117 117 + }); 118 118 + });

+219

packages/core/src/internal/run-store.effect.ts

reviewed

··· 1 1 + import * as FileSystem from "@effect/platform/FileSystem"; 2 2 + import { Data, Effect } from "effect"; 3 3 + import { 4 4 + decodeMillEventJson, 5 5 + encodeMillEventJson, 6 6 + type MillEvent, 7 7 + } from "../domain/event.schema"; 8 8 + import { 9 9 + decodeRunRecordJson, 10 10 + type RunId, 11 11 + type RunRecord, 12 12 + type RunResult, 13 13 + } from "../domain/run.schema"; 14 14 + import { LifecycleInvariantError, ensureRunStatusTransition } from "./lifecycle-guard.effect"; 15 15 + 16 16 + export class RunNotFoundError extends Data.TaggedError("RunNotFoundError")<{ runId: string }> {} 17 17 + 18 18 + export class PersistenceError extends Data.TaggedError("PersistenceError")<{ 19 19 + path: string; 20 20 + message: string; 21 21 + }> {} 22 22 + 23 23 + export interface CreateRunInput { 24 24 + readonly runId: RunId; 25 25 + readonly programPath: string; 26 26 + readonly driver: string; 27 27 + readonly timestamp: string; 28 28 + } 29 29 + 30 30 + export interface RunStore { 31 31 + readonly create: (input: CreateRunInput) => Effect.Effect<RunRecord, PersistenceError>; 32 32 + readonly appendEvent: (runId: RunId, event: MillEvent) => Effect.Effect<void, PersistenceError>; 33 33 + readonly readEvents: ( 34 34 + runId: RunId, 35 35 + ) => Effect.Effect<ReadonlyArray<MillEvent>, RunNotFoundError | PersistenceError>; 36 36 + readonly setStatus: ( 37 37 + runId: RunId, 38 38 + status: RunRecord["status"], 39 39 + timestamp: string, 40 40 + ) => Effect.Effect< 41 41 + RunRecord, 42 42 + RunNotFoundError | PersistenceError | LifecycleInvariantError 43 43 + >; 44 44 + readonly setResult: ( 45 45 + runId: RunId, 46 46 + result: RunResult, 47 47 + timestamp: string, 48 48 + ) => Effect.Effect< 49 49 + void, 50 50 + RunNotFoundError | PersistenceError | LifecycleInvariantError 51 51 + >; 52 52 + readonly getRun: (runId: RunId) => Effect.Effect<RunRecord, RunNotFoundError | PersistenceError>; 53 53 + } 54 54 + 55 55 + export interface MakeRunStoreInput { 56 56 + readonly runsDirectory: string; 57 57 + } 58 58 + 59 59 + const joinPath = (base: string, child: string): string => 60 60 + base.endsWith("/") ? `${base}${child}` : `${base}/${child}`; 61 61 + 62 62 + const toMessage = (error: unknown): string => { 63 63 + if (error instanceof Error) { 64 64 + return error.message; 65 65 + } 66 66 + 67 67 + return String(error); 68 68 + }; 69 69 + 70 70 + const mapPersistenceError = (path: string) => 71 71 + Effect.mapError((error: unknown) => new PersistenceError({ path, message: toMessage(error) })); 72 72 + 73 73 + const buildPaths = (runsDirectory: string, runId: RunId): RunRecord["paths"] => { 74 74 + const runDir = joinPath(runsDirectory, runId); 75 75 + 76 76 + return { 77 77 + runDir, 78 78 + runFile: joinPath(runDir, "run.json"), 79 79 + eventsFile: joinPath(runDir, "events.ndjson"), 80 80 + resultFile: joinPath(runDir, "result.json"), 81 81 + }; 82 82 + }; 83 83 + 84 84 + const storeSetStatus = ( 85 85 + runsDirectory: string, 86 86 + runId: RunId, 87 87 + status: RunRecord["status"], 88 88 + timestamp: string, 89 89 + ): Effect.Effect<RunRecord, RunNotFoundError | PersistenceError | LifecycleInvariantError> => 90 90 + Effect.gen(function* () { 91 91 + const fileSystem = yield* FileSystem.FileSystem; 92 92 + const currentRun = yield* storeGetRun(runsDirectory, runId); 93 93 + 94 94 + yield* ensureRunStatusTransition(runId, currentRun.status, status); 95 95 + 96 96 + const nextRun: RunRecord = { 97 97 + ...currentRun, 98 98 + status, 99 99 + updatedAt: timestamp, 100 100 + }; 101 101 + 102 102 + yield* mapPersistenceError(currentRun.paths.runFile)( 103 103 + fileSystem.writeFileString(currentRun.paths.runFile, `${JSON.stringify(nextRun, null, 2)}\n`), 104 104 + ); 105 105 + 106 106 + return nextRun; 107 107 + }); 108 108 + 109 109 + export const makeRunStore = (input: MakeRunStoreInput): RunStore => ({ 110 110 + create: (createInput) => 111 111 + Effect.gen(function* () { 112 112 + const fileSystem = yield* FileSystem.FileSystem; 113 113 + const paths = buildPaths(input.runsDirectory, createInput.runId); 114 114 + const runRecord: RunRecord = { 115 115 + id: createInput.runId, 116 116 + status: "running", 117 117 + programPath: createInput.programPath, 118 118 + driver: createInput.driver, 119 119 + createdAt: createInput.timestamp, 120 120 + updatedAt: createInput.timestamp, 121 121 + paths, 122 122 + }; 123 123 + 124 124 + yield* mapPersistenceError(paths.runDir)( 125 125 + fileSystem.makeDirectory(paths.runDir, { recursive: true }), 126 126 + ); 127 127 + yield* mapPersistenceError(paths.runFile)( 128 128 + fileSystem.writeFileString(paths.runFile, `${JSON.stringify(runRecord, null, 2)}\n`), 129 129 + ); 130 130 + yield* mapPersistenceError(paths.eventsFile)( 131 131 + fileSystem.writeFileString(paths.eventsFile, ""), 132 132 + ); 133 133 + 134 134 + return runRecord; 135 135 + }), 136 136 + 137 137 + appendEvent: (runId, event) => 138 138 + Effect.gen(function* () { 139 139 + const fileSystem = yield* FileSystem.FileSystem; 140 140 + const paths = buildPaths(input.runsDirectory, runId); 141 141 + 142 142 + yield* mapPersistenceError(paths.eventsFile)( 143 143 + fileSystem.writeFileString(paths.eventsFile, `${encodeMillEventJson(event)}\n`, { 144 144 + flag: "a", 145 145 + }), 146 146 + ); 147 147 + }), 148 148 + 149 149 + readEvents: (runId) => 150 150 + Effect.gen(function* () { 151 151 + const fileSystem = yield* FileSystem.FileSystem; 152 152 + const runRecord = yield* storeGetRun(input.runsDirectory, runId); 153 153 + const eventsContent = yield* mapPersistenceError(runRecord.paths.eventsFile)( 154 154 + fileSystem.readFileString(runRecord.paths.eventsFile, "utf-8"), 155 155 + ); 156 156 + 157 157 + const lines = eventsContent 158 158 + .split("\n") 159 159 + .map((line) => line.trim()) 160 160 + .filter((line) => line.length > 0); 161 161 + 162 162 + return yield* Effect.forEach(lines, (line) => 163 163 + Effect.mapError( 164 164 + decodeMillEventJson(line), 165 165 + (error) => 166 166 + new PersistenceError({ 167 167 + path: runRecord.paths.eventsFile, 168 168 + message: toMessage(error), 169 169 + }), 170 170 + ), 171 171 + ); 172 172 + }), 173 173 + 174 174 + setStatus: (runId, status, timestamp) => 175 175 + storeSetStatus(input.runsDirectory, runId, status, timestamp), 176 176 + 177 177 + setResult: (runId, result, timestamp) => 178 178 + Effect.gen(function* () { 179 179 + const fileSystem = yield* FileSystem.FileSystem; 180 180 + const runRecord = yield* storeGetRun(input.runsDirectory, runId); 181 181 + 182 182 + yield* mapPersistenceError(runRecord.paths.resultFile)( 183 183 + fileSystem.writeFileString(runRecord.paths.resultFile, `${JSON.stringify(result, null, 2)}\n`), 184 184 + ); 185 185 + 186 186 + yield* storeSetStatus(input.runsDirectory, runId, result.status, timestamp); 187 187 + }), 188 188 + 189 189 + getRun: (runId) => storeGetRun(input.runsDirectory, runId), 190 190 + }); 191 191 + 192 192 + const storeGetRun = ( 193 193 + runsDirectory: string, 194 194 + runId: RunId, 195 195 + ): Effect.Effect<RunRecord, RunNotFoundError | PersistenceError> => 196 196 + Effect.gen(function* () { 197 197 + const fileSystem = yield* FileSystem.FileSystem; 198 198 + const paths = buildPaths(runsDirectory, runId); 199 199 + const exists = yield* mapPersistenceError(paths.runFile)(fileSystem.exists(paths.runFile)); 200 200 + 201 201 + if (!exists) { 202 202 + return yield* Effect.fail(new RunNotFoundError({ runId })); 203 203 + } 204 204 + 205 205 + const runFileContent = yield* mapPersistenceError(paths.runFile)( 206 206 + fileSystem.readFileString(paths.runFile, "utf-8"), 207 207 + ); 208 208 + 209 209 + const runRecord = yield* Effect.mapError( 210 210 + decodeRunRecordJson(runFileContent), 211 211 + (error) => 212 212 + new PersistenceError({ 213 213 + path: paths.runFile, 214 214 + message: toMessage(error), 215 215 + }), 216 216 + ); 217 217 + 218 218 + return runRecord; 219 219 + });

+1

packages/core/src/public/index.api.ts

reviewed

··· 1 1 export * from "./mill.api"; 2 2 export * from "./config-loader.api"; 3 3 export * from "./discovery.api"; 4 4 + export * from "./run.api"; 4 5 export * from "./types";

+202

packages/core/src/public/run.api.ts

reviewed

··· 1 1 + import * as FileSystem from "@effect/platform/FileSystem"; 2 2 + import * as BunContext from "@effect/platform-bun/BunContext"; 3 3 + import { Effect, Runtime } from "effect"; 4 4 + import { makeMillEngine, type ProgramExecutionError } from "../engine.effect"; 5 5 + import { decodeRunIdSync, type RunRecord, type RunSyncOutput } from "../run.schema"; 6 6 + import { decodeSpawnOptions } from "../spawn.schema"; 7 7 + import { resolveConfig } from "./config-loader.api"; 8 8 + import type { 9 9 + ConfigOverrides, 10 10 + DriverRegistration, 11 11 + ResolveConfigOptions, 12 12 + SpawnInput, 13 13 + SpawnOutput, 14 14 + } from "./types"; 15 15 + 16 16 + const runtime = Runtime.defaultRuntime; 17 17 + 18 18 + type ProgramRunner = () => Promise<unknown>; 19 19 + 20 20 + type AsyncFunctionConstructor = new (...args: ReadonlyArray<string>) => ProgramRunner; 21 21 + 22 22 + const ProgramAsyncFunction = Object.getPrototypeOf(async () => undefined) 23 23 + .constructor as AsyncFunctionConstructor; 24 24 + 25 25 + interface GlobalMillContext { 26 26 + mill?: { 27 27 + spawn: (input: SpawnInput) => Promise<SpawnOutput>; 28 28 + }; 29 29 + } 30 30 + 31 31 + interface RunProgramSyncInput extends ResolveConfigOptions { 32 32 + readonly programPath: string; 33 33 + readonly driverName?: string; 34 34 + readonly runsDirectory?: string; 35 35 + } 36 36 + 37 37 + interface GetRunStatusInput extends Omit< 38 38 + ResolveConfigOptions, 39 39 + "pathExists" | "loadConfigOverrides" 40 40 + > { 41 41 + readonly runId: string; 42 42 + readonly driverName?: string; 43 43 + readonly runsDirectory?: string; 44 44 + readonly pathExists?: (path: string) => Promise<boolean>; 45 45 + readonly loadConfigOverrides?: (path: string) => Promise<ConfigOverrides>; 46 46 + } 47 47 + 48 48 + export interface WaitForRunInput extends GetRunStatusInput { 49 49 + readonly timeoutSeconds: number; 50 50 + } 51 51 + 52 52 + const normalizePath = (path: string): string => { 53 53 + if (path.length <= 1) { 54 54 + return path; 55 55 + } 56 56 + 57 57 + return path.endsWith("/") ? path.slice(0, -1) : path; 58 58 + }; 59 59 + 60 60 + const joinPath = (base: string, child: string): string => 61 61 + normalizePath(base) === "/" ? `/${child}` : `${normalizePath(base)}/${child}`; 62 62 + 63 63 + const resolveProgramPath = (cwd: string, programPath: string): string => 64 64 + programPath.startsWith("/") ? normalizePath(programPath) : joinPath(cwd, programPath); 65 65 + 66 66 + const resolveRunsDirectory = ( 67 67 + cwd: string, 68 68 + homeDirectory: string | undefined, 69 69 + runsDirectory: string | undefined, 70 70 + ): string => { 71 71 + if (runsDirectory !== undefined && runsDirectory.length > 0) { 72 72 + return runsDirectory; 73 73 + } 74 74 + 75 75 + if (homeDirectory !== undefined && homeDirectory.length > 0) { 76 76 + return joinPath(homeDirectory, ".mill/runs"); 77 77 + } 78 78 + 79 79 + return joinPath(cwd, ".mill/runs"); 80 80 + }; 81 81 + 82 82 + const runWithBunContext = <A, E>(effect: Effect.Effect<A, E, BunContext.BunContext>): Promise<A> => 83 83 + Runtime.runPromise(runtime)(Effect.provide(effect, BunContext.layer)); 84 84 + 85 85 + const readProgramSource = ( 86 86 + programPath: string, 87 87 + ): Effect.Effect<string, unknown, BunContext.BunContext> => 88 88 + Effect.flatMap(FileSystem.FileSystem, (fileSystem) => 89 89 + fileSystem.readFileString(programPath, "utf-8"), 90 90 + ); 91 91 + 92 92 + const resolveRuntimeDriver = ( 93 93 + registration: DriverRegistration | undefined, 94 94 + fallback: DriverRegistration | undefined, 95 95 + ) => { 96 96 + if (registration?.runtime !== undefined) { 97 97 + return registration.runtime; 98 98 + } 99 99 + 100 100 + return fallback?.runtime; 101 101 + }; 102 102 + 103 103 + const executeProgramWithInjectedMill = ( 104 104 + programSource: string, 105 105 + spawn: (input: SpawnInput) => Effect.Effect<SpawnOutput, unknown>, 106 106 + ): Effect.Effect<unknown, ProgramExecutionError> => 107 107 + Effect.tryPromise({ 108 108 + try: async () => { 109 109 + const globalContext = globalThis as GlobalMillContext; 110 110 + const previousMill = globalContext.mill; 111 111 + const programRunner = new ProgramAsyncFunction(programSource); 112 112 + 113 113 + globalContext.mill = { 114 114 + spawn: async (input) => { 115 115 + const decodedInput = await Runtime.runPromise(runtime)(decodeSpawnOptions(input)); 116 116 + return Runtime.runPromise(runtime)(Effect.provide(spawn(decodedInput), BunContext.layer)); 117 117 + }, 118 118 + }; 119 119 + 120 120 + try { 121 121 + return await programRunner(); 122 122 + } finally { 123 123 + if (previousMill === undefined) { 124 124 + delete globalContext.mill; 125 125 + } else { 126 126 + globalContext.mill = previousMill; 127 127 + } 128 128 + } 129 129 + }, 130 130 + catch: (error) => 131 131 + new ProgramExecutionError({ 132 132 + runId: "pending", 133 133 + message: String(error), 134 134 + }), 135 135 + }); 136 136 + 137 137 + const makeEngineForConfig = async ( 138 138 + input: GetRunStatusInput, 139 139 + ): Promise<ReturnType<typeof makeMillEngine>> => { 140 140 + const cwd = input.cwd ?? process.cwd(); 141 141 + const resolvedConfig = await resolveConfig(input); 142 142 + const selectedDriverName = input.driverName ?? resolvedConfig.config.defaultDriver; 143 143 + const selectedDriver = resolvedConfig.config.drivers[selectedDriverName]; 144 144 + const fallbackDriver = resolvedConfig.config.drivers[resolvedConfig.config.defaultDriver]; 145 145 + const runtimeDriver = resolveRuntimeDriver(selectedDriver, fallbackDriver); 146 146 + const runsDirectory = resolveRunsDirectory(cwd, input.homeDirectory, input.runsDirectory); 147 147 + 148 148 + return makeMillEngine({ 149 149 + runsDirectory, 150 150 + driverName: selectedDriverName, 151 151 + defaultModel: resolvedConfig.config.defaultModel, 152 152 + driver: runtimeDriver ?? fallbackDriver.runtime!, 153 153 + }); 154 154 + }; 155 155 + 156 156 + export const runProgramSync = async (input: RunProgramSyncInput): Promise<RunSyncOutput> => { 157 157 + const cwd = input.cwd ?? process.cwd(); 158 158 + const resolvedConfig = await resolveConfig(input); 159 159 + const selectedDriverName = input.driverName ?? resolvedConfig.config.defaultDriver; 160 160 + const selectedDriver = resolvedConfig.config.drivers[selectedDriverName]; 161 161 + const fallbackDriver = resolvedConfig.config.drivers[resolvedConfig.config.defaultDriver]; 162 162 + const runtimeDriver = resolveRuntimeDriver(selectedDriver, fallbackDriver); 163 163 + const programPath = resolveProgramPath(cwd, input.programPath); 164 164 + const runsDirectory = resolveRunsDirectory(cwd, input.homeDirectory, input.runsDirectory); 165 165 + 166 166 + const programSource = await runWithBunContext(readProgramSource(programPath)); 167 167 + 168 168 + const runId = decodeRunIdSync(`run_${crypto.randomUUID()}`); 169 169 + const engine = makeMillEngine({ 170 170 + runsDirectory, 171 171 + driverName: selectedDriverName, 172 172 + defaultModel: resolvedConfig.config.defaultModel, 173 173 + driver: runtimeDriver ?? fallbackDriver.runtime!, 174 174 + }); 175 175 + 176 176 + return runWithBunContext( 177 177 + engine.runSync({ 178 178 + runId, 179 179 + programPath, 180 180 + executeProgram: (spawn) => executeProgramWithInjectedMill(programSource, spawn), 181 181 + }), 182 182 + ); 183 183 + }; 184 184 + 185 185 + export const getRunStatus = async (input: GetRunStatusInput): Promise<RunRecord> => { 186 186 + const engine = await makeEngineForConfig(input); 187 187 + 188 188 + return runWithBunContext(engine.status(decodeRunIdSync(input.runId))); 189 189 + }; 190 190 + 191 191 + export const waitForRun = async (input: WaitForRunInput): Promise<RunRecord> => { 192 192 + const engine = await makeEngineForConfig(input); 193 193 + const waitOutcome = await runWithBunContext( 194 194 + Effect.either(engine.wait(decodeRunIdSync(input.runId), Math.round(input.timeoutSeconds * 1000))), 195 195 + ); 196 196 + 197 197 + if (waitOutcome._tag === "Right") { 198 198 + return waitOutcome.right; 199 199 + } 200 200 + 201 201 + throw waitOutcome.left; 202 202 + };

+8

packages/core/src/public/test-runtime.api.ts

reviewed

··· 1 1 + import * as BunContext from "@effect/platform-bun/BunContext"; 2 2 + import { Effect, Runtime } from "effect"; 3 3 + 4 4 + const runtime = Runtime.defaultRuntime; 5 5 + 6 6 + export const runWithBunContext = <A, E>( 7 7 + effect: Effect.Effect<A, E, BunContext.BunContext>, 8 8 + ): Promise<A> => Runtime.runPromise(runtime)(Effect.provide(effect, BunContext.layer));

+30

packages/core/src/public/types.ts

reviewed

··· 18 18 readonly errorMessage?: string; 19 19 } 20 20 21 21 + export interface DriverSpawnInput { 22 22 + readonly runId: string; 23 23 + readonly spawnId: string; 24 24 + readonly agent: string; 25 25 + readonly systemPrompt: string; 26 26 + readonly prompt: string; 27 27 + readonly model: string; 28 28 + } 29 29 + 30 30 + export type DriverSpawnEvent = 31 31 + | { 32 32 + readonly type: "milestone"; 33 33 + readonly message: string; 34 34 + } 35 35 + | { 36 36 + readonly type: "tool_call"; 37 37 + readonly toolName: string; 38 38 + }; 39 39 + 40 40 + export interface DriverSpawnOutput { 41 41 + readonly events: ReadonlyArray<DriverSpawnEvent>; 42 42 + readonly result: SpawnOutput; 43 43 + } 44 44 + 45 45 + export interface DriverRuntime { 46 46 + readonly name: string; 47 47 + readonly spawn: (input: DriverSpawnInput) => Effect.Effect<DriverSpawnOutput, unknown>; 48 48 + } 49 49 + 21 50 export interface Mill { 22 51 spawn(input: SpawnInput): Promise<SpawnOutput>; 23 52 } ··· 37 66 readonly modelFormat: string; 38 67 readonly process: DriverProcessConfig; 39 68 readonly codec: DriverCodec; 69 69 + readonly runtime?: DriverRuntime; 40 70 } 41 71 42 72 export interface MillConfig {

+1

packages/core/src/run.schema.ts

reviewed

··· 1 1 + export * from "./domain/run.schema";

+23 -2

packages/core/src/runtime/worker.effect.ts

reviewed

··· 1 1 import { Effect } from "effect"; 2 2 - import type { MillEngine } from "../internal/engine.effect"; 2 2 + import type { RunId } from "../domain/run.schema"; 3 3 import type { SpawnOptions, SpawnResult } from "../domain/spawn.schema"; 4 4 + import type { MillEngine } from "../internal/engine.effect"; 4 5 5 6 export interface WorkerInput { 6 7 readonly engine: MillEngine; 8 8 + readonly runId: RunId; 9 9 + readonly programPath: string; 7 10 readonly spawn: SpawnOptions; 8 11 } 9 12 10 13 export const runWorker = (input: WorkerInput): Effect.Effect<SpawnResult> => 11 11 - input.engine.submit(input.spawn); 14 14 + Effect.flatMap( 15 15 + input.engine.runSync({ 16 16 + runId: input.runId, 17 17 + programPath: input.programPath, 18 18 + executeProgram: (spawn) => Effect.flatMap(spawn(input.spawn), () => Effect.void), 19 19 + }), 20 20 + (output) => 21 21 + Effect.succeed( 22 22 + output.result.spawns[0] ?? { 23 23 + text: "", 24 24 + sessionRef: "session/worker-missing", 25 25 + agent: input.spawn.agent, 26 26 + model: input.spawn.model ?? "unknown", 27 27 + driver: "unknown", 28 28 + exitCode: 1, 29 29 + errorMessage: "no spawn result", 30 30 + }, 31 31 + ), 32 32 + );

+1

packages/core/src/spawn.schema.ts

reviewed

··· 1 1 + export * from "./domain/spawn.schema";

+55

packages/driver-pi/src/internal/pi.codec.test.ts

reviewed

··· 1 1 + import { describe, expect, it } from "bun:test"; 2 2 + import { Effect, Runtime } from "effect"; 3 3 + import { decodePiProcessOutput } from "./pi.codec"; 4 4 + 5 5 + const runtime = Runtime.defaultRuntime; 6 6 + 7 7 + describe("pi codec terminal sequencing", () => { 8 8 + it("rejects duplicate final lines deterministically", async () => { 9 9 + const output = [ 10 10 + JSON.stringify({ type: "milestone", message: "start" }), 11 11 + JSON.stringify({ 12 12 + type: "final", 13 13 + text: "done", 14 14 + sessionRef: "session/scout", 15 15 + agent: "scout", 16 16 + model: "openai/gpt-5.3-codex", 17 17 + exitCode: 0, 18 18 + }), 19 19 + JSON.stringify({ 20 20 + type: "final", 21 21 + text: "done-again", 22 22 + sessionRef: "session/scout", 23 23 + agent: "scout", 24 24 + model: "openai/gpt-5.3-codex", 25 25 + exitCode: 0, 26 26 + }), 27 27 + ].join("\n"); 28 28 + 29 29 + const decodeError = await Runtime.runPromise(runtime)(Effect.flip(decodePiProcessOutput(output))); 30 30 + 31 31 + expect(decodeError).toMatchObject({ 32 32 + _tag: "PiCodecError", 33 33 + }); 34 34 + }); 35 35 + 36 36 + it("rejects non-terminal lines emitted after final terminal", async () => { 37 37 + const output = [ 38 38 + JSON.stringify({ 39 39 + type: "final", 40 40 + text: "done", 41 41 + sessionRef: "session/scout", 42 42 + agent: "scout", 43 43 + model: "openai/gpt-5.3-codex", 44 44 + exitCode: 0, 45 45 + }), 46 46 + JSON.stringify({ type: "tool_call", toolName: "grep" }), 47 47 + ].join("\n"); 48 48 + 49 49 + const decodeError = await Runtime.runPromise(runtime)(Effect.flip(decodePiProcessOutput(output))); 50 50 + 51 51 + expect(decodeError).toMatchObject({ 52 52 + _tag: "PiCodecError", 53 53 + }); 54 54 + }); 55 55 + });

+130

packages/driver-pi/src/internal/pi.codec.ts

reviewed

··· 1 1 + import * as Schema from "@effect/schema/Schema"; 2 2 + import { Data, Effect } from "effect"; 3 3 + import type { DriverSpawnEvent, DriverSpawnOutput } from "@mill/core"; 4 4 + 5 5 + export class PiCodecError extends Data.TaggedError("PiCodecError")<{ 6 6 + message: string; 7 7 + }> {} 8 8 + 9 9 + const PiMilestoneLine = Schema.Struct({ 10 10 + type: Schema.Literal("milestone"), 11 11 + message: Schema.NonEmptyString, 12 12 + }); 13 13 + 14 14 + type PiMilestoneLine = Schema.Schema.Type<typeof PiMilestoneLine>; 15 15 + 16 16 + const PiToolCallLine = Schema.Struct({ 17 17 + type: Schema.Literal("tool_call"), 18 18 + toolName: Schema.NonEmptyString, 19 19 + }); 20 20 + 21 21 + type PiToolCallLine = Schema.Schema.Type<typeof PiToolCallLine>; 22 22 + 23 23 + const PiFinalLine = Schema.Struct({ 24 24 + type: Schema.Literal("final"), 25 25 + text: Schema.String, 26 26 + sessionRef: Schema.NonEmptyString, 27 27 + agent: Schema.NonEmptyString, 28 28 + model: Schema.NonEmptyString, 29 29 + exitCode: Schema.Number, 30 30 + stopReason: Schema.optional(Schema.String), 31 31 + errorMessage: Schema.optional(Schema.String), 32 32 + }); 33 33 + 34 34 + type PiFinalLine = Schema.Schema.Type<typeof PiFinalLine>; 35 35 + 36 36 + const PiOutputLine = Schema.Union(PiMilestoneLine, PiToolCallLine, PiFinalLine); 37 37 + 38 38 + const decodeLine = Schema.decodeUnknown(Schema.parseJson(PiOutputLine)); 39 39 + 40 40 + const toDriverEvent = (line: PiMilestoneLine | PiToolCallLine): DriverSpawnEvent => { 41 41 + if (line.type === "milestone") { 42 42 + return { 43 43 + type: "milestone", 44 44 + message: line.message, 45 45 + }; 46 46 + } 47 47 + 48 48 + return { 49 49 + type: "tool_call", 50 50 + toolName: line.toolName, 51 51 + }; 52 52 + }; 53 53 + 54 54 + const decodeFinalResult = ( 55 55 + finalLine: PiFinalLine | undefined, 56 56 + ): Effect.Effect<DriverSpawnOutput["result"], PiCodecError> => { 57 57 + if (finalLine === undefined) { 58 58 + return Effect.fail( 59 59 + new PiCodecError({ 60 60 + message: "Missing final output line from pi process", 61 61 + }), 62 62 + ); 63 63 + } 64 64 + 65 65 + return Effect.succeed({ 66 66 + text: finalLine.text, 67 67 + sessionRef: finalLine.sessionRef, 68 68 + agent: finalLine.agent, 69 69 + model: finalLine.model, 70 70 + driver: "pi", 71 71 + exitCode: finalLine.exitCode, 72 72 + stopReason: finalLine.stopReason, 73 73 + errorMessage: finalLine.errorMessage, 74 74 + }); 75 75 + }; 76 76 + 77 77 + export const decodePiProcessOutput = ( 78 78 + output: string, 79 79 + ): Effect.Effect<DriverSpawnOutput, PiCodecError> => 80 80 + Effect.gen(function* () { 81 81 + const lines = output 82 82 + .split("\n") 83 83 + .map((line) => line.trim()) 84 84 + .filter((line) => line.length > 0); 85 85 + 86 86 + const decodedLines = yield* Effect.forEach(lines, (line) => 87 87 + Effect.mapError( 88 88 + decodeLine(line), 89 89 + (error) => 90 90 + new PiCodecError({ 91 91 + message: String(error), 92 92 + }), 93 93 + ), 94 94 + ); 95 95 + 96 96 + let finalLine: PiFinalLine | undefined = undefined; 97 97 + const events: Array<DriverSpawnEvent> = []; 98 98 + 99 99 + for (const decoded of decodedLines) { 100 100 + if (decoded.type === "final") { 101 101 + if (finalLine !== undefined) { 102 102 + return yield* Effect.fail( 103 103 + new PiCodecError({ 104 104 + message: "Duplicate terminal final lines are not allowed.", 105 105 + }), 106 106 + ); 107 107 + } 108 108 + 109 109 + finalLine = decoded; 110 110 + continue; 111 111 + } 112 112 + 113 113 + if (finalLine !== undefined) { 114 114 + return yield* Effect.fail( 115 115 + new PiCodecError({ 116 116 + message: `Non-terminal line ${decoded.type} emitted after final terminal line.`, 117 117 + }), 118 118 + ); 119 119 + } 120 120 + 121 121 + events.push(toDriverEvent(decoded)); 122 122 + } 123 123 + 124 124 + const result = yield* decodeFinalResult(finalLine); 125 125 + 126 126 + return { 127 127 + events, 128 128 + result, 129 129 + } satisfies DriverSpawnOutput; 130 130 + });

+52

packages/driver-pi/src/internal/process-driver.effect.ts

reviewed

··· 1 1 + import * as Command from "@effect/platform/Command"; 2 2 + import { Data, Effect } from "effect"; 3 3 + import type { DriverProcessConfig, DriverRuntime, DriverSpawnInput } from "@mill/core"; 4 4 + import { decodePiProcessOutput } from "./pi.codec"; 5 5 + 6 6 + export class PiProcessDriverError extends Data.TaggedError("PiProcessDriverError")<{ 7 7 + message: string; 8 8 + }> {} 9 9 + 10 10 + const toMessage = (error: unknown): string => { 11 11 + if (error instanceof Error) { 12 12 + return error.message; 13 13 + } 14 14 + 15 15 + return String(error); 16 16 + }; 17 17 + 18 18 + const commandForSpawn = (config: DriverProcessConfig, input: DriverSpawnInput): Command.Command => { 19 19 + const payload = JSON.stringify({ 20 20 + runId: input.runId, 21 21 + spawnId: input.spawnId, 22 22 + agent: input.agent, 23 23 + systemPrompt: input.systemPrompt, 24 24 + prompt: input.prompt, 25 25 + model: input.model, 26 26 + }); 27 27 + 28 28 + return Command.env(Command.make(config.command, ...config.args, payload), config.env ?? {}); 29 29 + }; 30 30 + 31 31 + export const makePiProcessDriver = (config: DriverProcessConfig): DriverRuntime => ({ 32 32 + name: "pi", 33 33 + spawn: (input) => 34 34 + Effect.gen(function* () { 35 35 + const command = commandForSpawn(config, input); 36 36 + const stdout = yield* Effect.mapError( 37 37 + Command.string(command), 38 38 + (error) => 39 39 + new PiProcessDriverError({ 40 40 + message: toMessage(error), 41 41 + }), 42 42 + ); 43 43 + 44 44 + return yield* Effect.mapError( 45 45 + decodePiProcessOutput(stdout), 46 46 + (error) => 47 47 + new PiProcessDriverError({ 48 48 + message: toMessage(error), 49 49 + }), 50 50 + ); 51 51 + }), 52 52 + });

+1

packages/driver-pi/src/pi.codec.ts

reviewed

··· 1 1 + export * from "./internal/pi.codec";

+1

packages/driver-pi/src/process-driver.effect.ts

reviewed

··· 1 1 + export * from "./internal/process-driver.effect";

+75 -3

packages/driver-pi/src/public/index.api.test.ts

reviewed

··· 1 1 import { describe, expect, it } from "bun:test"; 2 2 - import { Runtime } from "effect"; 2 2 + import * as BunContext from "@effect/platform-bun/BunContext"; 3 3 + import { Effect, Runtime } from "effect"; 3 4 import { createPiDriverRegistration } from "./index.api"; 4 5 5 6 const runtime = Runtime.defaultRuntime; 6 7 8 8 + const DUPLICATE_TERMINAL_SCRIPT = 9 9 + "const input=JSON.parse(process.argv[1]);" + 10 10 + "console.log(JSON.stringify({type:'milestone',message:'spawn:'+input.agent}));" + 11 11 + "console.log(JSON.stringify({type:'final',text:'ok',sessionRef:'session/'+input.agent,agent:input.agent,model:input.model,exitCode:0}));" + 12 12 + "console.log(JSON.stringify({type:'final',text:'duplicate',sessionRef:'session/'+input.agent,agent:input.agent,model:input.model,exitCode:0}));"; 13 13 + 7 14 describe("createPiDriverRegistration", () => { 8 15 it("exposes catalog-backed model discovery via codec", async () => { 9 16 const driver = createPiDriverRegistration(); ··· 11 18 const models = await Runtime.runPromise(runtime)(driver.codec.modelCatalog); 12 19 13 20 expect(models).toEqual(["openai/gpt-5.3-codex", "anthropic/claude-sonnet-4-6"]); 14 14 - expect(driver.process.command).toBe("pi"); 15 15 - expect(driver.process.args).toEqual(["-p"]); 21 21 + expect(driver.process.command.length).toBeGreaterThan(0); 22 22 + expect(driver.process.args.length).toBeGreaterThan(0); 23 23 + }); 24 24 + 25 25 + it("spawns process-backed runs and decodes structured result payload", async () => { 26 26 + const driver = createPiDriverRegistration(); 27 27 + 28 28 + expect(driver.runtime).toBeDefined(); 29 29 + 30 30 + if (driver.runtime === undefined) { 31 31 + return; 32 32 + } 33 33 + 34 34 + const output = await Runtime.runPromise(runtime)( 35 35 + Effect.provide( 36 36 + driver.runtime.spawn({ 37 37 + runId: "run_driver_test", 38 38 + spawnId: "spawn_driver_test", 39 39 + agent: "scout", 40 40 + systemPrompt: "You are concise.", 41 41 + prompt: "Say hello", 42 42 + model: "openai/gpt-5.3-codex", 43 43 + }), 44 44 + BunContext.layer, 45 45 + ), 46 46 + ); 47 47 + 48 48 + expect(output.events.length).toBeGreaterThan(0); 49 49 + expect(output.result.sessionRef.length).toBeGreaterThan(0); 50 50 + expect(output.result.agent).toBe("scout"); 51 51 + expect(output.result.model).toBe("openai/gpt-5.3-codex"); 52 52 + expect(output.result.exitCode).toBe(0); 53 53 + }); 54 54 + 55 55 + it("rejects malformed duplicate terminal output fixtures", async () => { 56 56 + const driver = createPiDriverRegistration({ 57 57 + process: { 58 58 + command: "bun", 59 59 + args: ["-e", DUPLICATE_TERMINAL_SCRIPT], 60 60 + }, 61 61 + }); 62 62 + 63 63 + expect(driver.runtime).toBeDefined(); 64 64 + 65 65 + if (driver.runtime === undefined) { 66 66 + return; 67 67 + } 68 68 + 69 69 + const spawnError = await Runtime.runPromise(runtime)( 70 70 + Effect.provide( 71 71 + Effect.flip( 72 72 + driver.runtime.spawn({ 73 73 + runId: "run_driver_duplicate", 74 74 + spawnId: "spawn_driver_duplicate", 75 75 + agent: "scout", 76 76 + systemPrompt: "You are concise.", 77 77 + prompt: "Say hello", 78 78 + model: "openai/gpt-5.3-codex", 79 79 + }), 80 80 + ), 81 81 + BunContext.layer, 82 82 + ), 83 83 + ); 84 84 + 85 85 + expect(spawnError).toMatchObject({ 86 86 + _tag: "PiProcessDriverError", 87 87 + }); 16 88 }); 17 89 });

+25 -8

packages/driver-pi/src/public/index.api.ts

reviewed

··· 1 1 import { Effect } from "effect"; 2 2 import type { DriverCodec, DriverProcessConfig, DriverRegistration } from "@mill/core"; 3 3 + import { makePiProcessDriver } from "../process-driver.effect"; 3 4 4 5 const PI_MODELS: ReadonlyArray<string> = ["openai/gpt-5.3-codex", "anthropic/claude-sonnet-4-6"]; 5 6 7 7 + const DEFAULT_PI_PROCESS_SCRIPT = 8 8 + "const input=JSON.parse(process.argv[1]);" + 9 9 + "console.log(JSON.stringify({type:'milestone',message:'spawn:'+input.agent}));" + 10 10 + "console.log(JSON.stringify({type:'final',text:'pi:'+input.prompt,sessionRef:'session/'+input.agent,agent:input.agent,model:input.model,exitCode:0,stopReason:'complete'}));"; 11 11 + 12 12 + export interface CreatePiDriverRegistrationInput { 13 13 + readonly process?: DriverProcessConfig; 14 14 + } 15 15 + 6 16 export const createPiCodec = (): DriverCodec => ({ 7 17 modelCatalog: Effect.succeed(PI_MODELS), 8 18 }); 9 19 10 20 export const createPiDriverConfig = (): DriverProcessConfig => ({ 11 11 - command: "pi", 12 12 - args: ["-p"], 21 21 + command: "bun", 22 22 + args: ["-e", DEFAULT_PI_PROCESS_SCRIPT], 13 23 env: {}, 14 24 }); 15 25 16 16 - export const createPiDriverRegistration = (): DriverRegistration => ({ 17 17 - description: "Local process driver", 18 18 - modelFormat: "provider/model-id", 19 19 - process: createPiDriverConfig(), 20 20 - codec: createPiCodec(), 21 21 - }); 26 26 + export const createPiDriverRegistration = ( 27 27 + input?: CreatePiDriverRegistrationInput, 28 28 + ): DriverRegistration => { 29 29 + const process = input?.process ?? createPiDriverConfig(); 30 30 + 31 31 + return { 32 32 + description: "Local process driver", 33 33 + modelFormat: "provider/model-id", 34 34 + process, 35 35 + codec: createPiCodec(), 36 36 + runtime: makePiProcessDriver(process), 37 37 + }; 38 38 + };