+13 -11

README.md

reviewed

··· 27 27 ``` 28 28 29 29 ```bash 30 30 - mill run review.ts # returns runId, executes in background 31 31 - mill watch abc123 # stream events live 32 32 - mill run review.ts --sync # or block until done 30 30 + mill run review.ts # returns runId, executes in background 31 31 + mill watch --run abc123 # stream events live 32 32 + mill watch --run abc123 --channel io 33 33 + mill run review.ts --sync # or block until done 33 34 ``` 34 35 35 36 ## CLI 36 37 37 38 ``` 38 39 mill run <program.ts> [--sync] [--json] [--driver <name>] 39 39 - mill status <runId> show run state 40 40 - mill wait <runId> --timeout block until complete/failed/cancelled 41 41 - mill watch [--run <runId>] stream tier-1 events (global if --run omitted) 42 42 - mill inspect <id>[.<spawnId>] inspect run or spawn detail 43 43 - mill inspect <id> --session resolve full agent session via driver 44 44 - mill cancel <runId> mark cancelled + kill worker process tree 45 45 - mill ls [--status <filter>] list runs 46 46 - mill init [--global] generate starter config (local or ~/.mill/config.ts) 40 40 + mill status <runId> show run state 41 41 + mill wait <runId> --timeout block until complete/failed/cancelled 42 42 + mill watch [--run <runId>] watch streams (default: events) 43 43 + --channel events|io|all choose stream channel 44 44 + --source driver|program io source filter (io/all only) 45 45 + --spawn <spawnId> io spawn filter (io/all only) 46 46 + mill cancel <runId> mark cancelled + kill worker process tree 47 47 + mill ls [--status <filter>] list runs 48 48 + mill init [--global] generate starter config (local or ~/.mill/config.ts) 47 49 ``` 48 50 49 51 All commands accept `--json` for machine-readable output on stdout (diagnostics go to stderr).

+61 -138

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

reviewed

··· 64 64 }), 65 65 ); 66 66 67 67 - const InspectRunEnvelope = Schema.parseJson( 68 68 - Schema.Struct({ 69 69 - kind: Schema.Literal("run"), 70 70 - run: Schema.Struct({ 71 71 - id: Schema.String, 72 72 - status: Schema.String, 73 73 - }), 74 74 - events: Schema.Array( 75 75 - Schema.Struct({ 67 67 + const WatchOutputEnvelope = Schema.parseJson( 68 68 + Schema.Union( 69 69 + Schema.Struct({ 70 70 + kind: Schema.Literal("event"), 71 71 + runId: Schema.String, 72 72 + event: Schema.Struct({ 76 73 type: Schema.String, 77 77 - sequence: Schema.Number, 78 78 - }), 79 79 - ), 80 80 - }), 81 81 - ); 82 82 - 83 83 - const InspectSpawnEnvelope = Schema.parseJson( 84 84 - Schema.Struct({ 85 85 - kind: Schema.Literal("spawn"), 86 86 - runId: Schema.String, 87 87 - spawnId: Schema.String, 88 88 - result: Schema.optional( 89 89 - Schema.Struct({ 90 90 - sessionRef: Schema.String, 74 74 + runId: Schema.String, 91 75 }), 92 92 - ), 93 93 - }), 94 94 - ); 95 95 - 96 96 - const SessionEnvelope = Schema.parseJson( 97 97 - Schema.Struct({ 98 98 - runId: Schema.String, 99 99 - spawnId: Schema.String, 100 100 - sessionRef: Schema.String, 101 101 - pointer: Schema.String, 102 102 - driver: Schema.String, 103 103 - }), 76 76 + }), 77 77 + Schema.Struct({ 78 78 + kind: Schema.Literal("io"), 79 79 + runId: Schema.String, 80 80 + source: Schema.Union(Schema.Literal("driver"), Schema.Literal("program")), 81 81 + stream: Schema.Union(Schema.Literal("stdout"), Schema.Literal("stderr")), 82 82 + line: Schema.String, 83 83 + timestamp: Schema.String, 84 84 + spawnId: Schema.optional(Schema.String), 85 85 + }), 86 86 + ), 104 87 ); 105 88 106 89 const CancelEnvelope = Schema.parseJson( ··· 129 112 timeoutSeconds: Schema.Number, 130 113 message: Schema.String, 131 114 }), 132 132 - }), 133 133 - ); 134 134 - 135 135 - const WatchEventEnvelope = Schema.parseJson( 136 136 - Schema.Struct({ 137 137 - type: Schema.String, 138 138 - runId: Schema.String, 139 115 }), 140 116 ); 141 117 ··· 498 474 } 499 475 }); 500 476 501 501 - it("supports watch/inspect/session/ls JSON contracts for completed runs", async () => { 502 502 - const tempDirectory = await mkdtemp(join(tmpdir(), "mill-cli-inspect-watch-")); 477 477 + it("supports watch/ls JSON contracts for completed runs", async () => { 478 478 + const tempDirectory = await mkdtemp(join(tmpdir(), "mill-cli-watch-")); 503 479 const homeDirectory = join(tempDirectory, "home"); 504 480 const programPath = join(tempDirectory, "program.ts"); 505 481 ··· 534 510 const runPayload = Schema.decodeUnknownSync(RunSyncEnvelope)(runStdout[0]); 535 511 536 512 const watchStdout: Array<string> = []; 537 537 - const watchCode = await runCli(["watch", "--run", runPayload.run.id, "--json"], { 538 538 - cwd: tempDirectory, 539 539 - homeDirectory, 540 540 - pathExists: async () => false, 541 541 - io: { 542 542 - stdout: (line) => { 543 543 - watchStdout.push(line); 544 544 - }, 545 545 - stderr: () => undefined, 546 546 - }, 547 547 - }); 548 548 - 549 549 - expect(watchCode).toBe(0); 550 550 - expect(watchStdout.length).toBeGreaterThan(0); 551 551 - 552 552 - for (const line of watchStdout) { 553 553 - const parsed = Schema.decodeUnknownSync(WatchEventEnvelope)(line); 554 554 - expect(parsed.runId).toBe(runPayload.run.id); 555 555 - expect(typeof parsed.type).toBe("string"); 556 556 - } 557 557 - 558 558 - const inspectRunStdout: Array<string> = []; 559 559 - const inspectRunCode = await runCli(["inspect", runPayload.run.id, "--json"], { 560 560 - cwd: tempDirectory, 561 561 - homeDirectory, 562 562 - pathExists: async () => false, 563 563 - io: { 564 564 - stdout: (line) => { 565 565 - inspectRunStdout.push(line); 566 566 - }, 567 567 - stderr: () => undefined, 568 568 - }, 569 569 - }); 570 570 - 571 571 - expect(inspectRunCode).toBe(0); 572 572 - const inspectedRun = Schema.decodeUnknownSync(InspectRunEnvelope)(inspectRunStdout[0]); 573 573 - expect(inspectedRun.run.id).toBe(runPayload.run.id); 574 574 - expect(inspectedRun.events.length).toBeGreaterThan(0); 575 575 - 576 576 - const inspectSpawnStdout: Array<string> = []; 577 577 - const inspectSpawnCode = await runCli(["inspect", `${runPayload.run.id}.spawn_1`, "--json"], { 578 578 - cwd: tempDirectory, 579 579 - homeDirectory, 580 580 - pathExists: async () => false, 581 581 - io: { 582 582 - stdout: (line) => { 583 583 - inspectSpawnStdout.push(line); 584 584 - }, 585 585 - stderr: () => undefined, 586 586 - }, 587 587 - }); 588 588 - 589 589 - expect(inspectSpawnCode).toBe(0); 590 590 - const inspectedSpawn = Schema.decodeUnknownSync(InspectSpawnEnvelope)(inspectSpawnStdout[0]); 591 591 - expect(inspectedSpawn.spawnId).toBe("spawn_1"); 592 592 - expect(inspectedSpawn.result?.sessionRef.length).toBeGreaterThan(0); 593 593 - 594 594 - const inspectSessionStdout: Array<string> = []; 595 595 - const inspectSessionCode = await runCli( 596 596 - ["inspect", `${runPayload.run.id}.spawn_1`, "--session", "--json"], 513 513 + const watchCode = await runCli( 514 514 + ["watch", "--run", runPayload.run.id, "--channel", "all", "--json"], 597 515 { 598 516 cwd: tempDirectory, 599 517 homeDirectory, 600 518 pathExists: async () => false, 601 519 io: { 602 520 stdout: (line) => { 603 603 - inspectSessionStdout.push(line); 521 521 + watchStdout.push(line); 604 522 }, 605 523 stderr: () => undefined, 606 524 }, 607 525 }, 608 526 ); 609 527 610 610 - expect(inspectSessionCode).toBe(0); 611 611 - const sessionPayload = Schema.decodeUnknownSync(SessionEnvelope)(inspectSessionStdout[0]); 612 612 - expect(sessionPayload.runId).toBe(runPayload.run.id); 613 613 - expect(sessionPayload.spawnId).toBe("spawn_1"); 614 614 - expect(sessionPayload.pointer.length).toBeGreaterThan(0); 528 528 + expect(watchCode).toBe(0); 529 529 + expect(watchStdout.length).toBeGreaterThan(0); 530 530 + 531 531 + const parsedWatch = watchStdout.map((line) => 532 532 + Schema.decodeUnknownSync(WatchOutputEnvelope)(line), 533 533 + ); 534 534 + expect(parsedWatch.every((entry) => entry.runId === runPayload.run.id)).toBe(true); 535 535 + expect( 536 536 + parsedWatch.some((entry) => entry.kind === "event" && entry.event.type === "run:complete"), 537 537 + ).toBe(true); 615 538 616 539 const listStdout: Array<string> = []; 617 540 const listCode = await runCli(["ls", "--json"], { ··· 697 620 const secondCancel = Schema.decodeUnknownSync(CancelEnvelope)(cancelStdout2[0]); 698 621 expect(secondCancel.runId).toBe(submittedRun.runId); 699 622 700 700 - const inspectStdout: Array<string> = []; 701 701 - const inspectCode = await runCli(["inspect", submittedRun.runId, "--json"], { 702 702 - cwd: tempDirectory, 703 703 - homeDirectory, 704 704 - pathExists: async () => false, 705 705 - io: { 706 706 - stdout: (line) => { 707 707 - inspectStdout.push(line); 708 708 - }, 709 709 - stderr: () => undefined, 710 710 - }, 711 711 - }); 623 623 + const eventsContent = await readFile(submittedRun.paths.eventsFile, "utf-8"); 624 624 + const persistedEvents = eventsContent 625 625 + .split("\n") 626 626 + .map((line) => line.trim()) 627 627 + .filter((line) => line.length > 0) 628 628 + .map((line) => JSON.parse(line) as { type?: string }); 712 629 713 713 - expect(inspectCode).toBe(0); 714 714 - const inspectedRun = Schema.decodeUnknownSync(InspectRunEnvelope)(inspectStdout[0]); 715 715 - const terminalEvents = inspectedRun.events.filter( 630 630 + const terminalEvents = persistedEvents.filter( 716 631 (event) => 717 632 event.type === "run:complete" || 718 633 event.type === "run:failed" || ··· 743 658 } 744 659 }); 745 660 746 746 - it("watch --raw streams tier-2 passthrough without persistence", async () => { 747 747 - const tempDirectory = await mkdtemp(join(tmpdir(), "mill-cli-watch-raw-")); 661 661 + it("watch --channel io streams io envelopes without persisted event payloads", async () => { 662 662 + const tempDirectory = await mkdtemp(join(tmpdir(), "mill-cli-watch-io-")); 748 663 const homeDirectory = join(tempDirectory, "home"); 749 664 const programPath = join(tempDirectory, "program.ts"); 750 665 ··· 806 721 const submittedRun = Schema.decodeUnknownSync(RunSubmitEnvelope)(runStdout[0]); 807 722 808 723 const watchStdout: Array<string> = []; 809 809 - const watchCode = await runCli(["watch", "--run", submittedRun.runId, "--raw"], { 810 810 - cwd: tempDirectory, 811 811 - homeDirectory, 812 812 - pathExists: async () => false, 813 813 - io: { 814 814 - stdout: (line) => { 815 815 - watchStdout.push(line); 724 724 + const watchCode = await runCli( 725 725 + ["watch", "--run", submittedRun.runId, "--channel", "io", "--json"], 726 726 + { 727 727 + cwd: tempDirectory, 728 728 + homeDirectory, 729 729 + pathExists: async () => false, 730 730 + io: { 731 731 + stdout: (line) => { 732 732 + watchStdout.push(line); 733 733 + }, 734 734 + stderr: () => undefined, 816 735 }, 817 817 - stderr: () => undefined, 818 736 }, 819 819 - }); 737 737 + ); 820 738 821 739 expect(watchCode).toBe(0); 822 740 expect(watchStdout.length).toBeGreaterThan(0); 823 741 742 742 + const parsedWatch = watchStdout.map((line) => 743 743 + Schema.decodeUnknownSync(WatchOutputEnvelope)(line), 744 744 + ); 745 745 + expect(parsedWatch.every((entry) => entry.kind === "io")).toBe(true); 746 746 + 824 747 const eventsContent = await readFile(submittedRun.paths.eventsFile, "utf-8"); 825 825 - for (const rawLine of watchStdout) { 826 826 - expect(eventsContent.includes(rawLine.trim())).toBe(false); 748 748 + for (const line of watchStdout) { 749 749 + expect(eventsContent.includes(line.trim())).toBe(false); 827 750 } 828 751 } finally { 829 752 await rm(tempDirectory, { recursive: true, force: true });

+18 -55

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

reviewed

··· 8 8 cancelRun, 9 9 defineConfig, 10 10 getRunStatus, 11 11 - inspectRun, 12 11 listRuns, 13 12 processDriver, 14 13 resolveConfig, ··· 481 480 return 1; 482 481 }; 483 482 483 483 + const WATCH_CHANNELS = ["events", "io", "all"] as const; 484 484 + const WATCH_SOURCES = ["driver", "program"] as const; 485 485 + 486 486 + type WatchChannel = (typeof WATCH_CHANNELS)[number]; 487 487 + type WatchSource = (typeof WATCH_SOURCES)[number]; 488 488 + 484 489 interface WatchCommandInput { 485 490 readonly run: Option.Option<string>; 486 491 readonly sinceTime: Option.Option<string>; 492 492 + readonly channel: Option.Option<WatchChannel>; 493 493 + readonly source: Option.Option<WatchSource>; 494 494 + readonly spawn: Option.Option<string>; 487 495 readonly json: boolean; 488 488 - readonly raw: boolean; 489 496 readonly runsDir: Option.Option<string>; 490 497 readonly driver: Option.Option<string>; 491 498 } ··· 498 505 await watchRun({ 499 506 defaults: defaultConfig, 500 507 runId: fromOption(command.run), 508 508 + channel: fromOption(command.channel), 509 509 + source: fromOption(command.source), 510 510 + spawnId: fromOption(command.spawn), 501 511 sinceTimeIso: fromOption(command.sinceTime), 502 502 - raw: command.raw, 503 512 cwd: options.cwd, 504 513 homeDirectory: options.homeDirectory, 505 514 runsDirectory: fromOption(command.runsDir) ?? options.runsDirectory, ··· 514 523 return 0; 515 524 }; 516 525 517 517 - interface InspectCommandInput { 518 518 - readonly ref: string; 519 519 - readonly json: boolean; 520 520 - readonly session: boolean; 521 521 - readonly runsDir: Option.Option<string>; 522 522 - readonly driver: Option.Option<string>; 523 523 - } 524 524 - 525 525 - const inspectCommand = async ( 526 526 - command: InspectCommandInput, 527 527 - options: RunCliOptions, 528 528 - io: CliIo, 529 529 - ): Promise<number> => { 530 530 - const inspected = await inspectRun({ 531 531 - defaults: defaultConfig, 532 532 - ref: command.ref, 533 533 - session: command.session, 534 534 - cwd: options.cwd, 535 535 - homeDirectory: options.homeDirectory, 536 536 - runsDirectory: fromOption(command.runsDir) ?? options.runsDirectory, 537 537 - driverName: fromOption(command.driver), 538 538 - pathExists: options.pathExists, 539 539 - loadConfigModule: options.loadConfigModule, 540 540 - }); 541 541 - 542 542 - if (command.json) { 543 543 - io.stdout(JSON.stringify(inspected)); 544 544 - return 0; 545 545 - } 546 546 - 547 547 - io.stdout(JSON.stringify(inspected, null, 2)); 548 548 - return 0; 549 549 - }; 550 550 - 551 526 interface CancelCommandInput { 552 527 readonly runId: string; 553 528 readonly json: boolean; ··· 676 651 { 677 652 run: optionalTextOption("run"), 678 653 sinceTime: optionalTextOption("since-time"), 654 654 + channel: Options.choice("channel", WATCH_CHANNELS).pipe(Options.optional), 655 655 + source: Options.choice("source", WATCH_SOURCES).pipe(Options.optional), 656 656 + spawn: optionalTextOption("spawn"), 679 657 json: Options.boolean("json"), 680 680 - raw: Options.boolean("raw"), 681 658 runsDir: optionalTextOption("runs-dir"), 682 659 driver: optionalTextOption("driver"), 683 660 }, 684 661 (command) => toCliEffect(watchCommand(command, options, io)), 685 662 ).pipe( 686 663 CliCommand.withDescription( 687 687 - "Stream run events. Use --run <runId> for scoped watch, omit for global watch.", 664 664 + "Watch run streams. --channel events|io|all (default: events). --channel io|all requires --run.", 688 665 ), 689 666 ); 690 667 691 691 - const inspect = CliCommand.make( 692 692 - "inspect", 693 693 - { 694 694 - ref: Args.text({ name: "runId[.spawnId]" }), 695 695 - json: Options.boolean("json"), 696 696 - session: Options.boolean("session"), 697 697 - runsDir: optionalTextOption("runs-dir"), 698 698 - driver: optionalTextOption("driver"), 699 699 - }, 700 700 - (command) => toCliEffect(inspectCommand(command, options, io)), 701 701 - ).pipe(CliCommand.withDescription("Inspect run, spawn, or session output.")); 702 702 - 703 668 const cancel = CliCommand.make( 704 669 "cancel", 705 670 { ··· 736 701 737 702 return CliCommand.make("mill").pipe( 738 703 CliCommand.withDescription("Mill orchestration runtime."), 739 739 - CliCommand.withSubcommands([run, status, wait, watch, inspect, cancel, ls, init, worker]), 704 704 + CliCommand.withSubcommands([run, status, wait, watch, cancel, ls, init, worker]), 740 705 ); 741 706 }; 742 707 ··· 763 728 run <program.ts> Run a mill program 764 729 status <runId> Show run state 765 730 wait <runId> --timeout <s> Wait for terminal state 766 766 - watch [--run <runId>] Stream run events (global if --run omitted) 767 767 - inspect <ref> Inspect run or spawn detail 731 731 + watch [--run <runId>] Watch events/io streams (use --channel events|io|all) 768 732 cancel <runId> Cancel a running execution 769 733 ls List runs 770 734 init [--global] Create starter config (local or ~/.mill/config.ts) ··· 802 766 "status", 803 767 "wait", 804 768 "watch", 805 805 - "inspect", 806 769 "cancel", 807 770 "ls", 808 771 "init",

+45 -56

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

reviewed

··· 68 68 }), 69 69 ); 70 70 71 71 - const InspectRunEnvelope = Schema.parseJson( 72 72 - Schema.Struct({ 73 73 - kind: Schema.Literal("run"), 74 74 - run: Schema.Struct({ 75 75 - id: Schema.String, 76 76 - status: Schema.String, 77 77 - }), 78 78 - events: Schema.Array( 79 79 - Schema.Struct({ 71 71 + const WatchOutputEnvelope = Schema.parseJson( 72 72 + Schema.Union( 73 73 + Schema.Struct({ 74 74 + kind: Schema.Literal("event"), 75 75 + runId: Schema.String, 76 76 + event: Schema.Struct({ 80 77 type: Schema.String, 81 81 - sequence: Schema.Number, 78 78 + runId: Schema.String, 82 79 }), 83 83 - ), 84 84 - }), 85 85 - ); 86 86 - 87 87 - const SessionEnvelope = Schema.parseJson( 88 88 - Schema.Struct({ 89 89 - runId: Schema.String, 90 90 - spawnId: Schema.String, 91 91 - sessionRef: Schema.String, 92 92 - pointer: Schema.String, 93 93 - driver: Schema.String, 94 94 - }), 80 80 + }), 81 81 + Schema.Struct({ 82 82 + kind: Schema.Literal("io"), 83 83 + runId: Schema.String, 84 84 + source: Schema.Union(Schema.Literal("driver"), Schema.Literal("program")), 85 85 + stream: Schema.Union(Schema.Literal("stdout"), Schema.Literal("stderr")), 86 86 + line: Schema.String, 87 87 + timestamp: Schema.String, 88 88 + spawnId: Schema.optional(Schema.String), 89 89 + }), 90 90 + ), 95 91 ); 96 92 97 93 const ListEnvelope = Schema.parseJson( ··· 132 128 expect(output).toContain("Usage: mill <command>"); 133 129 expect(output).toContain("Commands:"); 134 130 expect(output).toContain("run <program.ts>"); 131 131 + expect(output).not.toContain("inspect <ref>"); 135 132 expect(output).not.toContain("discovery"); 136 133 expect(output).not.toContain("Effect-first"); 137 134 }); ··· 403 400 } 404 401 }, 20_000); 405 402 406 406 - it("runs inspect/session/cancel/watch matrix across concurrent runs", async () => { 403 403 + it("runs cancel/watch/list matrix across concurrent runs", async () => { 407 404 const tempDirectory = await mkdtemp(join(tmpdir(), "mill-cli-matrix-e2e-")); 408 405 const runsDirectory = join(tempDirectory, "runs"); 409 406 const completeProgramPath = join(tempDirectory, "complete.ts"); ··· 522 519 "watch", 523 520 "--run", 524 521 completeRun.runId, 522 522 + "--channel", 523 523 + "all", 525 524 "--json", 526 525 "--runs-dir", 527 526 runsDirectory, ··· 531 530 const watchLines = watchOutput 532 531 .split("\n") 533 532 .map((line) => line.trim()) 534 534 - .filter((line) => line.length > 0); 533 533 + .filter((line) => line.length > 0) 534 534 + .map((line) => Schema.decodeUnknownSync(WatchOutputEnvelope)(line)); 535 535 536 536 expect(watchLines.length).toBeGreaterThan(0); 537 537 - const watchTerminalCount = watchLines 538 538 - .map((line) => Schema.decodeUnknownSync(EventTypeEnvelope)(line)) 539 539 - .filter( 540 540 - (event) => 541 541 - event.type === "run:complete" || 542 542 - event.type === "run:failed" || 543 543 - event.type === "run:cancelled", 544 544 - ).length; 537 537 + const watchTerminalCount = watchLines.filter( 538 538 + (entry) => 539 539 + entry.kind === "event" && 540 540 + (entry.event.type === "run:complete" || 541 541 + entry.event.type === "run:failed" || 542 542 + entry.event.type === "run:cancelled"), 543 543 + ).length; 545 544 expect(watchTerminalCount).toBe(1); 546 545 547 547 - const inspectCancelledOutput = await commandOutput( 546 546 + const watchCancelledOutput = await commandOutput( 548 547 Command.make( 549 548 "bun", 550 549 "run", 551 550 "packages/cli/src/bin/mill.ts", 552 552 - "inspect", 551 551 + "watch", 552 552 + "--run", 553 553 cancelRun.runId, 554 554 + "--channel", 555 555 + "events", 554 556 "--json", 555 557 "--runs-dir", 556 558 runsDirectory, 557 559 ), 558 560 ); 559 561 560 560 - const inspectedCancelled = 561 561 - Schema.decodeUnknownSync(InspectRunEnvelope)(inspectCancelledOutput); 562 562 - expect(inspectedCancelled.run.status).toBe("cancelled"); 562 562 + const cancelledLines = watchCancelledOutput 563 563 + .split("\n") 564 564 + .map((line) => line.trim()) 565 565 + .filter((line) => line.length > 0) 566 566 + .map((line) => Schema.decodeUnknownSync(WatchOutputEnvelope)(line)); 567 567 + 563 568 expect( 564 564 - inspectedCancelled.events.filter((event) => event.type === "run:cancelled"), 565 565 - ).toHaveLength(1); 566 566 - 567 567 - const inspectSessionOutput = await commandOutput( 568 568 - Command.make( 569 569 - "bun", 570 570 - "run", 571 571 - "packages/cli/src/bin/mill.ts", 572 572 - "inspect", 573 573 - `${completeRun.runId}.spawn_1`, 574 574 - "--session", 575 575 - "--json", 576 576 - "--runs-dir", 577 577 - runsDirectory, 569 569 + cancelledLines.some( 570 570 + (entry) => entry.kind === "event" && entry.event.type === "run:cancelled", 578 571 ), 579 579 - ); 580 580 - 581 581 - const sessionPayload = Schema.decodeUnknownSync(SessionEnvelope)(inspectSessionOutput); 582 582 - expect(sessionPayload.runId).toBe(completeRun.runId); 583 583 - expect(sessionPayload.spawnId).toBe("spawn_1"); 572 572 + ).toBe(true); 584 573 585 574 const listOutput = await commandOutput( 586 575 Command.make(

+10 -10

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

reviewed

··· 370 370 } 371 371 }); 372 372 373 373 - it("watch and watchRaw surface tier-1 persisted events and tier-2 raw passthrough", async () => { 373 373 + it("watch and watchIo surface tier-1 persisted events and io passthrough", async () => { 374 374 const runsDirectory = await mkdtemp(join(tmpdir(), "mill-engine-watch-")); 375 375 const runId = decodeRunIdSync(`run_${crypto.randomUUID()}`); 376 376 ··· 428 428 ), 429 429 ); 430 430 431 431 - const watchRawEffect = Effect.scoped( 432 432 - Stream.runCollect(Stream.take(engine.watchRaw(runId), 2)), 433 433 - ); 431 431 + const watchIoEffect = Effect.scoped(Stream.runCollect(Stream.take(engine.watchIo(runId), 2))); 434 432 435 433 const executionEffect = engine.runSync({ 436 434 runId, ··· 446 444 ), 447 445 }); 448 446 449 449 - const [tier1EventsChunk, rawEventsChunk] = await runWithBunContext( 447 447 + const [tier1EventsChunk, ioEventsChunk] = await runWithBunContext( 450 448 Effect.map( 451 451 - Effect.all([watchTier1Effect, watchRawEffect, executionEffect], { 449 449 + Effect.all([watchTier1Effect, watchIoEffect, executionEffect], { 452 450 concurrency: "unbounded", 453 451 }), 454 454 - ([tier1Events, rawEvents]) => [tier1Events, rawEvents] as const, 452 452 + ([tier1Events, ioEvents]) => [tier1Events, ioEvents] as const, 455 453 ), 456 454 ); 457 455 458 456 const tier1Events = [...tier1EventsChunk]; 459 459 - const rawEvents = [...rawEventsChunk]; 457 457 + const ioEvents = [...ioEventsChunk]; 460 458 461 459 expect(tier1Events.some((event) => event.type === "run:start")).toBe(true); 462 460 expect(tier1Events.some((event) => event.type === "run:complete")).toBe(true); 463 463 - expect(rawEvents).toHaveLength(2); 464 464 - expect(rawEvents[0]).toContain("raw:scout"); 461 461 + expect(ioEvents).toHaveLength(2); 462 462 + expect(ioEvents[0]?.source).toBe("driver"); 463 463 + expect(ioEvents[0]?.stream).toBe("stdout"); 464 464 + expect(ioEvents[0]?.line).toContain("raw:scout"); 465 465 466 466 const eventsFile = await readFile(join(runsDirectory, runId, "events.ndjson"), "utf-8"); 467 467 expect(eventsFile.includes('"type":"final"')).toBe(false);

+18 -6

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

reviewed

··· 30 30 type RunStore, 31 31 } from "./run-store.effect"; 32 32 import { 33 33 - publishRawEvent, 33 33 + publishIoEvent, 34 34 publishTier1Event, 35 35 - watchRawLive, 35 35 + watchIoLive, 36 36 watchTier1GlobalLive, 37 37 watchTier1Live, 38 38 + type IoStreamEvent, 38 39 } from "./observer-hub.effect"; 39 40 40 41 export class ConfigError extends Data.TaggedError("ConfigError")<{ message: string }> {} ··· 118 119 ) => Effect.Effect<ReadonlyArray<RunSyncOutput["run"]>, PersistenceError>; 119 120 readonly watch: (runId: RunId) => Stream.Stream<MillEvent, RunNotFoundError | PersistenceError>; 120 121 readonly watchAll: (sinceTimeIso?: string) => Stream.Stream<MillEvent, PersistenceError>; 121 121 - readonly watchRaw: (runId: RunId) => Stream.Stream<string, RunNotFoundError | PersistenceError>; 122 122 + readonly watchIo: ( 123 123 + runId: RunId, 124 124 + ) => Stream.Stream<IoStreamEvent, RunNotFoundError | PersistenceError>; 122 125 readonly inspect: ( 123 126 ref: InspectRef, 124 127 ) => Effect.Effect<InspectResult, RunNotFoundError | PersistenceError>; ··· 800 803 }); 801 804 802 805 for (const rawLine of driverOutputExit.value.raw ?? []) { 803 803 - yield* publishRawEvent(runInput.runId, rawLine); 806 806 + const timestamp = yield* toIsoTimestamp; 807 807 + 808 808 + yield* publishIoEvent({ 809 809 + runId: runInput.runId, 810 810 + source: "driver", 811 811 + stream: "stdout", 812 812 + line: rawLine, 813 813 + timestamp, 814 814 + spawnId, 815 815 + }); 804 816 } 805 817 806 818 for (const driverEvent of driverOutputExit.value.events) { ··· 1069 1081 }), 1070 1082 ), 1071 1083 1072 1072 - watchRaw: (runId) => 1084 1084 + watchIo: (runId) => 1073 1085 Stream.unwrapScoped( 1074 1074 - Effect.zipRight(runStore.getRun(runId), Effect.succeed(watchRawLive(runId))), 1086 1086 + Effect.zipRight(runStore.getRun(runId), Effect.succeed(watchIoLive(runId))), 1075 1087 ), 1076 1088 1077 1089 inspect: (ref) =>

+20 -9

packages/core/src/internal/observer-hub.effect.ts

reviewed

··· 1 1 import { Effect, PubSub, Stream } from "effect"; 2 2 import type { MillEvent } from "../domain/event.schema"; 3 3 4 4 + export interface IoStreamEvent { 5 5 + readonly runId: string; 6 6 + readonly source: "driver" | "program"; 7 7 + readonly stream: "stdout" | "stderr"; 8 8 + readonly line: string; 9 9 + readonly timestamp: string; 10 10 + readonly spawnId?: string; 11 11 + } 12 12 + 4 13 const tier1PubSubByRun = new Map<string, PubSub.PubSub<MillEvent>>(); 5 5 - const rawPubSubByRun = new Map<string, PubSub.PubSub<string>>(); 14 14 + const ioPubSubByRun = new Map<string, PubSub.PubSub<IoStreamEvent>>(); 6 15 let tier1GlobalPubSub: PubSub.PubSub<MillEvent> | undefined; 7 16 8 17 const ensureTier1PubSub = (runId: string): Effect.Effect<PubSub.PubSub<MillEvent>> => ··· 20 29 ); 21 30 }); 22 31 23 23 - const ensureRawPubSub = (runId: string): Effect.Effect<PubSub.PubSub<string>> => 32 32 + const ensureIoPubSub = (runId: string): Effect.Effect<PubSub.PubSub<IoStreamEvent>> => 24 33 Effect.suspend(() => { 25 25 - const existing = rawPubSubByRun.get(runId); 34 34 + const existing = ioPubSubByRun.get(runId); 26 35 27 36 if (existing !== undefined) { 28 37 return Effect.succeed(existing); 29 38 } 30 39 31 31 - return Effect.tap(PubSub.unbounded<string>(), (pubSub) => 40 40 + return Effect.tap(PubSub.unbounded<IoStreamEvent>(), (pubSub) => 32 41 Effect.sync(() => { 33 33 - rawPubSubByRun.set(runId, pubSub); 42 42 + ioPubSubByRun.set(runId, pubSub); 34 43 }), 35 44 ); 36 45 }); ··· 58 67 ), 59 68 ); 60 69 61 61 - export const publishRawEvent = (runId: string, raw: string): Effect.Effect<void> => 62 62 - Effect.asVoid(Effect.flatMap(ensureRawPubSub(runId), (pubSub) => PubSub.publish(pubSub, raw))); 70 70 + export const publishIoEvent = (event: IoStreamEvent): Effect.Effect<void> => 71 71 + Effect.asVoid( 72 72 + Effect.flatMap(ensureIoPubSub(event.runId), (pubSub) => PubSub.publish(pubSub, event)), 73 73 + ); 63 74 64 75 export const watchTier1Live = (runId: string): Stream.Stream<MillEvent, never> => 65 76 Stream.unwrapScoped(Effect.map(ensureTier1PubSub(runId), (pubSub) => Stream.fromPubSub(pubSub))); 66 77 67 67 - export const watchRawLive = (runId: string): Stream.Stream<string, never> => 68 68 - Stream.unwrapScoped(Effect.map(ensureRawPubSub(runId), (pubSub) => Stream.fromPubSub(pubSub))); 78 78 + export const watchIoLive = (runId: string): Stream.Stream<IoStreamEvent, never> => 79 79 + Stream.unwrapScoped(Effect.map(ensureIoPubSub(runId), (pubSub) => Stream.fromPubSub(pubSub))); 69 80 70 81 export const watchTier1GlobalLive = (): Stream.Stream<MillEvent, never> => 71 82 Stream.unwrapScoped(Effect.map(ensureTier1GlobalPubSub(), (pubSub) => Stream.fromPubSub(pubSub)));

+1

packages/core/src/observer-hub.effect.ts

reviewed

··· 1 1 + export * from "./internal/observer-hub.effect";

+155 -128

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

reviewed

··· 2 2 import * as fs from "node:fs"; 3 3 import * as FileSystem from "@effect/platform/FileSystem"; 4 4 import * as BunContext from "@effect/platform-bun/BunContext"; 5 5 - import { Effect, Runtime, Stream } from "effect"; 6 6 - import { makeMillEngine, ProgramExecutionError, type InspectResult } from "../engine.effect"; 5 5 + import { Effect, Fiber, Runtime, Stream } from "effect"; 6 6 + import { makeMillEngine, ProgramExecutionError } from "../engine.effect"; 7 7 import { makeDriverRegistry } from "../driver-registry.effect"; 8 8 import { makeExecutorRegistry } from "../executor-registry.effect"; 9 9 - import { 10 10 - decodeRunIdSync, 11 11 - decodeSpawnIdSync, 12 12 - type RunRecord, 13 13 - type RunSyncOutput, 14 14 - } from "../run.schema"; 9 9 + import { type MillEvent } from "../event.schema"; 10 10 + import { decodeRunIdSync, type RunRecord, type RunSyncOutput } from "../run.schema"; 15 11 import { runDetachedWorker } from "../worker.effect"; 16 12 import { executeProgramInProcessHost } from "../program-host.effect"; 13 13 + import { publishIoEvent, type IoStreamEvent } from "../observer-hub.effect"; 17 14 import { resolveConfig } from "./config-loader.api"; 18 18 - import type { 19 19 - DriverSessionPointer, 20 20 - ExecutorRuntime, 21 21 - ExtensionRegistration, 22 22 - ResolveConfigOptions, 23 23 - } from "./types"; 15 15 + import type { ExecutorRuntime, ExtensionRegistration, ResolveConfigOptions } from "./types"; 24 16 25 17 const runtime = Runtime.defaultRuntime; 26 18 ··· 53 45 readonly timeoutSeconds: number; 54 46 } 55 47 48 48 + export type WatchChannel = "events" | "io" | "all"; 49 49 + export type WatchSource = "driver" | "program"; 50 50 + 51 51 + export type WatchOutput = 52 52 + | { 53 53 + readonly kind: "event"; 54 54 + readonly runId: string; 55 55 + readonly event: MillEvent; 56 56 + } 57 57 + | { 58 58 + readonly kind: "io"; 59 59 + readonly runId: string; 60 60 + readonly source: WatchSource; 61 61 + readonly stream: "stdout" | "stderr"; 62 62 + readonly line: string; 63 63 + readonly timestamp: string; 64 64 + readonly spawnId?: string; 65 65 + }; 66 66 + 56 67 export interface WatchRunInput extends Omit< 57 68 GetRunStatusInput, 58 69 "runId" | "pathExists" | "loadConfigModule" 59 70 > { 60 71 readonly runId?: string; 61 61 - readonly raw?: boolean; 72 72 + readonly channel?: WatchChannel; 73 73 + readonly source?: WatchSource; 74 74 + readonly spawnId?: string; 62 75 readonly sinceTimeIso?: string; 63 76 readonly pathExists?: (path: string) => Promise<boolean>; 64 77 readonly loadConfigModule?: (path: string) => Promise<unknown>; 65 78 readonly onEvent: (line: string) => void; 66 66 - } 67 67 - 68 68 - export interface InspectRunInput extends BaseRunInput { 69 69 - readonly ref: string; 70 70 - readonly session?: boolean; 71 79 } 72 80 73 81 export interface CancelRunInput extends GetRunStatusInput { ··· 404 412 }; 405 413 }; 406 414 407 407 - const parseInspectRef = ( 408 408 - ref: string, 409 409 - ): 410 410 - | { 411 411 - runId: string; 412 412 - spawnId?: string; 413 413 - } 414 414 - | undefined => { 415 415 - const [runIdPart, spawnIdPart] = ref.split("."); 416 416 - 417 417 - if (runIdPart === undefined || runIdPart.length === 0) { 418 418 - return undefined; 419 419 - } 420 420 - 421 421 - if (spawnIdPart === undefined || spawnIdPart.length === 0) { 422 422 - return { 423 423 - runId: runIdPart, 424 424 - }; 425 425 - } 426 426 - 427 427 - return { 428 428 - runId: runIdPart, 429 429 - spawnId: spawnIdPart, 430 430 - }; 431 431 - }; 432 432 - 433 415 const isRunTerminalEvent = (eventType: string): boolean => 434 416 eventType === "run:complete" || eventType === "run:failed" || eventType === "run:cancelled"; 435 417 ··· 443 425 return new Date(parsed).toISOString() === value; 444 426 }; 445 427 446 446 - export interface InspectSessionOutput extends DriverSessionPointer { 447 447 - readonly runId: string; 448 448 - readonly spawnId: string; 449 449 - } 428 428 + const toWatchEventOutput = (event: MillEvent): WatchOutput => ({ 429 429 + kind: "event", 430 430 + runId: event.runId, 431 431 + event, 432 432 + }); 433 433 + 434 434 + const toWatchIoOutput = (event: IoStreamEvent): WatchOutput => ({ 435 435 + kind: "io", 436 436 + runId: event.runId, 437 437 + source: event.source, 438 438 + stream: event.stream, 439 439 + line: event.line, 440 440 + timestamp: event.timestamp, 441 441 + spawnId: event.spawnId, 442 442 + }); 443 443 + 444 444 + const emitWatchOutput = ( 445 445 + onEvent: (line: string) => void, 446 446 + output: WatchOutput, 447 447 + ): Effect.Effect<void> => 448 448 + Effect.sync(() => { 449 449 + onEvent(JSON.stringify(output)); 450 450 + }); 451 451 + 452 452 + const filterIoEvent = ( 453 453 + event: IoStreamEvent, 454 454 + source: WatchSource | undefined, 455 455 + spawnId: string | undefined, 456 456 + ): boolean => { 457 457 + if (source !== undefined && event.source !== source) { 458 458 + return false; 459 459 + } 460 460 + 461 461 + if (spawnId !== undefined && event.spawnId !== spawnId) { 462 462 + return false; 463 463 + } 464 464 + 465 465 + return true; 466 466 + }; 450 467 451 468 export const submitRun = async (input: SubmitRunInput): Promise<RunRecord> => { 452 469 const cwd = input.cwd ?? process.cwd(); ··· 543 560 executorName: engineContext.selectedExecutorName, 544 561 extensions: engineContext.selectedExtensions, 545 562 spawn, 563 563 + onIo: ({ stream, line }) => 564 564 + Effect.flatMap( 565 565 + Effect.sync(() => new Date().toISOString()), 566 566 + (timestamp) => 567 567 + publishIoEvent({ 568 568 + runId: input.runId, 569 569 + source: "program", 570 570 + stream, 571 571 + line, 572 572 + timestamp, 573 573 + }), 574 574 + ), 546 575 }), 547 576 }), 548 577 (error) => ··· 589 618 ); 590 619 } 591 620 621 621 + const channel = input.channel ?? "events"; 622 622 + 623 623 + if (input.runId === undefined && channel !== "events") { 624 624 + return Promise.reject(new Error("watch --channel io|all requires --run <runId>.")); 625 625 + } 626 626 + 627 627 + if (input.runId === undefined && (input.source !== undefined || input.spawnId !== undefined)) { 628 628 + return Promise.reject(new Error("watch --source/--spawn requires --run <runId>.")); 629 629 + } 630 630 + 631 631 + if (channel === "io" && input.sinceTimeIso !== undefined) { 632 632 + return Promise.reject(new Error("watch --channel io does not support --since-time.")); 633 633 + } 634 634 + 635 635 + if (channel === "events" && (input.source !== undefined || input.spawnId !== undefined)) { 636 636 + return Promise.reject( 637 637 + new Error("watch --source/--spawn require --channel io or --channel all."), 638 638 + ); 639 639 + } 640 640 + 592 641 const engineContext = await makeEngineForConfig(input); 593 642 594 643 if (input.runId === undefined) { 595 595 - if (input.raw === true) { 596 596 - return Promise.reject(new Error("watch --raw requires a runId.")); 597 597 - } 598 598 - 599 644 await runWithBunContext( 600 645 Effect.scoped( 601 646 Stream.runForEach(engineContext.engine.watchAll(input.sinceTimeIso), (event) => 602 602 - Effect.sync(() => { 603 603 - input.onEvent(JSON.stringify(event)); 604 604 - }), 647 647 + emitWatchOutput(input.onEvent, toWatchEventOutput(event)), 605 648 ), 606 649 ), 607 650 ); ··· 611 654 612 655 const runId = decodeRunIdSync(input.runId); 613 656 614 614 - if (input.raw === true) { 657 657 + const eventStream = Stream.filter(engineContext.engine.watch(runId), (event) => 658 658 + input.sinceTimeIso === undefined ? true : event.timestamp >= input.sinceTimeIso, 659 659 + ); 660 660 + 661 661 + const ioStream = Stream.filter(engineContext.engine.watchIo(runId), (event) => 662 662 + filterIoEvent(event, input.source, input.spawnId), 663 663 + ); 664 664 + 665 665 + if (channel === "events") { 615 666 await runWithBunContext( 616 616 - Effect.raceFirst( 617 617 - Effect.scoped( 618 618 - Stream.runForEach(engineContext.engine.watchRaw(runId), (line) => 619 619 - Effect.sync(() => { 620 620 - input.onEvent(line); 621 621 - }), 622 622 - ), 667 667 + Effect.scoped( 668 668 + Stream.runForEach( 669 669 + Stream.takeUntil(eventStream, (event) => isRunTerminalEvent(event.type)), 670 670 + (event) => emitWatchOutput(input.onEvent, toWatchEventOutput(event)), 623 671 ), 624 624 - engineContext.engine.wait(runId, DEFAULT_SYNC_WAIT_TIMEOUT_SECONDS * 1000), 625 672 ), 626 673 ); 627 674 628 675 return; 629 676 } 630 677 631 631 - const watchStream = Stream.filter(engineContext.engine.watch(runId), (event) => 632 632 - input.sinceTimeIso === undefined ? true : event.timestamp >= input.sinceTimeIso, 633 633 - ); 678 678 + const currentRun = await runWithBunContext(engineContext.engine.status(runId)); 634 679 635 635 - await runWithBunContext( 636 636 - Effect.scoped( 637 637 - Stream.runForEach( 638 638 - Stream.takeUntil(watchStream, (event) => isRunTerminalEvent(event.type)), 639 639 - (event) => 640 640 - Effect.sync(() => { 641 641 - input.onEvent(JSON.stringify(event)); 642 642 - }), 643 643 - ), 644 644 - ), 645 645 - ); 646 646 - }; 680 680 + if ( 681 681 + currentRun.status === "complete" || 682 682 + currentRun.status === "failed" || 683 683 + currentRun.status === "cancelled" 684 684 + ) { 685 685 + if (channel === "all") { 686 686 + await runWithBunContext( 687 687 + Effect.scoped( 688 688 + Stream.runForEach( 689 689 + Stream.takeUntil(eventStream, (event) => isRunTerminalEvent(event.type)), 690 690 + (event) => emitWatchOutput(input.onEvent, toWatchEventOutput(event)), 691 691 + ), 692 692 + ), 693 693 + ); 694 694 + } 647 695 648 648 - export const inspectRun = async ( 649 649 - input: InspectRunInput, 650 650 - ): Promise<InspectResult | InspectSessionOutput> => { 651 651 - const parsedRef = parseInspectRef(input.ref); 652 652 - 653 653 - if (parsedRef === undefined) { 654 654 - return Promise.reject(new Error("inspect reference requires a runId")); 696 696 + return; 655 697 } 656 698 657 657 - const engineContext = await makeEngineForConfig(input); 658 658 - const inspected = await runWithBunContext( 659 659 - engineContext.engine.inspect({ 660 660 - runId: decodeRunIdSync(parsedRef.runId), 661 661 - spawnId: parsedRef.spawnId === undefined ? undefined : decodeSpawnIdSync(parsedRef.spawnId), 662 662 - }), 663 663 - ); 699 699 + if (channel === "io") { 700 700 + await runWithBunContext( 701 701 + Effect.raceFirst( 702 702 + Effect.scoped( 703 703 + Stream.runForEach(ioStream, (event) => 704 704 + emitWatchOutput(input.onEvent, toWatchIoOutput(event)), 705 705 + ), 706 706 + ), 707 707 + engineContext.engine.wait(runId, DEFAULT_SYNC_WAIT_TIMEOUT_SECONDS * 1000), 708 708 + ), 709 709 + ); 664 710 665 665 - if (input.session !== true) { 666 666 - return inspected; 711 711 + return; 667 712 } 668 713 669 669 - if (inspected.kind !== "spawn" || inspected.result === undefined) { 670 670 - return Promise.reject( 671 671 - new Error("inspect --session requires a runId.spawnId reference with completed spawn result"), 672 672 - ); 673 673 - } 674 674 - 675 675 - const resolvedConfig = await resolveConfig(input); 676 676 - const driverRegistry = makeDriverRegistry({ 677 677 - defaultDriver: resolvedConfig.config.defaultDriver, 678 678 - drivers: resolvedConfig.config.drivers, 679 679 - }); 680 680 - const run = await runWithBunContext( 681 681 - engineContext.engine.status(decodeRunIdSync(parsedRef.runId)), 682 682 - ); 683 683 - const resolvedDriver = await Runtime.runPromise(runtime)(driverRegistry.resolve(run.driver)); 714 714 + await runWithBunContext( 715 715 + Effect.scoped( 716 716 + Effect.gen(function* () { 717 717 + const ioFiber = yield* Effect.forkScoped( 718 718 + Stream.runForEach(ioStream, (event) => 719 719 + emitWatchOutput(input.onEvent, toWatchIoOutput(event)), 720 720 + ), 721 721 + ); 684 722 685 685 - if (resolvedDriver.runtime.resolveSession === undefined) { 686 686 - return Promise.reject( 687 687 - new Error(`Driver ${resolvedDriver.name} does not support session inspection`), 688 688 - ); 689 689 - } 723 723 + yield* Stream.runForEach( 724 724 + Stream.takeUntil(eventStream, (event) => isRunTerminalEvent(event.type)), 725 725 + (event) => emitWatchOutput(input.onEvent, toWatchEventOutput(event)), 726 726 + ); 690 727 691 691 - const sessionPointer = await Runtime.runPromise(runtime)( 692 692 - Effect.provide( 693 693 - resolvedDriver.runtime.resolveSession({ 694 694 - sessionRef: inspected.result.sessionRef, 728 728 + yield* Fiber.interrupt(ioFiber); 695 729 }), 696 696 - BunContext.layer, 697 730 ), 698 731 ); 699 699 - 700 700 - return { 701 701 - runId: parsedRef.runId, 702 702 - spawnId: inspected.spawnId, 703 703 - ...sessionPointer, 704 704 - } satisfies InspectSessionOutput; 705 732 }; 706 733 707 734 export const cancelRun = async (

+20 -1

packages/core/src/runtime/program-host.effect.ts

reviewed

··· 24 24 readonly executorName: string; 25 25 readonly extensions: ReadonlyArray<ExtensionRegistration>; 26 26 readonly spawn: (input: SpawnOptions) => Effect.Effect<SpawnResult, unknown>; 27 27 + readonly onIo?: (input: { 28 28 + readonly stream: "stdout" | "stderr"; 29 29 + readonly line: string; 30 30 + }) => Effect.Effect<void>; 27 31 } 28 32 29 33 type ProgramHostResultMessage = Extract<ProgramHostInboundMessage, { readonly kind: "result" }>; ··· 322 326 Stream.runForEach(Stream.splitLines(Stream.decodeText(processHandle.stdout)), (line) => 323 327 Effect.gen(function* () { 324 328 if (!line.startsWith(ProgramHostProtocolPrefix)) { 329 329 + if (line.length > 0 && input.onIo !== undefined) { 330 330 + yield* input.onIo({ 331 331 + stream: "stdout", 332 332 + line, 333 333 + }); 334 334 + } 325 335 return; 326 336 } 327 337 ··· 408 418 409 419 const stderrFiber = yield* Effect.forkScoped( 410 420 Stream.runForEach(Stream.splitLines(Stream.decodeText(processHandle.stderr)), (line) => 411 411 - Ref.update(stderrLinesRef, (lines) => [...lines, line]), 421 421 + Effect.gen(function* () { 422 422 + yield* Ref.update(stderrLinesRef, (lines) => [...lines, line]); 423 423 + 424 424 + if (line.length > 0 && input.onIo !== undefined) { 425 425 + yield* input.onIo({ 426 426 + stream: "stderr", 427 427 + line, 428 428 + }); 429 429 + } 430 430 + }), 412 431 ), 413 432 ); 414 433

+1 -1

packages/pi-mill/README.md

reviewed

··· 62 62 ]); 63 63 ``` 64 64 65 65 - Each `mill.spawn()` submits an async mill run (`mill run --json`) and then follows completion via mill APIs (`wait` + `inspect`). Model selection, driver routing, and execution behavior all come from your mill configuration. 65 65 + Each `mill.spawn()` submits an async mill run (`mill run --json`) and then follows completion via mill APIs (`wait` + `watch --channel events`). Model selection, driver routing, and execution behavior all come from your mill configuration. 66 66 67 67 By default, mill run storage uses mill's global default (`~/.mill/runs`) unless you explicitly pass `--runs-dir` (or set `millRunsDir`). 68 68

+1 -1

packages/pi-mill/index.ts

reviewed

··· 360 360 label: "Subagent", 361 361 description: [ 362 362 "Spawn subagents for delegated or orchestrated work.", 363 363 - "Execution backend: mill async APIs (submit + watch + inspect). Configure drivers/executors/models via mill.config.ts.", 363 363 + "Execution backend: mill async APIs (submit + watch + wait). Configure drivers/executors/models via mill.config.ts.", 364 364 `Enabled models: ${modelsText}`, 365 365 "Write a TypeScript script. `mill` is a global (like `process` or `console`). Use mill.spawn() to orchestrate agents.", 366 366 "mill.spawn() returns a Promise<ExecutionResult>. Use `await` for sequential, `Promise.all` for parallel.",

+99 -39

packages/pi-mill/runtime.ts

reviewed

··· 136 136 runId?: string; 137 137 } 138 138 139 139 - interface MillRunInspectPayload { 140 140 - run?: { 141 141 - status?: string; 142 142 - }; 143 143 - result?: { 144 144 - status?: string; 145 145 - errorMessage?: string; 146 146 - spawns?: ReadonlyArray<MillSpawnResult>; 139 139 + interface MillWatchOutputEnvelope { 140 140 + kind?: string; 141 141 + runId?: string; 142 142 + event?: { 143 143 + type?: string; 144 144 + payload?: Record<string, unknown>; 147 145 }; 148 146 } 149 147 ··· 186 184 } 187 185 188 186 return undefined; 187 187 + }; 188 188 + 189 189 + const parseJsonObjectsFromText = (text: string): Array<Record<string, unknown>> => 190 190 + text 191 191 + .split("\n") 192 192 + .map((line) => line.trim()) 193 193 + .filter((line) => line.length > 0) 194 194 + .flatMap((line) => { 195 195 + try { 196 196 + const parsed = JSON.parse(line) as unknown; 197 197 + return typeof parsed === "object" && parsed !== null 198 198 + ? [parsed as Record<string, unknown>] 199 199 + : []; 200 200 + } catch { 201 201 + return []; 202 202 + } 203 203 + }); 204 204 + 205 205 + const isRecord = (value: unknown): value is Record<string, unknown> => 206 206 + typeof value === "object" && value !== null; 207 207 + 208 208 + const readStringField = (record: Record<string, unknown>, key: string): string | undefined => { 209 209 + const value = record[key]; 210 210 + return typeof value === "string" ? value : undefined; 189 211 }; 190 212 191 213 const formatCommand = (command: string, args: ReadonlyArray<string>): string => ··· 328 350 } 329 351 330 352 const decodeMillResult = ( 331 331 - payload: MillRunInspectPayload, 353 353 + payloads: ReadonlyArray<Record<string, unknown>>, 332 354 fallback: { agent: string; modelId: string; prompt: string }, 333 355 ): ExecutionResult => { 334 334 - const spawns = payload.result?.spawns; 335 335 - if (!Array.isArray(spawns) || spawns.length === 0) { 336 336 - const failedStatus = payload.result?.status ?? payload.run?.status; 337 337 - const message = payload.result?.errorMessage; 356 356 + const spawnResults: Array<MillSpawnResult> = []; 357 357 + let runFailedMessage: string | undefined; 358 358 + 359 359 + for (const payload of payloads) { 360 360 + const envelope = payload as MillWatchOutputEnvelope; 361 361 + if (envelope.kind !== "event") { 362 362 + continue; 363 363 + } 364 364 + 365 365 + const event = envelope.event; 366 366 + if (!isRecord(event)) { 367 367 + continue; 368 368 + } 369 369 + 370 370 + const eventType = readStringField(event, "type"); 371 371 + const eventPayload = event.payload; 372 372 + 373 373 + if (eventType === "spawn:complete" && isRecord(eventPayload) && isRecord(eventPayload.result)) { 374 374 + const result = eventPayload.result; 375 375 + spawnResults.push({ 376 376 + text: readStringField(result, "text"), 377 377 + sessionRef: readStringField(result, "sessionRef"), 378 378 + agent: readStringField(result, "agent"), 379 379 + model: readStringField(result, "model"), 380 380 + driver: readStringField(result, "driver"), 381 381 + exitCode: typeof result.exitCode === "number" ? result.exitCode : undefined, 382 382 + stopReason: readStringField(result, "stopReason"), 383 383 + errorMessage: readStringField(result, "errorMessage"), 384 384 + }); 385 385 + continue; 386 386 + } 387 387 + 388 388 + if (eventType === "run:failed" && isRecord(eventPayload)) { 389 389 + runFailedMessage = readStringField(eventPayload, "message"); 390 390 + } 391 391 + } 392 392 + 393 393 + if (spawnResults.length === 0) { 338 394 throw new MillError({ 339 395 code: "RUNTIME", 340 396 message: 341 341 - message && message.length > 0 342 342 - ? `mill run ${failedStatus ?? "unknown"}: ${message}` 397 397 + runFailedMessage && runFailedMessage.length > 0 398 398 + ? `mill run failed: ${runFailedMessage}` 343 399 : "mill run completed without spawn results.", 344 400 recoverable: false, 345 401 }); 346 402 } 347 403 348 404 const selectedSpawn = 349 349 - spawns.find((spawn) => spawn.agent === fallback.agent) ?? spawns[0] ?? ({} as MillSpawnResult); 405 405 + spawnResults.find((spawn) => spawn.agent === fallback.agent) ?? 406 406 + spawnResults[0] ?? 407 407 + ({} as MillSpawnResult); 350 408 351 351 - const runStatus = payload.result?.status ?? payload.run?.status; 352 352 - const derivedExitCode = 353 353 - typeof selectedSpawn.exitCode === "number" 354 354 - ? selectedSpawn.exitCode 355 355 - : runStatus === "complete" 356 356 - ? 0 357 357 - : 1; 409 409 + const derivedExitCode = typeof selectedSpawn.exitCode === "number" ? selectedSpawn.exitCode : 0; 358 410 359 359 - const errorMessage = selectedSpawn.errorMessage ?? payload.result?.errorMessage; 411 411 + const errorMessage = selectedSpawn.errorMessage; 360 412 const stopReason = selectedSpawn.stopReason; 361 413 362 414 if (derivedExitCode !== 0 || stopReason === "error" || (errorMessage?.length ?? 0) > 0) { ··· 621 673 if (terminalStatus === "failed") { 622 674 throw new MillError({ 623 675 code: "RUNTIME", 624 624 - message: "Subagent run failed before inspect.", 676 676 + message: "Subagent run failed before watch replay.", 625 677 recoverable: false, 626 678 }); 627 679 } 628 680 629 629 - const inspectArgs = [...input.millArgs, "inspect", submittedRunId, "--json"]; 681 681 + const watchArgs = [ 682 682 + ...input.millArgs, 683 683 + "watch", 684 684 + "--run", 685 685 + submittedRunId, 686 686 + "--channel", 687 687 + "events", 688 688 + "--json", 689 689 + ]; 630 690 if (input.millRunsDir && input.millRunsDir.trim().length > 0) { 631 631 - inspectArgs.push("--runs-dir", input.millRunsDir); 691 691 + watchArgs.push("--runs-dir", input.millRunsDir); 632 692 } 633 693 634 634 - const inspected = await runCommandCapture({ 694 694 + const watched = await runCommandCapture({ 635 695 command: input.millCommand, 636 636 - args: inspectArgs, 696 696 + args: watchArgs, 637 697 cwd: input.cwd, 638 698 env: process.env, 639 699 signal: input.signal, 640 700 }); 641 641 - appendCommandLog(stdoutPath, input.millCommand, inspectArgs, inspected); 701 701 + appendCommandLog(stdoutPath, input.millCommand, watchArgs, watched); 642 702 643 643 - if (inspected.aborted || aborted) { 703 703 + if (watched.aborted || aborted) { 644 704 await requestRunCancel(); 645 705 throw new MillError({ 646 706 code: "CANCELLED", ··· 649 709 }); 650 710 } 651 711 652 652 - if (inspected.code !== 0) { 712 712 + if (watched.code !== 0) { 653 713 throw new MillError({ 654 714 code: "RUNTIME", 655 715 message: 656 656 - inspected.combined.trim().length > 0 657 657 - ? `mill inspect failed:\n${inspected.combined.trim()}` 658 658 - : "mill inspect failed.", 716 716 + watched.combined.trim().length > 0 717 717 + ? `mill watch failed:\n${watched.combined.trim()}` 718 718 + : "mill watch failed.", 659 719 recoverable: false, 660 720 }); 661 721 } 662 722 663 663 - const parsed = parseJsonObjectFromText([inspected.stdout, inspected.stderr].join("\n")); 664 664 - if (!parsed) { 723 723 + const parsedEvents = parseJsonObjectsFromText([watched.stdout, watched.stderr].join("\n")); 724 724 + if (parsedEvents.length === 0) { 665 725 throw new MillError({ 666 726 code: "RUNTIME", 667 667 - message: "mill inspect output was not valid JSON.", 727 727 + message: "mill watch output did not contain JSON events.", 668 728 recoverable: false, 669 729 }); 670 730 } 671 731 672 672 - const decoded = decodeMillResult(parsed as MillRunInspectPayload, { 732 732 + const decoded = decodeMillResult(parsedEvents, { 673 733 agent: input.agent, 674 734 modelId: input.modelId, 675 735 prompt: input.prompt,