From cdf862fdeec53451aee9d49a9be0c9d2b4017b2c Mon Sep 17 00:00:00 2001 From: joshwheelock Date: Thu, 9 Jul 2026 23:34:05 +0200 Subject: [PATCH] fix(tui): harden daemon worker runtime --- CHANGELOG.md | 4 + docs/guides/advanced-workflows.md | 6 + src/application/agents/AgentInvocation.ts | 6 + .../goals/codify/CodifierProcessManager.ts | 93 ++--- .../goals/refine/RefinerProcessManager.ts | 98 +++-- .../goals/review/ReviewerProcessManager.ts | 126 ++++--- src/application/daemons/IProcessManager.ts | 3 + .../daemons/ProcessManagerEventEmitter.ts | 66 ++++ src/infrastructure/agents/AgentCliGateway.ts | 55 ++- .../NodeWorkerDaemonProcessController.ts | 5 +- .../cockpit/CockpitDaemonAgentConfigCycler.ts | 2 +- .../tui/cockpit/CockpitDaemonPanel.tsx | 64 +++- .../cockpit/DaemonEventRowMessageFormatter.ts | 12 + .../tui/cockpit/DaemonEventRows.ts | 48 ++- .../DaemonEventCategory.ts | 2 + .../DaemonEventSnapshot.ts | 3 + .../DaemonOutputEventParser.ts | 2 + .../daemon-subprocesses/ManagedSubprocess.ts | 2 + .../SubprocessLifecycleEventRecorder.ts | 59 ++- .../daemon-subprocesses/SubprocessManager.ts | 77 +++- .../codify/CodifierProcessManager.test.ts | 20 +- .../refine/RefinerProcessManager.test.ts | 38 +- .../review/ReviewerProcessManager.test.ts | 51 ++- .../agents/AgentCliGateway.test.ts | 84 +++++ .../NodeWorkerDaemonProcessController.test.ts | 44 ++- .../tui/application-shell/App.test.tsx | 177 +++++++++ .../ApplicationLauncher.test.tsx | 2 +- .../CockpitDaemonAgentConfigCycler.test.ts | 10 + .../tui/cockpit/CockpitDaemonPanel.test.tsx | 43 +++ .../tui/cockpit/DaemonEventRows.test.ts | 22 +- .../SubprocessManager.test.ts | 84 ++++- ...rker-daemon.production.integration.test.ts | 335 ++++++++++++++++++ 32 files changed, 1429 insertions(+), 214 deletions(-) create mode 100644 src/application/daemons/ProcessManagerEventEmitter.ts create mode 100644 tests/presentation/work/worker-daemon.production.integration.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 85985f04..77c2e1b5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- **TUI daemon reliability**: Cockpit refiner, reviewer, and codifier workers now stream bounded activity and errors while running, retain meaningful outcomes and elapsed-time heartbeats, survive navigation between TUI screens, and stop exactly once on request or application shutdown. + ## [3.16.0] - 2026-07-09 ### Changed diff --git a/docs/guides/advanced-workflows.md b/docs/guides/advanced-workflows.md index 9844b580..a3f34fa3 100644 --- a/docs/guides/advanced-workflows.md +++ b/docs/guides/advanced-workflows.md @@ -65,6 +65,12 @@ Tune the daemon for your workflow: See the [work refine reference](/reference/commands/work/#jumbo-work-refine) for full option details. +## Run unattended workflows from Cockpit + +The TUI Cockpit can run the refiner, reviewer, and codifier as independent background workers. Use **Tab** to focus a daemon panel and **S** to start or stop it. Open the panel configuration with **@** to cycle the agent, poll interval, and retry count before starting. + +Running workers continue polling while you navigate to Goals, Memory, or Settings. Returning to Cockpit shows each worker's process status, active goal, attempt, current phase, latest meaningful activity, elapsed time, and recent events. Completion and failure outcomes remain visible across idle polls, and exiting the TUI shuts down every worker through the application lifecycle. + ## Run multiple agents in parallel safely Jumbo supports concurrent workers in separate terminals. diff --git a/src/application/agents/AgentInvocation.ts b/src/application/agents/AgentInvocation.ts index 0cbf3b3a..bba5e243 100644 --- a/src/application/agents/AgentInvocation.ts +++ b/src/application/agents/AgentInvocation.ts @@ -1,6 +1,12 @@ export interface AgentInvocation { readonly agentId: string; readonly prompt: string; + readonly onActivity?: (activity: AgentActivity) => void; +} + +export interface AgentActivity { + readonly stream: "stdout" | "stderr"; + readonly text: string; } export interface AgentInvocationResult { diff --git a/src/application/context/goals/codify/CodifierProcessManager.ts b/src/application/context/goals/codify/CodifierProcessManager.ts index 25c92404..f92b9e2f 100644 --- a/src/application/context/goals/codify/CodifierProcessManager.ts +++ b/src/application/context/goals/codify/CodifierProcessManager.ts @@ -5,6 +5,7 @@ import type { ProcessManagerResult, ProcessManagerStatus, } from "../../../daemons/IProcessManager.js"; +import { ProcessManagerEventEmitter } from "../../../daemons/ProcessManagerEventEmitter.js"; import { IAgentGateway } from "../../../agents/IAgentGateway.js"; import { ITelemetryClient } from "../../../telemetry/ITelemetryClient.js"; import { IWorkerIdentityReader } from "../../../host/workers/IWorkerIdentityReader.js"; @@ -68,25 +69,26 @@ export class CodifierProcessManager implements IProcessManager { async processNext(options: CodifierProcessOptions): Promise { const startedAt = process.hrtime.bigint(); + const events = new ProcessManagerEventEmitter(CODIFIER_EVENT_SOURCE, options, startedAt); + events.emit("processing", "polling", "polling for approved goals", { phase: "polling" }); const goals = await this.selectEligibleGoals(); if (goals.length === 0) { - this.emit(options, { - daemon: CODIFIER_EVENT_SOURCE, - status: "idle", - source: CODIFIER_EVENT_SOURCE, - ...CODIFIER_EVENT_COPY.noWork, - }); + events.emit("idle", CODIFIER_EVENT_COPY.noWork.category, CODIFIER_EVENT_COPY.noWork.message, { phase: "idle" }); this.track("codifier_process_completed", startedAt, { status: "idle", attempts: 0 }); return { status: "idle", attempts: 0 }; } const goal = goals[0]; + events.emit("processing", "selection", "selected goal for codification", { + phase: "selection", + goalId: goal.goalId, + }); try { await this.codifyGoalController.handle({ goalId: goal.goalId }); } catch (error) { - this.emitFailure(options, goal.goalId, error); + this.emitFailure(events, goal.goalId, error); this.track("codifier_process_completed", startedAt, { status: "failed", attempts: 0, @@ -97,31 +99,35 @@ export class CodifierProcessManager implements IProcessManager { } for (let attempt = 1; attempt <= options.maxRetries; attempt++) { - this.emit(options, { - daemon: CODIFIER_EVENT_SOURCE, - status: "processing", - source: CODIFIER_EVENT_SOURCE, + const attemptContext = { goalId: goal.goalId, attempt, maxRetries: options.maxRetries, - ...CODIFIER_EVENT_COPY.workStarted, + }; + events.emit("processing", CODIFIER_EVENT_COPY.workStarted.category, CODIFIER_EVENT_COPY.workStarted.message, { + phase: "working", + ...attemptContext, }); - const result = await this.agentGateway.invoke({ + let sawAgentActivity = false; + const invocation = { agentId: options.agentId, prompt: this.buildPrompt(goal.goalId), - }); + onActivity: (activity: { readonly stream: "stdout" | "stderr"; readonly text: string }) => { + sawAgentActivity = true; + events.emitAgentActivity(activity.stream, activity.text, attemptContext); + }, + }; + const result = await this.agentGateway.invoke(invocation); + if (!sawAgentActivity) { + this.emitReturnedAgentOutput(events, result, attemptContext); + } if (await this.isGoalDone(goal.goalId)) { - this.emit(options, { - daemon: CODIFIER_EVENT_SOURCE, - status: "completed", - source: CODIFIER_EVENT_SOURCE, - goalId: goal.goalId, - attempt, - maxRetries: options.maxRetries, + events.emit("completed", CODIFIER_EVENT_COPY.completed.category, CODIFIER_EVENT_COPY.completed.message, { + phase: "completed", + ...attemptContext, exitCode: result.exitCode, - ...CODIFIER_EVENT_COPY.completed, }); this.track("codifier_process_completed", startedAt, { status: "completed", @@ -133,16 +139,17 @@ export class CodifierProcessManager implements IProcessManager { } const retryExhausted = attempt === options.maxRetries; - this.emit(options, { - daemon: CODIFIER_EVENT_SOURCE, - status: retryExhausted ? "exhausted" : "skipped", - source: CODIFIER_EVENT_SOURCE, - goalId: goal.goalId, - attempt, - maxRetries: options.maxRetries, + events.emit(retryExhausted ? "exhausted" : "skipped", retryExhausted ? CODIFIER_EVENT_COPY.exhausted.category : CODIFIER_EVENT_COPY.skipped.category, retryExhausted ? CODIFIER_EVENT_COPY.exhausted.message : CODIFIER_EVENT_COPY.skipped.message, { + phase: retryExhausted ? "exhausted" : "retry", + ...attemptContext, exitCode: result.exitCode, - ...(retryExhausted ? CODIFIER_EVENT_COPY.exhausted : CODIFIER_EVENT_COPY.skipped), }); + if (!retryExhausted) { + events.emit("processing", "retry", "retrying codification", { + phase: "retry", + ...attemptContext, + }); + } } this.track("codifier_process_completed", startedAt, { @@ -176,21 +183,14 @@ export class CodifierProcessManager implements IProcessManager { return goal?.status === GoalStatus.DONE; } - private emit(options: CodifierProcessOptions, event: CodifierProcessEvent): void { - options.emit?.(event); - } - private emitFailure( - options: CodifierProcessOptions, + events: ProcessManagerEventEmitter, goalId: string, error: unknown, ): void { - this.emit(options, { - daemon: CODIFIER_EVENT_SOURCE, - status: "failed", - source: CODIFIER_EVENT_SOURCE, + events.emit("failed", CODIFIER_EVENT_COPY.failed.category, CODIFIER_EVENT_COPY.failed.message, { + phase: "failed", goalId, - ...CODIFIER_EVENT_COPY.failed, ...this.errorProperties(error), }); } @@ -207,6 +207,19 @@ export class CodifierProcessManager implements IProcessManager { }); } + private emitReturnedAgentOutput( + events: ProcessManagerEventEmitter, + result: { readonly stdout?: string; readonly stderr?: string }, + context: { readonly goalId: string; readonly attempt: number; readonly maxRetries: number }, + ): void { + if (result.stdout !== undefined) { + events.emitAgentActivity("stdout", result.stdout, context); + } + if (result.stderr !== undefined) { + events.emitAgentActivity("stderr", result.stderr, context); + } + } + private errorProperties(error: unknown): { errorType: string; errorMessage: string; diff --git a/src/application/context/goals/refine/RefinerProcessManager.ts b/src/application/context/goals/refine/RefinerProcessManager.ts index b3c00cdb..125a2816 100644 --- a/src/application/context/goals/refine/RefinerProcessManager.ts +++ b/src/application/context/goals/refine/RefinerProcessManager.ts @@ -1,5 +1,6 @@ import { IAgentGateway } from "../../../agents/IAgentGateway.js"; -import { IProcessManager, ProcessManagerEvent, ProcessManagerOptions, ProcessManagerResult } from "../../../daemons/IProcessManager.js"; +import { IProcessManager, ProcessManagerOptions, ProcessManagerResult } from "../../../daemons/IProcessManager.js"; +import { ProcessManagerEventEmitter } from "../../../daemons/ProcessManagerEventEmitter.js"; import { IWorkerIdentityReader } from "../../../host/workers/IWorkerIdentityReader.js"; import { ITelemetryClient } from "../../../telemetry/ITelemetryClient.js"; import { GoalStatus } from "../../../../domain/goals/Constants.js"; @@ -51,31 +52,29 @@ export class RefinerProcessManager implements IProcessManager { async processNext(options: ProcessManagerOptions): Promise { const startedAt = process.hrtime.bigint(); + const events = new ProcessManagerEventEmitter(REFINER_EVENT_SOURCE, options, startedAt); + events.emit("processing", "polling", "polling for defined goals", { phase: "polling" }); const goals = await this.selectEligibleGoals(); if (goals.length === 0) { - this.emit(options, { - daemon: REFINER_EVENT_SOURCE, - status: "idle", - source: REFINER_EVENT_SOURCE, - ...REFINER_EVENT_COPY.noWork, - }); + events.emit("idle", REFINER_EVENT_COPY.noWork.category, REFINER_EVENT_COPY.noWork.message, { phase: "idle" }); this.track(startedAt, { status: "idle", attempts: 0 }); return { status: "idle", attempts: 0 }; } const goal = goals[0]; + events.emit("processing", "selection", "selected goal for refinement", { + phase: "selection", + goalId: goal.goalId, + }); try { await this.refineGoalController.handle({ goalId: goal.goalId }); } catch (error) { const errorProperties = this.errorProperties(error); - this.emit(options, { - daemon: REFINER_EVENT_SOURCE, - status: "failed", - source: REFINER_EVENT_SOURCE, + events.emit("failed", REFINER_EVENT_COPY.failed.category, REFINER_EVENT_COPY.failed.message, { + phase: "failed", goalId: goal.goalId, - ...REFINER_EVENT_COPY.failed, ...errorProperties, }); this.track(startedAt, { status: "failed", attempts: 0, goalId: goal.goalId, ...errorProperties }); @@ -83,44 +82,52 @@ export class RefinerProcessManager implements IProcessManager { } for (let attempt = 1; attempt <= options.maxRetries; attempt++) { - this.emit(options, { - daemon: REFINER_EVENT_SOURCE, - status: "processing", - source: REFINER_EVENT_SOURCE, + const attemptContext = { goalId: goal.goalId, attempt, maxRetries: options.maxRetries, - ...REFINER_EVENT_COPY.workStarted, + }; + events.emit("processing", REFINER_EVENT_COPY.workStarted.category, REFINER_EVENT_COPY.workStarted.message, { + phase: "working", + ...attemptContext, }); - const result = await this.agentGateway.invoke({ agentId: options.agentId, prompt: this.buildPrompt(goal.goalId) }); + let sawAgentActivity = false; + const invocation = { + agentId: options.agentId, + prompt: this.buildPrompt(goal.goalId), + onActivity: (activity: { readonly stream: "stdout" | "stderr"; readonly text: string }) => { + sawAgentActivity = true; + events.emitAgentActivity(activity.stream, activity.text, attemptContext); + }, + }; + const result = await this.agentGateway.invoke(invocation); + if (!sawAgentActivity) { + this.emitReturnedAgentOutput(events, result, attemptContext); + } if (await this.isGoalRefined(goal.goalId)) { - this.emit(options, { - daemon: REFINER_EVENT_SOURCE, - status: "completed", - source: REFINER_EVENT_SOURCE, - goalId: goal.goalId, - attempt, - maxRetries: options.maxRetries, + events.emit("completed", REFINER_EVENT_COPY.completed.category, REFINER_EVENT_COPY.completed.message, { + phase: "completed", + ...attemptContext, exitCode: result.exitCode, - ...REFINER_EVENT_COPY.completed, }); this.track(startedAt, { status: "completed", attempts: attempt, goalId: goal.goalId, agentExitCode: result.exitCode }); return { status: "completed", goalId: goal.goalId, attempts: attempt }; } const retryExhausted = attempt === options.maxRetries; - this.emit(options, { - daemon: REFINER_EVENT_SOURCE, - status: retryExhausted ? "exhausted" : "skipped", - source: REFINER_EVENT_SOURCE, - goalId: goal.goalId, - attempt, - maxRetries: options.maxRetries, + events.emit(retryExhausted ? "exhausted" : "skipped", retryExhausted ? REFINER_EVENT_COPY.exhausted.category : REFINER_EVENT_COPY.skipped.category, retryExhausted ? REFINER_EVENT_COPY.exhausted.message : REFINER_EVENT_COPY.skipped.message, { + phase: retryExhausted ? "exhausted" : "retry", + ...attemptContext, exitCode: result.exitCode, - ...(retryExhausted ? REFINER_EVENT_COPY.exhausted : REFINER_EVENT_COPY.skipped), ...this.agentFailureProperties(result), }); + if (!retryExhausted) { + events.emit("processing", "retry", "retrying refinement", { + phase: "retry", + ...attemptContext, + }); + } } this.track(startedAt, { status: "exhausted", attempts: options.maxRetries, goalId: goal.goalId }); @@ -136,7 +143,11 @@ export class RefinerProcessManager implements IProcessManager { } buildPrompt(goalId: string): string { - return `Run the Jumbo refinement workflow for goal ${goalId}. Execute: jumbo goal refine --id ${goalId}`; + return [ + `Continue the Jumbo refinement workflow for goal ${goalId}.`, + "The daemon has already moved the goal into refinement.", + `Update the goal context as needed, then complete refinement with: jumbo goal commit --id ${goalId}`, + ].join(" "); } private async isGoalRefined(goalId: string): Promise { @@ -144,10 +155,6 @@ export class RefinerProcessManager implements IProcessManager { return goal?.status === GoalStatus.REFINED; } - private emit(options: ProcessManagerOptions, event: ProcessManagerEvent): void { - options.emit?.(event); - } - private track(startedAt: bigint, properties: Record): void { this.telemetryClient.track("refiner_process_completed", { daemon: "refiner", @@ -156,6 +163,19 @@ export class RefinerProcessManager implements IProcessManager { }); } + private emitReturnedAgentOutput( + events: ProcessManagerEventEmitter, + result: { readonly stdout?: string; readonly stderr?: string }, + context: { readonly goalId: string; readonly attempt: number; readonly maxRetries: number }, + ): void { + if (result.stdout !== undefined) { + events.emitAgentActivity("stdout", result.stdout, context); + } + if (result.stderr !== undefined) { + events.emitAgentActivity("stderr", result.stderr, context); + } + } + private errorProperties(error: unknown): { errorType: string; errorMessage: string; errorStack?: string } { if (error instanceof Error) { return { diff --git a/src/application/context/goals/review/ReviewerProcessManager.ts b/src/application/context/goals/review/ReviewerProcessManager.ts index 3f593253..9fabe027 100644 --- a/src/application/context/goals/review/ReviewerProcessManager.ts +++ b/src/application/context/goals/review/ReviewerProcessManager.ts @@ -1,5 +1,6 @@ import { IAgentGateway } from "../../../agents/IAgentGateway.js"; -import { IProcessManager, ProcessManagerEvent, ProcessManagerOptions, ProcessManagerResult } from "../../../daemons/IProcessManager.js"; +import { IProcessManager, ProcessManagerOptions, ProcessManagerResult } from "../../../daemons/IProcessManager.js"; +import { ProcessManagerEventEmitter } from "../../../daemons/ProcessManagerEventEmitter.js"; import { IWorkerIdentityReader } from "../../../host/workers/IWorkerIdentityReader.js"; import { ITelemetryClient } from "../../../telemetry/ITelemetryClient.js"; import { GoalStatus } from "../../../../domain/goals/Constants.js"; @@ -52,31 +53,29 @@ export class ReviewerProcessManager implements IProcessManager { async processNext(options: ProcessManagerOptions): Promise { const startedAt = process.hrtime.bigint(); + const events = new ProcessManagerEventEmitter(REVIEWER_EVENT_SOURCE, options, startedAt); + events.emit("processing", "polling", "polling for submitted goals", { phase: "polling" }); const goals = await this.selectEligibleGoals(); if (goals.length === 0) { - this.emit(options, { - daemon: REVIEWER_EVENT_SOURCE, - status: "idle", - source: REVIEWER_EVENT_SOURCE, - ...REVIEWER_EVENT_COPY.noWork, - }); + events.emit("idle", REVIEWER_EVENT_COPY.noWork.category, REVIEWER_EVENT_COPY.noWork.message, { phase: "idle" }); this.track(startedAt, { status: "idle", attempts: 0 }); return { status: "idle", attempts: 0 }; } const goal = goals[0]; + events.emit("processing", "selection", "selected goal for review", { + phase: "selection", + goalId: goal.goalId, + }); try { await this.reviewGoalController.handle({ goalId: goal.goalId }); } catch (error) { const errorProperties = this.errorProperties(error); - this.emit(options, { - daemon: REVIEWER_EVENT_SOURCE, - status: "failed", - source: REVIEWER_EVENT_SOURCE, + events.emit("failed", REVIEWER_EVENT_COPY.failed.category, REVIEWER_EVENT_COPY.failed.message, { + phase: "failed", goalId: goal.goalId, - ...REVIEWER_EVENT_COPY.failed, ...errorProperties, }); this.track(startedAt, { status: "failed", attempts: 0, goalId: goal.goalId, ...errorProperties }); @@ -84,44 +83,51 @@ export class ReviewerProcessManager implements IProcessManager { } for (let attempt = 1; attempt <= options.maxRetries; attempt++) { - this.emit(options, { - daemon: REVIEWER_EVENT_SOURCE, - status: "processing", - source: REVIEWER_EVENT_SOURCE, + const attemptContext = { goalId: goal.goalId, attempt, maxRetries: options.maxRetries, - ...REVIEWER_EVENT_COPY.workStarted, + }; + events.emit("processing", REVIEWER_EVENT_COPY.workStarted.category, REVIEWER_EVENT_COPY.workStarted.message, { + phase: "working", + ...attemptContext, }); - const result = await this.agentGateway.invoke({ agentId: options.agentId, prompt: this.buildPrompt(goal.goalId) }); - this.emitModelOutput(options, goal.goalId, result.stdout); + let sawAgentActivity = false; + const invocation = { + agentId: options.agentId, + prompt: this.buildPrompt(goal.goalId), + onActivity: (activity: { readonly stream: "stdout" | "stderr"; readonly text: string }) => { + sawAgentActivity = true; + events.emitAgentActivity(activity.stream, activity.text, attemptContext); + }, + }; + const result = await this.agentGateway.invoke(invocation); + if (!sawAgentActivity) { + this.emitReturnedAgentOutput(events, result, attemptContext); + } if (await this.isReviewComplete(goal.goalId)) { - this.emit(options, { - daemon: REVIEWER_EVENT_SOURCE, - status: "completed", - source: REVIEWER_EVENT_SOURCE, - goalId: goal.goalId, - attempt, - maxRetries: options.maxRetries, + events.emit("completed", REVIEWER_EVENT_COPY.completed.category, REVIEWER_EVENT_COPY.completed.message, { + phase: "completed", + ...attemptContext, exitCode: result.exitCode, - ...REVIEWER_EVENT_COPY.completed, }); this.track(startedAt, { status: "completed", attempts: attempt, goalId: goal.goalId, agentExitCode: result.exitCode }); return { status: "completed", goalId: goal.goalId, attempts: attempt }; } const retryExhausted = attempt === options.maxRetries; - this.emit(options, { - daemon: REVIEWER_EVENT_SOURCE, - status: retryExhausted ? "exhausted" : "skipped", - source: REVIEWER_EVENT_SOURCE, - goalId: goal.goalId, - attempt, - maxRetries: options.maxRetries, + events.emit(retryExhausted ? "exhausted" : "skipped", retryExhausted ? REVIEWER_EVENT_COPY.exhausted.category : REVIEWER_EVENT_COPY.skipped.category, retryExhausted ? REVIEWER_EVENT_COPY.exhausted.message : REVIEWER_EVENT_COPY.skipped.message, { + phase: retryExhausted ? "exhausted" : "retry", + ...attemptContext, exitCode: result.exitCode, - ...(retryExhausted ? REVIEWER_EVENT_COPY.exhausted : REVIEWER_EVENT_COPY.skipped), }); + if (!retryExhausted) { + events.emit("processing", "retry", "retrying review", { + phase: "retry", + ...attemptContext, + }); + } } this.track(startedAt, { status: "exhausted", attempts: options.maxRetries, goalId: goal.goalId }); @@ -137,7 +143,12 @@ export class ReviewerProcessManager implements IProcessManager { } buildPrompt(goalId: string): string { - return `Run the Jumbo review workflow for goal ${goalId}. Execute: jumbo goal review --id ${goalId}`; + return [ + `Continue the Jumbo review workflow for goal ${goalId}.`, + "The daemon has already moved the goal into review.", + `If the implementation passes QA, run: jumbo goal approve --id ${goalId}`, + `If it fails QA, run: jumbo goal reject --id ${goalId} --review-issues "describe the issues"`, + ].join(" "); } private async isReviewComplete(goalId: string): Promise { @@ -145,36 +156,6 @@ export class ReviewerProcessManager implements IProcessManager { return REVIEW_COMPLETE_STATUSES.has(goal?.status ?? "unknown"); } - private emit(options: ProcessManagerOptions, event: ProcessManagerEvent): void { - options.emit?.(event); - } - - private emitModelOutput( - options: ProcessManagerOptions, - goalId: string, - stdout: string | undefined, - ): void { - if (stdout === undefined) { - return; - } - - const lines = stdout - .split(/\r?\n/) - .map((line) => line.trim()) - .filter((line) => line.length > 0); - - for (const line of lines) { - this.emit(options, { - daemon: REVIEWER_EVENT_SOURCE, - status: "processing", - source: REVIEWER_EVENT_SOURCE, - category: "model-output", - message: limitTextTail(line, REVIEWER_EVENT_TEXT_FIELD_MAX_LENGTH), - goalId, - }); - } - } - private track(startedAt: bigint, properties: Record): void { this.telemetryClient.track("reviewer_process_completed", { daemon: "reviewer", @@ -183,6 +164,19 @@ export class ReviewerProcessManager implements IProcessManager { }); } + private emitReturnedAgentOutput( + events: ProcessManagerEventEmitter, + result: { readonly stdout?: string; readonly stderr?: string }, + context: { readonly goalId: string; readonly attempt: number; readonly maxRetries: number }, + ): void { + if (result.stdout !== undefined) { + events.emitAgentActivity("stdout", result.stdout, context); + } + if (result.stderr !== undefined) { + events.emitAgentActivity("stderr", result.stderr, context); + } + } + private errorProperties(error: unknown): { errorType: string; errorMessage: string; errorStack?: string } { if (error instanceof Error) { return { diff --git a/src/application/daemons/IProcessManager.ts b/src/application/daemons/IProcessManager.ts index 0b4821e0..409248df 100644 --- a/src/application/daemons/IProcessManager.ts +++ b/src/application/daemons/IProcessManager.ts @@ -18,6 +18,9 @@ export interface ProcessManagerEvent { readonly source?: string; readonly category?: string; readonly message?: string; + readonly timestampMs?: number; + readonly phase?: string; + readonly elapsedMs?: number; readonly goalId?: string; readonly attempt?: number; readonly maxRetries?: number; diff --git a/src/application/daemons/ProcessManagerEventEmitter.ts b/src/application/daemons/ProcessManagerEventEmitter.ts new file mode 100644 index 00000000..5ec9d1f5 --- /dev/null +++ b/src/application/daemons/ProcessManagerEventEmitter.ts @@ -0,0 +1,66 @@ +import type { ProcessManagerEvent, ProcessManagerOptions, ProcessManagerStatus } from "./IProcessManager.js"; + +const EVENT_TEXT_FIELD_MAX_LENGTH = 2_048; + +export class ProcessManagerEventEmitter { + constructor( + private readonly daemon: string, + private readonly options: ProcessManagerOptions, + private readonly startedAt: bigint, + ) {} + + emit( + status: ProcessManagerStatus, + category: string, + message: string, + event: Omit = {}, + ): void { + this.options.emit?.({ + daemon: this.daemon, + status, + source: this.daemon, + category, + message, + timestampMs: Date.now(), + elapsedMs: this.elapsedMs(), + ...event, + }); + } + + emitAgentActivity( + stream: "stdout" | "stderr", + text: string, + context: Pick, + ): void { + for (const line of text.split(/\r?\n/)) { + const message = line.trim(); + if (message.length === 0) { + continue; + } + + this.options.emit?.({ + daemon: this.daemon, + status: "processing", + source: "agent", + category: stream === "stdout" ? "model-output" : "agent-stderr", + message: limitTextTail(message, EVENT_TEXT_FIELD_MAX_LENGTH), + timestampMs: Date.now(), + elapsedMs: this.elapsedMs(), + phase: "agent", + ...context, + }); + } + } + + private elapsedMs(): number { + return Number((process.hrtime.bigint() - this.startedAt) / BigInt(1_000_000)); + } +} + +export function limitProcessEventTextTail(value: string, maxLength = EVENT_TEXT_FIELD_MAX_LENGTH): string { + return limitTextTail(value, maxLength); +} + +function limitTextTail(value: string, maxLength: number): string { + return value.length > maxLength ? value.slice(-maxLength) : value; +} diff --git a/src/infrastructure/agents/AgentCliGateway.ts b/src/infrastructure/agents/AgentCliGateway.ts index 7df9172d..c202b613 100644 --- a/src/infrastructure/agents/AgentCliGateway.ts +++ b/src/infrastructure/agents/AgentCliGateway.ts @@ -3,13 +3,14 @@ import { IAgentGateway } from "../../application/agents/IAgentGateway.js"; import { AgentInvocation, AgentInvocationResult } from "../../application/agents/AgentInvocation.js"; import { ITelemetryClient } from "../../application/telemetry/ITelemetryClient.js"; -export const SUPPORTED_AGENT_IDS = ["claude", "antigravity", "copilot", "codex", "cursor", "vibe"] as const; +export const SUPPORTED_AGENT_IDS = ["claude", "antigravity", "copilot", "codex", "vibe"] as const; export type SupportedAgentId = typeof SUPPORTED_AGENT_IDS[number]; interface AgentCommand { readonly executable: string; readonly args?: readonly string[]; readonly promptFlag?: string; + readonly direct?: boolean; } const CODEX_NON_INTERACTIVE_EXEC_ARGS = ["exec", "--skip-git-repo-check"] as const; @@ -20,7 +21,6 @@ const AGENT_COMMANDS: Record = { antigravity: { executable: "agy", promptFlag: "-p" }, copilot: { executable: "gh copilot", promptFlag: "-p" }, codex: { executable: "codex", args: CODEX_NON_INTERACTIVE_EXEC_ARGS }, - cursor: { executable: "cursor", promptFlag: "-p" }, vibe: { executable: "vibe", promptFlag: "-p" }, }; @@ -30,22 +30,30 @@ export class AgentCliGateway implements IAgentGateway { async invoke(invocation: AgentInvocation): Promise { const startedAt = process.hrtime.bigint(); const command = this.resolveCommand(invocation.agentId); - const commandLine = this.buildCommandLine(command, invocation.prompt); return new Promise((resolve) => { const stdout = new BoundedTextTail(AGENT_OUTPUT_TAIL_MAX_LENGTH); const stderr = new BoundedTextTail(AGENT_OUTPUT_TAIL_MAX_LENGTH); - const child = spawn(commandLine, [], { + const child = command.direct + ? spawn(command.executable, this.buildCommandArguments(command, invocation.prompt), { + stdio: ["ignore", "pipe", "pipe"], + shell: false, + }) + : spawn(this.buildCommandLine(command, invocation.prompt), [], { stdio: ["ignore", "pipe", "pipe"], shell: true, }); child.stdout?.on("data", (chunk: Buffer) => { - stdout.append(chunk.toString()); + const text = chunk.toString(); + stdout.append(text); + invocation.onActivity?.({ stream: "stdout", text }); }); child.stderr?.on("data", (chunk: Buffer) => { - stderr.append(chunk.toString()); + const text = chunk.toString(); + stderr.append(text); + invocation.onActivity?.({ stream: "stderr", text }); }); child.on("close", (code) => { @@ -70,7 +78,18 @@ export class AgentCliGateway implements IAgentGateway { throw new Error(`Unsupported agent: ${agentId}`); } - return AGENT_COMMANDS[agentId]; + const command = AGENT_COMMANDS[agentId]; + const override = process.env[`JUMBO_AGENT_COMMAND_${agentId.toUpperCase()}`]; + if (override !== undefined && override.trim().length > 0) { + const overrideArgs = this.resolveOverrideArgs(agentId); + return { + ...command, + executable: override.trim(), + ...(overrideArgs === undefined ? {} : { args: overrideArgs, direct: true }), + }; + } + + return command; } private buildCommandLine(command: AgentCommand, prompt: string): string { @@ -83,6 +102,28 @@ export class AgentCliGateway implements IAgentGateway { ].join(" "); } + private buildCommandArguments(command: AgentCommand, prompt: string): string[] { + return [ + ...(command.args ?? []), + ...(command.promptFlag === undefined ? [] : [command.promptFlag]), + prompt, + ]; + } + + private resolveOverrideArgs(agentId: SupportedAgentId): readonly string[] | undefined { + const value = process.env[`JUMBO_AGENT_ARGS_${agentId.toUpperCase()}`]; + if (value === undefined) { + return undefined; + } + + const parsed: unknown = JSON.parse(value); + if (!Array.isArray(parsed) || !parsed.every((argument) => typeof argument === "string")) { + throw new Error(`Invalid agent argument override for ${agentId}`); + } + + return parsed; + } + private isSupportedAgent(agentId: string): agentId is SupportedAgentId { return (SUPPORTED_AGENT_IDS as readonly string[]).includes(agentId); } diff --git a/src/infrastructure/daemons/NodeWorkerDaemonProcessController.ts b/src/infrastructure/daemons/NodeWorkerDaemonProcessController.ts index 6bf98ca0..fd12dba6 100644 --- a/src/infrastructure/daemons/NodeWorkerDaemonProcessController.ts +++ b/src/infrastructure/daemons/NodeWorkerDaemonProcessController.ts @@ -20,6 +20,8 @@ const __dirname = path.dirname(__filename); const PRESENTATION_WORK_DAEMON_DIRECTORY_SEGMENTS = ["..", "..", "presentation", "work"] as const; export class NodeWorkerDaemonProcessController implements IWorkerDaemonProcessController { + constructor(private readonly environment?: NodeJS.ProcessEnv) {} + spawnDaemonProcess( name: WorkerDaemonName, config: WorkerDaemonConfig, @@ -36,6 +38,7 @@ export class NodeWorkerDaemonProcessController implements IWorkerDaemonProcessCo cwd: process.cwd(), stdio: ["ignore", "pipe", "pipe"], detached: process.platform !== "win32", + ...(this.environment === undefined ? {} : { env: this.environment }), }); } @@ -83,7 +86,7 @@ export function getNodeWorkerDaemonTerminationStrategy( return { kind: "windows-tree", command: "taskkill", - args: ["/T", "/PID", String(pid)], + args: ["/F", "/T", "/PID", String(pid)], escalationArgs: ["/F", "/T", "/PID", String(pid)], }; } diff --git a/src/presentation/tui/cockpit/CockpitDaemonAgentConfigCycler.ts b/src/presentation/tui/cockpit/CockpitDaemonAgentConfigCycler.ts index f7e21e63..22f2cf75 100644 --- a/src/presentation/tui/cockpit/CockpitDaemonAgentConfigCycler.ts +++ b/src/presentation/tui/cockpit/CockpitDaemonAgentConfigCycler.ts @@ -1,6 +1,6 @@ import type { DaemonConfig } from "../daemon-subprocesses/ISubprocessManager.js"; -const COCKPIT_DAEMON_AGENT_OPTIONS = ["codex", "claude", "antigravity", "copilot", "cursor", "vibe"] as const; +const COCKPIT_DAEMON_AGENT_OPTIONS = ["codex", "claude", "antigravity", "copilot", "vibe"] as const; export function getNextCockpitDaemonAgentConfig(config: DaemonConfig): DaemonConfig { const currentIndex = COCKPIT_DAEMON_AGENT_OPTIONS.indexOf( diff --git a/src/presentation/tui/cockpit/CockpitDaemonPanel.tsx b/src/presentation/tui/cockpit/CockpitDaemonPanel.tsx index 48928acd..a1f14d2b 100644 --- a/src/presentation/tui/cockpit/CockpitDaemonPanel.tsx +++ b/src/presentation/tui/cockpit/CockpitDaemonPanel.tsx @@ -1,5 +1,5 @@ import React from "react"; -import { Box } from "ink"; +import { Box, Text } from "ink"; import { BaseColors } from "../../shared/DesignTokens.js"; import { Panel } from "../ui-primitives/Panel.js"; import type { DaemonConfig, SubprocessSnapshot } from "../daemon-subprocesses/ISubprocessManager.js"; @@ -41,6 +41,7 @@ export function CockpitDaemonPanel({ selected={selected} infoVisible={infoVisible} /> + {configuring && ( + candidate.category !== "heartbeat" && candidate.category !== "waiting" && candidate.category !== "foraging" + ) ?? latestEvent; + + if (event === undefined) { + return ( + + no recent daemon activity + + ); + } + + const parts = [ + event.goalId === undefined ? undefined : shortGoalId(event.goalId), + event.attempt === undefined && event.maxRetries === undefined + ? undefined + : `${event.attempt ?? "-"}/${event.maxRetries ?? "-"}`, + event.phase, + formatElapsed(latestElapsedMs(event, latestEvent)), + event.message, + ].filter((part): part is string => part !== undefined && part.trim().length > 0); + + return ( + + {truncateTail(parts.join(" "), 60)} + + ); +} + +function latestElapsedMs( + meaningfulEvent: SubprocessSnapshot["events"][number], + latestEvent: SubprocessSnapshot["events"][number] | undefined, +): number | undefined { + if ( + latestEvent?.category === "heartbeat" && + (latestEvent.goalId === undefined || latestEvent.goalId === meaningfulEvent.goalId) + ) { + return latestEvent.elapsedMs ?? meaningfulEvent.elapsedMs; + } + + return meaningfulEvent.elapsedMs; +} + +function shortGoalId(goalId: string): string { + return goalId.length > 8 ? goalId.slice(0, 8) : goalId; +} + +function formatElapsed(elapsedMs: number | undefined): string | undefined { + return elapsedMs === undefined ? undefined : `${Math.max(0, Math.floor(elapsedMs / 1000))}s`; +} + +function truncateTail(text: string, max: number): string { + return text.length <= max ? text : text.slice(0, max - 3) + "..."; +} + diff --git a/src/presentation/tui/cockpit/DaemonEventRowMessageFormatter.ts b/src/presentation/tui/cockpit/DaemonEventRowMessageFormatter.ts index 25267fd6..71fff4c2 100644 --- a/src/presentation/tui/cockpit/DaemonEventRowMessageFormatter.ts +++ b/src/presentation/tui/cockpit/DaemonEventRowMessageFormatter.ts @@ -19,7 +19,10 @@ function format( : event.message.trim(), event.goalId === undefined ? undefined : shortGoalId(event.goalId), formatAttemptDetails(event), + event.phase === undefined ? undefined : `[${event.phase}]`, + formatElapsed(event.elapsedMs), formatExitDetails(event), + event.errorType, event.errorMessage ?? snapshot.stderr[snapshot.stderr.length - 1], ].filter((part): part is string => part !== undefined && part.length > 0); @@ -38,6 +41,15 @@ function formatExitDetails(event: DaemonEventSnapshot): string | undefined { return event.exitCode === undefined ? undefined : `exit ${event.exitCode}`; } +function formatElapsed(elapsedMs: number | undefined): string | undefined { + if (elapsedMs === undefined) { + return undefined; + } + + const seconds = Math.max(0, Math.floor(elapsedMs / 1000)); + return `${seconds}s`; +} + function shortGoalId(goalId: string | undefined): string { if (goalId === undefined) { return "-"; diff --git a/src/presentation/tui/cockpit/DaemonEventRows.ts b/src/presentation/tui/cockpit/DaemonEventRows.ts index 6f2f552d..cace79ef 100644 --- a/src/presentation/tui/cockpit/DaemonEventRows.ts +++ b/src/presentation/tui/cockpit/DaemonEventRows.ts @@ -3,6 +3,12 @@ import type { DaemonEventRow } from "./DaemonEventRow.js"; import { DaemonEventRowNormalizer } from "./DaemonEventRowNormalizer.js"; const RENDERED_DAEMON_EVENT_LIMIT = 10; +const TERMINAL_ROW_CATEGORIES = new Set([ + "completed", + "failed", + "exhausted", + "stopped", +]); export const DaemonEventRows = { append, @@ -17,7 +23,7 @@ function fromSnapshots( DaemonEventRowNormalizer.fromSnapshot(snapshot, observedAtMs) ); - return newestFirst(rows).slice(0, RENDERED_DAEMON_EVENT_LIMIT); + return trimRows(newestFirst(rows)); } function append( @@ -31,12 +37,44 @@ function append( return currentRows; } - return newestFirst([...currentRows, ...appendedRows]).slice( - 0, - RENDERED_DAEMON_EVENT_LIMIT, - ); + return trimRows(newestFirst([...currentRows, ...appendedRows])); } function newestFirst(rows: readonly DaemonEventRow[]): readonly DaemonEventRow[] { return [...rows].sort((left, right) => right.timestampMs - left.timestampMs); } + +function trimRows(rows: readonly DaemonEventRow[]): readonly DaemonEventRow[] { + if (rows.length <= RENDERED_DAEMON_EVENT_LIMIT) { + return rows; + } + + const selected = rows.slice(0, RENDERED_DAEMON_EVENT_LIMIT); + for (const terminalRow of rows.filter(isTerminalRow)) { + if (selected.some((row) => row.key === terminalRow.key)) { + continue; + } + + const replaceIndex = findOldestNonTerminalIndex(selected); + if (replaceIndex === -1) { + break; + } + selected[replaceIndex] = terminalRow; + } + + return newestFirst(selected); +} + +function findOldestNonTerminalIndex(rows: readonly DaemonEventRow[]): number { + for (let index = rows.length - 1; index >= 0; index--) { + if (!isTerminalRow(rows[index])) { + return index; + } + } + + return -1; +} + +function isTerminalRow(row: DaemonEventRow): boolean { + return TERMINAL_ROW_CATEGORIES.has(row.category); +} diff --git a/src/presentation/tui/daemon-subprocesses/DaemonEventCategory.ts b/src/presentation/tui/daemon-subprocesses/DaemonEventCategory.ts index 36fd08ef..c42a6680 100644 --- a/src/presentation/tui/daemon-subprocesses/DaemonEventCategory.ts +++ b/src/presentation/tui/daemon-subprocesses/DaemonEventCategory.ts @@ -2,6 +2,8 @@ import { DaemonEventStatus } from "./DaemonEventStatus.js"; export const DaemonEventCategory = { MODEL_OUTPUT: "model-output", + STDERR: "stderr", + HEARTBEAT: "heartbeat", STOPPING: DaemonEventStatus.STOPPING, STOPPED: DaemonEventStatus.STOPPED, FAILED: DaemonEventStatus.FAILED, diff --git a/src/presentation/tui/daemon-subprocesses/DaemonEventSnapshot.ts b/src/presentation/tui/daemon-subprocesses/DaemonEventSnapshot.ts index bc72bd2d..7e1caa8c 100644 --- a/src/presentation/tui/daemon-subprocesses/DaemonEventSnapshot.ts +++ b/src/presentation/tui/daemon-subprocesses/DaemonEventSnapshot.ts @@ -9,9 +9,12 @@ export interface DaemonEventSnapshot { readonly category?: string; readonly message?: string; readonly timestampMs?: number; + readonly phase?: string; + readonly elapsedMs?: number; readonly goalId?: string; readonly attempt?: number; readonly maxRetries?: number; readonly exitCode?: number; + readonly errorType?: string; readonly errorMessage?: string; } diff --git a/src/presentation/tui/daemon-subprocesses/DaemonOutputEventParser.ts b/src/presentation/tui/daemon-subprocesses/DaemonOutputEventParser.ts index cb0bbd12..57d2fe9f 100644 --- a/src/presentation/tui/daemon-subprocesses/DaemonOutputEventParser.ts +++ b/src/presentation/tui/daemon-subprocesses/DaemonOutputEventParser.ts @@ -42,7 +42,9 @@ export class DaemonOutputEventParser { source: this.boundOptionalTextField(event.source), category: this.boundOptionalTextField(event.category), message: this.boundOptionalTextField(event.message), + phase: this.boundOptionalTextField(event.phase), goalId: this.boundOptionalTextField(event.goalId), + errorType: this.boundOptionalTextField(event.errorType), errorMessage: this.boundOptionalTextField(event.errorMessage), }; } diff --git a/src/presentation/tui/daemon-subprocesses/ManagedSubprocess.ts b/src/presentation/tui/daemon-subprocesses/ManagedSubprocess.ts index a1d72e68..68cbdd53 100644 --- a/src/presentation/tui/daemon-subprocesses/ManagedSubprocess.ts +++ b/src/presentation/tui/daemon-subprocesses/ManagedSubprocess.ts @@ -12,10 +12,12 @@ export interface ManagedSubprocess { readonly stdout: string[]; readonly stderr: string[]; readonly events: DaemonEventSnapshot[]; + readonly startedAtMs: number; status: SubprocessStatusValue; exitCode?: number | null; exitSignal?: string | null; stopRequested: boolean; terminationTimedOut: boolean; termination?: Promise; + heartbeatTimer?: ReturnType; } diff --git a/src/presentation/tui/daemon-subprocesses/SubprocessLifecycleEventRecorder.ts b/src/presentation/tui/daemon-subprocesses/SubprocessLifecycleEventRecorder.ts index abc93e95..aaa2dc32 100644 --- a/src/presentation/tui/daemon-subprocesses/SubprocessLifecycleEventRecorder.ts +++ b/src/presentation/tui/daemon-subprocesses/SubprocessLifecycleEventRecorder.ts @@ -8,6 +8,18 @@ import { SubprocessStatus } from "./SubprocessStatus.js"; import { DaemonOutputEventParser } from "./DaemonOutputEventParser.js"; const EVENT_RING_BUFFER_SIZE = 50; +const COALESCED_EVENT_CATEGORIES = new Set([ + "foraging", + "waiting", + "polling", + "heartbeat", +]); +const RETAINED_OUTCOME_CATEGORIES = new Set([ + "completed", + "failed", + "exhausted", + "stopped", +]); const SUBPROCESS_EVENT_COPY = { stopping: { @@ -73,9 +85,14 @@ export class SubprocessLifecycleEventRecorder { ): void { const boundedEvent = this.outputEventParser.boundEvent(event); this.logger.info(SubprocessCopy.eventLog, { daemon: process.name, event: boundedEvent }); + const coalescedIndex = this.findCoalescedEventIndex(process, boundedEvent); + if (coalescedIndex !== -1) { + process.events.splice(coalescedIndex, 1); + } process.events.push(boundedEvent); while (process.events.length > EVENT_RING_BUFFER_SIZE) { - process.events.shift(); + const removableIndex = this.findOldestRemovableEventIndex(process.events); + process.events.splice(removableIndex, 1); } } @@ -106,4 +123,44 @@ export class SubprocessLifecycleEventRecorder { ...event, }); } + + private findCoalescedEventIndex( + process: ManagedSubprocess, + event: DaemonEventSnapshot, + ): number { + if (event.category === undefined || !COALESCED_EVENT_CATEGORIES.has(event.category)) { + return -1; + } + + for (let index = process.events.length - 1; index >= 0; index--) { + const candidate = process.events[index]; + if (candidate.category === undefined || !COALESCED_EVENT_CATEGORIES.has(candidate.category)) { + return -1; + } + if ( + candidate.source === event.source && + candidate.status === event.status && + candidate.category === event.category && + candidate.message === event.message && + candidate.goalId === event.goalId + ) { + return index; + } + } + + return -1; + } + + private findOldestRemovableEventIndex(events: readonly DaemonEventSnapshot[]): number { + let retainedOutcomeIndex = -1; + for (let index = events.length - 1; index >= 0; index--) { + const category = events[index].category; + if (category !== undefined && RETAINED_OUTCOME_CATEGORIES.has(category)) { + retainedOutcomeIndex = index; + break; + } + } + const removableIndex = events.findIndex((_event, index) => index !== retainedOutcomeIndex); + return removableIndex === -1 ? 0 : removableIndex; + } } diff --git a/src/presentation/tui/daemon-subprocesses/SubprocessManager.ts b/src/presentation/tui/daemon-subprocesses/SubprocessManager.ts index e6a0ca90..dd5535fc 100644 --- a/src/presentation/tui/daemon-subprocesses/SubprocessManager.ts +++ b/src/presentation/tui/daemon-subprocesses/SubprocessManager.ts @@ -23,6 +23,8 @@ import { SubprocessNoOpLogger } from "./SubprocessNoOpLogger.js"; import { SubprocessOutputRingBuffer } from "./SubprocessOutputRingBuffer.js"; import { SubprocessSnapshotMapper } from "./SubprocessSnapshotMapper.js"; +const DAEMON_HEARTBEAT_INTERVAL_MS = 5_000; + export class SubprocessManager implements ISubprocessManager { private readonly processes = new Map(); private readonly configResolver = new SubprocessConfigResolver(); @@ -64,6 +66,7 @@ export class SubprocessManager implements ISubprocessManager { stdout: [], stderr: [], events: [], + startedAtMs: Date.now(), status: SubprocessStatus.RUNNING, stopRequested: false, terminationTimedOut: false, @@ -73,6 +76,7 @@ export class SubprocessManager implements ISubprocessManager { daemon: name, pid: child.pid, }); + this.startHeartbeat(managed); child.stdout?.on("data", (chunk) => { const text = this.outputRingBuffer.limitChunk(chunk.toString()); @@ -85,10 +89,22 @@ export class SubprocessManager implements ISubprocessManager { }); child.stderr?.on("data", (chunk) => { const text = this.outputRingBuffer.limitChunk(chunk.toString()); - this.outputRingBuffer.appendLines(managed.stderr, text); + const lines = this.outputRingBuffer.appendLines(managed.stderr, text); this.logger.warn(SubprocessCopy.stderrLog, { daemon: name, text }); + for (const line of lines) { + this.lifecycleEventRecorder.recordDaemonEvent(managed, { + daemon: name, + status: "processing", + source: name, + category: "stderr", + message: line, + timestampMs: Date.now(), + errorMessage: line, + }); + } }); child.on("close", (code, signal) => { + this.stopHeartbeat(managed); managed.exitCode = code; managed.exitSignal = signal; managed.status = this.resolveClosedStatus(managed, code); @@ -102,6 +118,7 @@ export class SubprocessManager implements ISubprocessManager { }); }); child.on("error", (error) => { + this.stopHeartbeat(managed); const errorMessage = this.outputEventParser.boundTextField(error.message); this.outputRingBuffer.appendLines(managed.stderr, errorMessage); if (!managed.stopRequested || managed.terminationTimedOut) { @@ -211,6 +228,7 @@ export class SubprocessManager implements ISubprocessManager { } if (result.status === "no-pid") { + this.stopHeartbeat(managed); managed.status = SubprocessStatus.STOPPED; this.lifecycleEventRecorder.recordStopped(managed); } @@ -234,6 +252,7 @@ export class SubprocessManager implements ISubprocessManager { managed: ManagedSubprocess, result: Extract, ): void { + this.stopHeartbeat(managed); managed.terminationTimedOut = true; managed.status = SubprocessStatus.FAILED; const errorMessage = this.outputEventParser.boundTextField( @@ -250,4 +269,60 @@ export class SubprocessManager implements ISubprocessManager { escalation: result.escalation, }); } + + private startHeartbeat(managed: ManagedSubprocess): void { + managed.heartbeatTimer = setInterval(() => { + if (managed.status !== SubprocessStatus.RUNNING) { + return; + } + + const latestEvent = managed.events[managed.events.length - 1]; + const now = Date.now(); + if ( + latestEvent?.timestampMs !== undefined && + now - latestEvent.timestampMs < DAEMON_HEARTBEAT_INTERVAL_MS + ) { + return; + } + + this.lifecycleEventRecorder.recordDaemonEvent(managed, { + daemon: managed.name, + status: "processing", + source: managed.name, + category: "heartbeat", + message: "daemon still running", + timestampMs: now, + phase: latestEvent?.phase ?? "running", + goalId: latestEvent?.goalId, + attempt: latestEvent?.attempt, + maxRetries: latestEvent?.maxRetries, + elapsedMs: this.resolveHeartbeatElapsedMs(managed, latestEvent, now), + }); + }, DAEMON_HEARTBEAT_INTERVAL_MS); + + if (typeof managed.heartbeatTimer === "object" && "unref" in managed.heartbeatTimer) { + managed.heartbeatTimer.unref(); + } + } + + private stopHeartbeat(managed: ManagedSubprocess): void { + if (managed.heartbeatTimer === undefined) { + return; + } + + clearInterval(managed.heartbeatTimer); + managed.heartbeatTimer = undefined; + } + + private resolveHeartbeatElapsedMs( + managed: ManagedSubprocess, + latestEvent: ManagedSubprocess["events"][number] | undefined, + now: number, + ): number { + if (latestEvent?.elapsedMs !== undefined && latestEvent.timestampMs !== undefined) { + return latestEvent.elapsedMs + Math.max(0, now - latestEvent.timestampMs); + } + + return Math.max(0, now - managed.startedAtMs); + } } diff --git a/tests/application/context/goals/codify/CodifierProcessManager.test.ts b/tests/application/context/goals/codify/CodifierProcessManager.test.ts index 97ceca33..d4b7a471 100644 --- a/tests/application/context/goals/codify/CodifierProcessManager.test.ts +++ b/tests/application/context/goals/codify/CodifierProcessManager.test.ts @@ -97,10 +97,11 @@ describe("CodifierProcessManager", () => { expect(result).toEqual({ status: "completed", goalId: "goal_1", attempts: 1 }); expect(codifyGoalController.handle).toHaveBeenCalledWith({ goalId: "goal_1" }); - expect(agentGateway.invoke).toHaveBeenCalledWith({ + expect(agentGateway.invoke).toHaveBeenCalledWith(expect.objectContaining({ agentId: "codex", prompt: expect.stringContaining("jumbo goal codify --id goal_1"), - }); + onActivity: expect.any(Function), + })); expect(agentGateway.invoke.mock.calls[0][0].prompt).toContain("jumbo goal close --id goal_1"); expect(events).toContainEqual(expect.objectContaining({ status: "processing", @@ -179,15 +180,22 @@ describe("CodifierProcessManager", () => { }); expect(result).toEqual({ status: "idle", attempts: 0 }); - expect(events).toEqual([ - { + expect(events).toEqual(expect.arrayContaining([ + expect.objectContaining({ daemon: "codifier", status: "idle", source: "codifier", category: "waiting", message: "awaiting approved goals", - }, - ]); + phase: "idle", + }), + expect.objectContaining({ + daemon: "codifier", + status: "processing", + category: "polling", + phase: "polling", + }), + ])); expect(codifyGoalController.handle).not.toHaveBeenCalled(); expect(agentGateway.invoke).not.toHaveBeenCalled(); }); diff --git a/tests/application/context/goals/refine/RefinerProcessManager.test.ts b/tests/application/context/goals/refine/RefinerProcessManager.test.ts index c0fd0e15..b5d0e09f 100644 --- a/tests/application/context/goals/refine/RefinerProcessManager.test.ts +++ b/tests/application/context/goals/refine/RefinerProcessManager.test.ts @@ -47,10 +47,11 @@ describe("RefinerProcessManager", () => { }); expect(goalStatusReader.findByStatus).toHaveBeenCalledWith(GoalStatus.TODO); expect(refineGoalController.handle).toHaveBeenCalledWith({ goalId: "goal_1" }); - expect(agentGateway.invoke).toHaveBeenCalledWith({ + expect(agentGateway.invoke).toHaveBeenCalledWith(expect.objectContaining({ agentId: "codex", - prompt: "Run the Jumbo refinement workflow for goal goal_1. Execute: jumbo goal refine --id goal_1", - }); + prompt: expect.stringContaining("jumbo goal commit --id goal_1"), + onActivity: expect.any(Function), + })); }); it("emits a structured foraging event when no goals are eligible", async () => { @@ -72,15 +73,24 @@ describe("RefinerProcessManager", () => { emit: (event) => events.push(event), })).resolves.toEqual({ status: "idle", attempts: 0 }); - expect(events).toEqual([ - { + expect(events).toEqual(expect.arrayContaining([ + expect.objectContaining({ daemon: "refiner", status: "idle", source: "refiner", category: "foraging", message: "foraging for defined goals", - }, - ]); + phase: "idle", + }), + ])); + expect(events).toEqual(expect.arrayContaining([ + expect.objectContaining({ + daemon: "refiner", + status: "processing", + category: "polling", + phase: "polling", + }), + ])); }); it("emits a structured failure event when refinement cannot start", async () => { @@ -141,13 +151,14 @@ describe("RefinerProcessManager", () => { attempts: 2, }); - expect(events).toEqual([ + expect(events).toEqual(expect.arrayContaining([ expect.objectContaining({ status: "processing", source: "refiner", category: "work-started", message: "refining goal", attempt: 1, + phase: "working", }), expect.objectContaining({ status: "skipped", @@ -155,14 +166,22 @@ describe("RefinerProcessManager", () => { category: "skipped", message: "goal not refined after agent attempt", attempt: 1, + phase: "retry", errorMessage: "Error loading configuration: config profile `prompt` not found", }), + expect.objectContaining({ + status: "processing", + category: "retry", + message: "retrying refinement", + attempt: 1, + }), expect.objectContaining({ status: "processing", source: "refiner", category: "work-started", message: "refining goal", attempt: 2, + phase: "working", }), expect.objectContaining({ status: "exhausted", @@ -170,9 +189,10 @@ describe("RefinerProcessManager", () => { category: "exhausted", message: "refinement attempts exhausted", attempt: 2, + phase: "exhausted", errorMessage: "Error loading configuration: config profile `prompt` not found", }), - ]); + ])); }); it("caps agent stderr in skipped and exhausted event error messages", async () => { diff --git a/tests/application/context/goals/review/ReviewerProcessManager.test.ts b/tests/application/context/goals/review/ReviewerProcessManager.test.ts index 30402d1b..fbf93978 100644 --- a/tests/application/context/goals/review/ReviewerProcessManager.test.ts +++ b/tests/application/context/goals/review/ReviewerProcessManager.test.ts @@ -47,10 +47,14 @@ describe("ReviewerProcessManager", () => { }); expect(goalStatusReader.findByStatus).toHaveBeenCalledWith(GoalStatus.SUBMITTED); expect(reviewGoalController.handle).toHaveBeenCalledWith({ goalId: "goal_1" }); - expect(agentGateway.invoke).toHaveBeenCalledWith({ + expect(agentGateway.invoke).toHaveBeenCalledWith(expect.objectContaining({ agentId: "codex", - prompt: "Run the Jumbo review workflow for goal goal_1. Execute: jumbo goal review --id goal_1", - }); + prompt: expect.stringContaining("jumbo goal approve --id goal_1"), + onActivity: expect.any(Function), + })); + expect(agentGateway.invoke.mock.calls[0][0].prompt).toContain( + 'jumbo goal reject --id goal_1 --review-issues "describe the issues"', + ); }); it("emits a structured waiting event when no goals are eligible", async () => { @@ -72,15 +76,22 @@ describe("ReviewerProcessManager", () => { emit: (event) => emittedEvents.push(event), })).resolves.toEqual({ status: "idle", attempts: 0 }); - expect(emittedEvents).toEqual([ - { + expect(emittedEvents).toEqual(expect.arrayContaining([ + expect.objectContaining({ daemon: "reviewer", status: "idle", source: "reviewer", category: "waiting", message: "awaiting submitted goals", - }, - ]); + phase: "idle", + }), + expect.objectContaining({ + daemon: "reviewer", + status: "processing", + category: "polling", + phase: "polling", + }), + ])); }); it("emits a structured failure event when review cannot start", async () => { @@ -136,8 +147,8 @@ describe("ReviewerProcessManager", () => { emit: (event) => emittedEvents.push(event), }); - expect(emittedEvents).toEqual([ - { + expect(emittedEvents).toEqual(expect.arrayContaining([ + expect.objectContaining({ daemon: "reviewer", status: "processing", source: "reviewer", @@ -146,24 +157,25 @@ describe("ReviewerProcessManager", () => { goalId: "goal_1", attempt: 1, maxRetries: 1, - }, - { + phase: "working", + }), + expect.objectContaining({ daemon: "reviewer", status: "processing", - source: "reviewer", + source: "agent", category: "model-output", message: "Review found a missing assertion.", goalId: "goal_1", - }, - { + }), + expect.objectContaining({ daemon: "reviewer", status: "processing", - source: "reviewer", + source: "agent", category: "model-output", message: "Recommend changes before approval.", goalId: "goal_1", - }, - { + }), + expect.objectContaining({ daemon: "reviewer", status: "completed", source: "reviewer", @@ -173,8 +185,9 @@ describe("ReviewerProcessManager", () => { attempt: 1, maxRetries: 1, exitCode: 0, - }, - ]); + phase: "completed", + }), + ])); }); it("caps reviewer model-output event messages from oversized agent stdout", async () => { diff --git a/tests/infrastructure/agents/AgentCliGateway.test.ts b/tests/infrastructure/agents/AgentCliGateway.test.ts index 7a7ee224..0e3c457b 100644 --- a/tests/infrastructure/agents/AgentCliGateway.test.ts +++ b/tests/infrastructure/agents/AgentCliGateway.test.ts @@ -32,6 +32,8 @@ describe("AgentCliGateway", () => { }); afterEach(() => { + delete process.env.JUMBO_AGENT_COMMAND_VIBE; + delete process.env.JUMBO_AGENT_ARGS_VIBE; stderrSpy.mockRestore(); }); @@ -140,6 +142,78 @@ describe("AgentCliGateway", () => { expect(stderrSpy).not.toHaveBeenCalled(); }); + it("allows a supported agent executable to be overridden for deterministic daemon automation", async () => { + const child = childProcess(); + spawnMock.mockReturnValue(child); + process.env.JUMBO_AGENT_COMMAND_VIBE = '"C:\\Tools\\fake-agent.exe" --mode success'; + + const promise = new AgentCliGateway(telemetryClient).invoke({ + agentId: "vibe", + prompt: "run review", + }); + child.emit("close", 0); + + await expect(promise).resolves.toEqual({ exitCode: 0, stdout: "", stderr: "" }); + expect(spawnMock).toHaveBeenCalledWith( + '"C:\\Tools\\fake-agent.exe" --mode success -p "run review"', + [], + expect.objectContaining({ shell: true }), + ); + }); + + it("launches structured override arguments directly without shell interpretation", async () => { + const child = childProcess(); + spawnMock.mockReturnValue(child); + process.env.JUMBO_AGENT_COMMAND_VIBE = "C:\\Program Files\\nodejs\\node.exe"; + process.env.JUMBO_AGENT_ARGS_VIBE = JSON.stringify([ + "C:\\fixtures\\fake-agent.js", + "--fake-mode", + "failure", + ]); + + const promise = new AgentCliGateway(telemetryClient).invoke({ + agentId: "vibe", + prompt: "run review", + }); + child.emit("close", 1); + + await expect(promise).resolves.toEqual({ exitCode: 1, stdout: "", stderr: "" }); + expect(spawnMock).toHaveBeenCalledWith( + "C:\\Program Files\\nodejs\\node.exe", + ["C:\\fixtures\\fake-agent.js", "--fake-mode", "failure", "-p", "run review"], + { + stdio: ["ignore", "pipe", "pipe"], + shell: false, + }, + ); + }); + + it("streams stdout and stderr activity while the agent child is still running", async () => { + const child = childProcess(); + spawnMock.mockReturnValue(child); + const activity: unknown[] = []; + + const promise = new AgentCliGateway(telemetryClient).invoke({ + agentId: "codex", + prompt: "run codify", + onActivity: (event: unknown) => activity.push(event), + }); + child.stdout.emit("data", Buffer.from("first stdout line\n")); + child.stderr.emit("data", Buffer.from("first stderr line\n")); + + expect(activity).toEqual([ + { stream: "stdout", text: "first stdout line\n" }, + { stream: "stderr", text: "first stderr line\n" }, + ]); + + child.emit("close", 0); + await expect(promise).resolves.toEqual({ + exitCode: 0, + stdout: "first stdout line\n", + stderr: "first stderr line\n", + }); + }); + it("returns a failed invocation result when the child process errors", async () => { const child = childProcess(); spawnMock.mockReturnValue(child); @@ -182,4 +256,14 @@ describe("AgentCliGateway", () => { ).rejects.toThrow("Unsupported agent: gemini"); expect(spawnMock).not.toHaveBeenCalled(); }); + + it("rejects Cursor because editor integrations are not daemon agent CLIs", async () => { + await expect( + new AgentCliGateway(telemetryClient).invoke({ + agentId: "cursor", + prompt: "run codify", + }), + ).rejects.toThrow("Unsupported agent: cursor"); + expect(spawnMock).not.toHaveBeenCalled(); + }); }); diff --git a/tests/infrastructure/daemons/NodeWorkerDaemonProcessController.test.ts b/tests/infrastructure/daemons/NodeWorkerDaemonProcessController.test.ts index 480d3773..3d70abc7 100644 --- a/tests/infrastructure/daemons/NodeWorkerDaemonProcessController.test.ts +++ b/tests/infrastructure/daemons/NodeWorkerDaemonProcessController.test.ts @@ -55,6 +55,28 @@ describe("NodeWorkerDaemonProcessController", () => { ); }); + it("passes an explicitly configured environment to the daemon child", () => { + const child = new EventEmitter(); + spawnMock.mockReturnValue(child); + const environment = { + PATH: "C:\\fake-agent-bin", + JUMBO_AGENT_COMMAND_VIBE: process.execPath, + }; + const controller = new NodeWorkerDaemonProcessController(environment); + + controller.spawnDaemonProcess("codifier", { + agentId: "vibe", + pollIntervalMs: 10_000, + maxRetries: 1, + }); + + expect(spawnMock).toHaveBeenCalledWith( + process.execPath, + expect.any(Array), + expect.objectContaining({ env: environment }), + ); + }); + it("defines Unix process-group signaling for non-Windows hosts", () => { expect(getNodeWorkerDaemonTerminationStrategy("linux", 456)).toEqual({ kind: "unix-process-group", @@ -68,7 +90,7 @@ describe("NodeWorkerDaemonProcessController", () => { expect(getNodeWorkerDaemonTerminationStrategy("win32", 456)).toEqual({ kind: "windows-tree", command: "taskkill", - args: ["/T", "/PID", "456"], + args: ["/F", "/T", "/PID", "456"], escalationArgs: ["/F", "/T", "/PID", "456"], }); }); @@ -107,7 +129,7 @@ describe("NodeWorkerDaemonProcessController", () => { jest.spyOn(controller, "getTerminationStrategy").mockReturnValue({ kind: "windows-tree", command: "taskkill", - args: ["/T", "/PID", "789"], + args: ["/F", "/T", "/PID", "789"], escalationArgs: ["/F", "/T", "/PID", "789"], }); @@ -117,17 +139,17 @@ describe("NodeWorkerDaemonProcessController", () => { await expect(termination).resolves.toEqual({ status: "closed", strategy: { - kind: "windows-tree", - command: "taskkill", - args: ["/T", "/PID", "789"], - escalationArgs: ["/F", "/T", "/PID", "789"], + kind: "windows-tree", + command: "taskkill", + args: ["/F", "/T", "/PID", "789"], + escalationArgs: ["/F", "/T", "/PID", "789"], }, exitCode: null, exitSignal: "SIGTERM", }); expect(execFileMock).toHaveBeenCalledWith( "taskkill", - ["/T", "/PID", "789"], + ["/F", "/T", "/PID", "789"], expect.any(Function), ); }); @@ -138,7 +160,7 @@ describe("NodeWorkerDaemonProcessController", () => { jest.spyOn(controller, "getTerminationStrategy").mockReturnValue({ kind: "windows-tree", command: "taskkill", - args: ["/T", "/PID", "790"], + args: ["/F", "/T", "/PID", "790"], escalationArgs: ["/F", "/T", "/PID", "790"], }); execFileMock.mockImplementation((_file, _args, callback) => callback(new Error("taskkill failed"), "", "")); @@ -187,7 +209,7 @@ describe("NodeWorkerDaemonProcessController", () => { jest.spyOn(controller, "getTerminationStrategy").mockReturnValue({ kind: "windows-tree", command: "taskkill", - args: ["/T", "/PID", "791"], + args: ["/F", "/T", "/PID", "791"], escalationArgs: ["/F", "/T", "/PID", "791"], }); @@ -199,7 +221,7 @@ describe("NodeWorkerDaemonProcessController", () => { strategy: { kind: "windows-tree", command: "taskkill", - args: ["/T", "/PID", "791"], + args: ["/F", "/T", "/PID", "791"], escalationArgs: ["/F", "/T", "/PID", "791"], }, timeoutMs: 25, @@ -211,7 +233,7 @@ describe("NodeWorkerDaemonProcessController", () => { }); expect(execFileMock).toHaveBeenCalledWith( "taskkill", - ["/T", "/PID", "791"], + ["/F", "/T", "/PID", "791"], expect.any(Function), ); expect(execFileMock).toHaveBeenCalledWith( diff --git a/tests/presentation/tui/application-shell/App.test.tsx b/tests/presentation/tui/application-shell/App.test.tsx index 0f63fbf1..b2049206 100644 --- a/tests/presentation/tui/application-shell/App.test.tsx +++ b/tests/presentation/tui/application-shell/App.test.tsx @@ -54,6 +54,30 @@ const waitForFrame = async ( throw new Error(`Timed out waiting for frame:\n${lastFrame() ?? ""}`); }; +const navigateFromCockpitToGoals = async ( + stdin: { write(input: string): void }, + lastFrame: () => string | undefined, +) => { + stdin.write("m"); + await waitForFrame(lastFrame, (frame) => frame.includes("Navigate")); + stdin.write("\u001B[B"); + await tick(); + stdin.write("\r"); + await waitForFrame(lastFrame, (frame) => frame.includes("GOALS//")); +}; + +const navigateFromGoalsToCockpit = async ( + stdin: { write(input: string): void }, + lastFrame: () => string | undefined, +) => { + stdin.write("m"); + await waitForFrame(lastFrame, (frame) => frame.includes("Navigate")); + stdin.write("\u001B[A"); + await tick(); + stdin.write("\r"); + await waitForFrame(lastFrame, (frame) => frame.includes("EVENTS//")); +}; + function App( props: React.ComponentProps, ): React.ReactElement { @@ -394,6 +418,159 @@ describe("App", () => { expect(terminateAll).not.toHaveBeenCalled(); }); + it("preserves one running daemon across navigation and stops it exactly once on request", async () => { + let refinerSnapshot: SubprocessSnapshot = { + name: "refiner", + status: "running", + pid: 4101, + config: stoppedDaemonSnapshot.config, + stdout: [], + stderr: [], + events: [{ + daemon: "refiner", + status: "working", + category: "activity", + goalId: "goal_single_navigation", + phase: "agent", + elapsedMs: 1_000, + timestampMs: 1_000, + message: "single navigation activity", + }], + }; + const terminate = jest.fn(async (name: DaemonName) => { + refinerSnapshot = { ...refinerSnapshot, name, status: "stopped" }; + return refinerSnapshot; + }); + const terminateAll = jest.fn(async () => {}); + const manager: ISubprocessManager = { + spawn: jest.fn(async () => refinerSnapshot), + terminate, + terminateAll, + getStatus: jest.fn(() => refinerSnapshot), + getAllStatuses: jest.fn(() => [refinerSnapshot]), + }; + const { stdin, lastFrame, unmount } = render( + , + ); + + try { + await waitForFrame(lastFrame, (frame) => frame.includes("EVENTS//")); + await navigateFromCockpitToGoals(stdin, lastFrame); + + refinerSnapshot = { + ...refinerSnapshot, + events: [ + ...refinerSnapshot.events, + { + daemon: "refiner", + status: "working", + category: "heartbeat", + goalId: "goal_single_navigation", + elapsedMs: 6_000, + timestampMs: 6_000, + message: "still working", + }, + ], + }; + + await navigateFromGoalsToCockpit(stdin, lastFrame); + await waitForFrame(lastFrame, (frame) => + frame.includes("single navigation activity") && frame.includes("6s"), + ); + + expect(manager.spawn).not.toHaveBeenCalled(); + expect(terminate).not.toHaveBeenCalled(); + expect(terminateAll).not.toHaveBeenCalled(); + expect(manager.getStatus("refiner").pid).toBe(4101); + + stdin.write("s"); + await waitForFrame(lastFrame, () => terminate.mock.calls.length === 1); + await tick(); + + expect(terminate).toHaveBeenCalledTimes(1); + expect(terminate).toHaveBeenCalledWith("refiner"); + expect(terminateAll).not.toHaveBeenCalled(); + } finally { + unmount(); + } + }, 10000); + + it("preserves multiple independently updating daemons across navigation", async () => { + const snapshots = new Map([ + ["refiner", { + name: "refiner", status: "running", pid: 5101, + config: stoppedDaemonSnapshot.config, stdout: [], stderr: [], + events: [{ daemon: "refiner", status: "working", category: "activity", goalId: "goal_refiner", timestampMs: 7_000, message: "refiner resumed" }], + }], + ["reviewer", { + name: "reviewer", status: "running", pid: 5102, + config: stoppedDaemonSnapshot.config, stdout: [], stderr: [], + events: [{ daemon: "reviewer", status: "working", category: "activity", goalId: "goal_reviewer", timestampMs: 8_000, message: "reviewer resumed" }], + }], + ["codifier", { + name: "codifier", status: "running", pid: 5103, + config: stoppedDaemonSnapshot.config, stdout: [], stderr: [], + events: [{ daemon: "codifier", status: "working", category: "activity", goalId: "goal_codifier", timestampMs: 9_000, message: "codifier resumed" }], + }], + ]); + const manager: ISubprocessManager = { + spawn: jest.fn(async (name) => snapshots.get(name)!), + terminate: jest.fn(async (name) => snapshots.get(name)!), + terminateAll: jest.fn(async () => {}), + getStatus: jest.fn((name) => snapshots.get(name)!), + getAllStatuses: jest.fn(() => [...snapshots.values()]), + }; + const { stdin, lastFrame, unmount } = render( + , + ); + + try { + await waitForFrame(lastFrame, (frame) => frame.includes("EVENTS//")); + await navigateFromCockpitToGoals(stdin, lastFrame); + + for (const [name, snapshot] of snapshots) { + snapshots.set(name, { + ...snapshot, + events: snapshot.events.map((event) => ({ + ...event, + elapsedMs: name === "refiner" ? 7_000 : name === "reviewer" ? 8_000 : 9_000, + })), + }); + } + + await navigateFromGoalsToCockpit(stdin, lastFrame); + const frame = await waitForFrame(lastFrame, (currentFrame) => + currentFrame.includes("refiner resumed") && + currentFrame.includes("reviewer resumed") && + currentFrame.includes("codifier resumed"), + ); + + expect(frame).toContain("7s"); + expect(frame).toContain("8s"); + expect(frame).toContain("9s"); + expect(manager.getStatus("refiner").pid).toBe(5101); + expect(manager.getStatus("reviewer").pid).toBe(5102); + expect(manager.getStatus("codifier").pid).toBe(5103); + expect(manager.spawn).not.toHaveBeenCalled(); + expect(manager.terminate).not.toHaveBeenCalled(); + expect(manager.terminateAll).not.toHaveBeenCalled(); + } finally { + unmount(); + } + }, 10000); + it("does not open goal authoring from uninitialized cockpit state", async () => { const { stdin, lastFrame, unmount } = render( { expect(element.props.subprocessManager).toBe(manager); }); - it("creates one subprocess manager per launch and terminates it after Ink exits", async () => { + it("keeps one launcher-owned subprocess manager for the App lifetime and terminates it exactly once after Ink exits", async () => { const terminateAll = jest.fn<() => Promise>().mockResolvedValue(); const manager = subprocessManager(terminateAll); const factory = jest.fn(() => manager); diff --git a/tests/presentation/tui/cockpit/CockpitDaemonAgentConfigCycler.test.ts b/tests/presentation/tui/cockpit/CockpitDaemonAgentConfigCycler.test.ts index f695291c..02334245 100644 --- a/tests/presentation/tui/cockpit/CockpitDaemonAgentConfigCycler.test.ts +++ b/tests/presentation/tui/cockpit/CockpitDaemonAgentConfigCycler.test.ts @@ -22,6 +22,16 @@ describe("CockpitDaemonAgentConfigCycler", () => { }); }); + it("cycles from Copilot to Vibe without offering Cursor", () => { + const config = { agentId: "copilot", pollIntervalMs: 30_000, maxRetries: 3 }; + + expect(getNextCockpitDaemonAgentConfig(config)).toEqual({ + agentId: "vibe", + pollIntervalMs: 30_000, + maxRetries: 3, + }); + }); + it("starts at the first agent when the current agent is not configured", () => { expect(getNextCockpitDaemonAgentConfig({ agentId: "unknown-agent", diff --git a/tests/presentation/tui/cockpit/CockpitDaemonPanel.test.tsx b/tests/presentation/tui/cockpit/CockpitDaemonPanel.test.tsx index 98f534be..14016e40 100644 --- a/tests/presentation/tui/cockpit/CockpitDaemonPanel.test.tsx +++ b/tests/presentation/tui/cockpit/CockpitDaemonPanel.test.tsx @@ -70,4 +70,47 @@ describe("CockpitDaemonPanel", () => { expect(lastFrame()).toContain(CockpitDaemonPanelCopy.pidLabel); unmount(); }); + + it("uses heartbeat elapsed time while retaining the latest meaningful activity", () => { + const { lastFrame, unmount } = render( + + daemon frame + , + ); + + expect(lastFrame()).toContain("goal_hea"); + expect(lastFrame()).toContain("agent"); + expect(lastFrame()).toContain("6s"); + expect(lastFrame()).toContain("inspecting goal"); + expect(lastFrame()).not.toContain("still working"); + unmount(); + }); }); diff --git a/tests/presentation/tui/cockpit/DaemonEventRows.test.ts b/tests/presentation/tui/cockpit/DaemonEventRows.test.ts index a9bc1f1a..f68771d0 100644 --- a/tests/presentation/tui/cockpit/DaemonEventRows.test.ts +++ b/tests/presentation/tui/cockpit/DaemonEventRows.test.ts @@ -41,9 +41,26 @@ describe("DaemonEventRows", () => { expect(DaemonEventRows.append(current, next).map((row) => row.message)).toEqual(["new", "old"]); expect(DaemonEventRows.append(current, current)).toBe(current); }); + + it("keeps terminal outcomes visible when newer idle rows exceed the rendered limit", () => { + const terminal = snapshot("goal failed", 1000, "failed", "failed"); + const idleRows = Array.from({ length: 11 }, (_, index) => + snapshot(`idle-${index}`, 2000 + index, "idle", "waiting") + ); + + const rows = DaemonEventRows.fromSnapshots([terminal, ...idleRows], 3000); + + expect(rows.map((row) => row.message)).toContain("goal failed"); + expect(rows).toHaveLength(10); + }); }); -function snapshot(message: string, timestampMs: number): SubprocessSnapshot { +function snapshot( + message: string, + timestampMs: number, + status = "processing", + category = "work", +): SubprocessSnapshot { return { name: "refiner", status: "running", @@ -52,7 +69,8 @@ function snapshot(message: string, timestampMs: number): SubprocessSnapshot { stderr: [], events: [{ daemon: "refiner", - status: "processing", + status, + category, message, timestampMs, }], diff --git a/tests/presentation/tui/daemon-subprocesses/SubprocessManager.test.ts b/tests/presentation/tui/daemon-subprocesses/SubprocessManager.test.ts index 9000e245..fe64189c 100644 --- a/tests/presentation/tui/daemon-subprocesses/SubprocessManager.test.ts +++ b/tests/presentation/tui/daemon-subprocesses/SubprocessManager.test.ts @@ -237,6 +237,73 @@ describe("SubprocessManager", () => { expect(events[49]).toEqual(expect.objectContaining({ goalId: "goal_54" })); }); + it("coalesces alternating idle polls without evicting the latest terminal outcome", async () => { + const child = childProcess(); + const manager = new SubprocessManager(processController(child)); + + await manager.spawn("refiner"); + child.stdout?.emit("data", Buffer.from(`${JSON.stringify({ + daemon: "refiner", + status: "completed", + category: "completed", + message: "goal completed", + goalId: "goal_done", + })}\n`)); + + const idleCycle = [ + JSON.stringify({ daemon: "refiner", status: "processing", category: "polling", message: "polling for goals" }), + JSON.stringify({ daemon: "refiner", status: "idle", category: "waiting", message: "no eligible goals" }), + ].join("\n"); + for (let index = 0; index < 60; index++) { + child.stdout?.emit("data", Buffer.from(`${idleCycle}\n`)); + } + + const events = manager.getStatus("refiner").events; + expect(events).toHaveLength(3); + expect(events).toEqual(expect.arrayContaining([ + expect.objectContaining({ category: "completed", goalId: "goal_done" }), + expect.objectContaining({ category: "polling" }), + expect.objectContaining({ category: "waiting" }), + ])); + }); + + it("emits a heartbeat with advancing elapsed time when daemon activity pauses", async () => { + jest.useFakeTimers(); + jest.setSystemTime(1767272400000); + const child = childProcess(); + const manager = new SubprocessManager(processController(child)); + + await manager.spawn("reviewer"); + child.stdout?.emit("data", Buffer.from(`${JSON.stringify({ + daemon: "reviewer", + status: "processing", + category: "selection", + message: "selected goal", + phase: "selection", + goalId: "goal_heartbeat", + elapsedMs: 125, + timestampMs: 1767272400000, + })}\n`)); + + jest.advanceTimersByTime(5_000); + const firstHeartbeat = manager.getStatus("reviewer").events.at(-1); + expect(firstHeartbeat).toEqual(expect.objectContaining({ + category: "heartbeat", + phase: "selection", + goalId: "goal_heartbeat", + elapsedMs: 5_125, + })); + + jest.advanceTimersByTime(5_000); + expect(manager.getStatus("reviewer").events.at(-1)).toEqual(expect.objectContaining({ + category: "heartbeat", + elapsedMs: 10_125, + })); + + child.emit("close", 0, null); + jest.useRealTimers(); + }); + it("logs daemon stderr and child process failures through ILogger", async () => { const child = childProcess(); const testLogger = logger(); @@ -256,8 +323,17 @@ describe("SubprocessManager", () => { expect.any(Error), { daemon: "refiner", stopRequested: false, status: "failed" }, ); - expect(manager.getStatus("refiner").events).toEqual([ - { + expect(manager.getStatus("refiner").events).toEqual(expect.arrayContaining([ + expect.objectContaining({ + daemon: "refiner", + status: "processing", + source: "refiner", + category: "stderr", + message: "refiner failed", + timestampMs: 1767272400000, + errorMessage: "refiner failed", + }), + expect.objectContaining({ daemon: "refiner", status: "failed", source: "refiner", @@ -265,8 +341,8 @@ describe("SubprocessManager", () => { message: "process failed", timestampMs: 1767272400000, errorMessage: "spawn failed", - }, - ]); + }), + ])); }); it("delegates process termination and records stopping and stopped lifecycle events", async () => { diff --git a/tests/presentation/work/worker-daemon.production.integration.test.ts b/tests/presentation/work/worker-daemon.production.integration.test.ts new file mode 100644 index 00000000..777d266d --- /dev/null +++ b/tests/presentation/work/worker-daemon.production.integration.test.ts @@ -0,0 +1,335 @@ +import { afterAll, beforeAll, describe, expect, it, jest } from "@jest/globals"; +import { pathToFileURL } from "node:url"; +import { fileURLToPath } from "node:url"; +import fs from "fs-extra"; +import os from "node:os"; +import path from "node:path"; +import { Host } from "../../../src/infrastructure/host/Host.js"; +import { SubprocessManager } from "../../../src/presentation/tui/daemon-subprocesses/SubprocessManager.js"; +import type { IApplicationContainer } from "../../../src/application/host/IApplicationContainer.js"; +import type { WorkerDaemonName } from "../../../src/application/daemons/WorkerDaemonCatalog.js"; +import type { DaemonEventSnapshot } from "../../../src/presentation/tui/daemon-subprocesses/ISubprocessManager.js"; +import { GoalStatus } from "../../../src/domain/goals/Constants.js"; + +jest.setTimeout(90_000); + +const __filename = fileURLToPath(import.meta.url); +const __dirname = path.dirname(__filename); +const PROJECT_ROOT = path.resolve(__dirname, "..", "..", ".."); +const COMPILED_CONTROLLER = path.join( + PROJECT_ROOT, + "dist", + "infrastructure", + "daemons", + "NodeWorkerDaemonProcessController.js", +); +const COMPILED_CLI = path.join(PROJECT_ROOT, "dist", "cli.js"); + +let workspaceRoot: string; +let originalPath: string | undefined; +let originalPathUpper: string | undefined; +let fakeAgentScript: string; +let fakeAgentBin: string; +let consoleErrorSpy: jest.SpiedFunction; + +describe("production worker daemons through TUI process wiring", () => { + beforeAll(async () => { + if (!(await fs.pathExists(COMPILED_CONTROLLER)) || !(await fs.pathExists(COMPILED_CLI))) { + throw new Error("Compiled daemon artifacts missing. Run `npm run build` before this integration suite."); + } + + workspaceRoot = await fs.realpath( + await fs.mkdtemp(path.join(os.tmpdir(), "jumbo-worker-daemons-")), + ); + fakeAgentScript = await writeFakeAgent(workspaceRoot); + + fakeAgentBin = path.join(workspaceRoot, "fake-agent-bin"); + await writeFakeAgentLaunchers(fakeAgentBin, fakeAgentScript); + + originalPath = process.env.Path; + originalPathUpper = process.env.PATH; + const inheritedPath = originalPath ?? originalPathUpper ?? ""; + process.env.Path = `${fakeAgentBin}${path.delimiter}${inheritedPath}`; + process.env.PATH = `${fakeAgentBin}${path.delimiter}${inheritedPath}`; + consoleErrorSpy = jest.spyOn(console, "error").mockImplementation(() => undefined); + }); + + afterAll(async () => { + consoleErrorSpy?.mockRestore(); + process.env.Path = originalPath; + process.env.PATH = originalPathUpper; + + if (workspaceRoot !== undefined) { + await removeIfAvailable(workspaceRoot); + } + }); + + it.each([ + ["refiner", GoalStatus.REFINED, "completed"], + ["reviewer", GoalStatus.QUALIFIED, "completed"], + ["codifier", GoalStatus.DONE, "completed"], + ] as const)( + "executes the %s entrypoint to %s through the real process controller", + async (daemon, expectedStatus, expectedEventStatus) => { + const fixture = await createFixtureProject(`${daemon}-success`); + const goalId = await prepareGoal(fixture.container, daemon); + const manager = await createProductionManager(); + + const previousCwd = process.cwd(); + process.chdir(fixture.projectRoot); + try { + await manager.spawn(daemon, { agentId: "vibe", maxRetries: 1, pollIntervalMs: 10_000 }); + const event = await waitForDaemonEvent(manager, daemon, expectedEventStatus); + expect(event).toEqual(expect.objectContaining({ + daemon, + goalId, + status: expectedEventStatus, + category: expectedEventStatus, + })); + expect(manager.getStatus(daemon).events).toEqual(expect.arrayContaining([ + expect.objectContaining({ category: "polling", phase: "polling" }), + expect.objectContaining({ category: "selection", goalId }), + expect.objectContaining({ source: "agent", category: "model-output", goalId }), + ])); + expect((await fixture.container.goalStatusReader.findById(goalId))?.status).toBe(expectedStatus); + } finally { + await manager.terminate(daemon); + process.chdir(previousCwd); + fixture.host.dispose(); + } + }, + ); + + it.each([ + ["refiner", "exhausted"], + ["reviewer", "exhausted"], + ["codifier", "exhausted"], + ] as const)( + "reports deterministic agent failure for the %s entrypoint", + async (daemon, expectedEventStatus) => { + const fixture = await createFixtureProject(`${daemon}-failure`); + const goalId = await prepareGoal(fixture.container, daemon); + const manager = await createProductionManager(); + + const previousCwd = process.cwd(); + process.chdir(fixture.projectRoot); + try { + await manager.spawn(daemon, { agentId: "antigravity", maxRetries: 1, pollIntervalMs: 10_000 }); + const event = await waitForDaemonEvent(manager, daemon, expectedEventStatus); + expect(event).toEqual(expect.objectContaining({ + daemon, + goalId, + status: expectedEventStatus, + category: expectedEventStatus, + exitCode: 1, + })); + expect(manager.getStatus(daemon).events).toEqual(expect.arrayContaining([ + expect.objectContaining({ source: "agent", category: "agent-stderr", goalId }), + ])); + } finally { + await manager.terminate(daemon); + process.chdir(previousCwd); + fixture.host.dispose(); + } + }, + ); +}); + +async function createProductionManager(): Promise { + const imported = await import(pathToFileURL(COMPILED_CONTROLLER).href); + const inheritedPath = originalPath ?? originalPathUpper ?? ""; + return new SubprocessManager(new imported.NodeWorkerDaemonProcessController({ + ...process.env, + Path: `${fakeAgentBin}${path.delimiter}${inheritedPath}`, + PATH: `${fakeAgentBin}${path.delimiter}${inheritedPath}`, + JUMBO_FAKE_AGENT_CLI: COMPILED_CLI, + JUMBO_AGENT_COMMAND_VIBE: process.execPath, + JUMBO_AGENT_ARGS_VIBE: JSON.stringify([ + fakeAgentScript, + "--fake-mode", + "success", + ]), + JUMBO_AGENT_COMMAND_ANTIGRAVITY: process.execPath, + JUMBO_AGENT_ARGS_ANTIGRAVITY: JSON.stringify([ + fakeAgentScript, + "--fake-mode", + "failure", + ]), + })); +} + +async function createFixtureProject(name: string): Promise<{ + readonly projectRoot: string; + readonly host: Host; + readonly container: IApplicationContainer; +}> { + const projectRootPath = path.join(workspaceRoot, name); + await fs.mkdirp(path.join(projectRootPath, ".jumbo")); + const projectRoot = await fs.realpath(projectRootPath); + const host = new Host(path.join(projectRoot, ".jumbo")); + const container = await host.createBuilder().build(); + return { projectRoot, host, container }; +} + +async function prepareGoal( + container: IApplicationContainer, + daemon: WorkerDaemonName, +): Promise { + const { goalId } = await container.addGoalController.handle({ + title: `${daemon} daemon fixture`, + objective: `Exercise ${daemon} daemon fixture`, + successCriteria: ["Daemon fixture completes"], + }); + + if (daemon === "refiner") { + return goalId; + } + + await container.refineGoalController.handle({ goalId }); + await container.commitGoalController.handle({ goalId }); + await container.startGoalController.handle({ goalId }); + await container.submitGoalController.handle({ goalId }); + + if (daemon === "reviewer") { + return goalId; + } + + await container.reviewGoalController.handle({ goalId }); + await container.qualifyGoalController.handle({ goalId }); + return goalId; +} + +async function waitForDaemonEvent( + manager: SubprocessManager, + daemon: WorkerDaemonName, + status: string, +): Promise { + const deadline = Date.now() + 45_000; + let lastSnapshot = manager.getStatus(daemon); + while (Date.now() < deadline) { + const snapshot = manager.getStatus(daemon); + lastSnapshot = snapshot; + const event = snapshot.events.find((candidate) => candidate.status === status); + if (event !== undefined) { + return event; + } + + const failed = snapshot.events.find((candidate) => candidate.status === "failed"); + if (failed !== undefined) { + throw new Error(`Daemon ${daemon} failed before ${status}: ${failed.errorMessage ?? failed.message ?? "unknown error"}`); + } + + await new Promise((resolve) => setTimeout(resolve, 50)); + } + + throw new Error( + `Timed out waiting for ${daemon} event ${status}. Snapshot: ${JSON.stringify({ + status: lastSnapshot.status, + stderr: lastSnapshot.stderr, + events: lastSnapshot.events, + exitCode: lastSnapshot.exitCode, + exitSignal: lastSnapshot.exitSignal, + })}`, + ); +} + +async function writeFakeAgent(directory: string): Promise { + const scriptPath = path.join(directory, "fake-agent.js"); + await fs.writeFile(scriptPath, FAKE_AGENT_SCRIPT, "utf8"); + return scriptPath; +} + +async function writeFakeAgentLaunchers(directory: string, scriptPath: string): Promise { + await fs.mkdirp(directory); + await Promise.all([ + writeFakeAgentLauncher(directory, "vibe", scriptPath, "success"), + writeFakeAgentLauncher(directory, "antigravity", scriptPath, "failure"), + ]); +} + +async function writeFakeAgentLauncher( + directory: string, + agentId: string, + scriptPath: string, + mode: "success" | "failure", +): Promise { + if (process.platform === "win32") { + const launcherPath = path.join(directory, `${agentId}.cmd`); + await fs.writeFile( + launcherPath, + `@echo off\r\n"${process.execPath}" "${scriptPath}" --fake-mode ${mode} %*\r\nexit /b %errorlevel%\r\n`, + "utf8", + ); + return; + } + + const launcherPath = path.join(directory, agentId); + await fs.writeFile( + launcherPath, + `#!/bin/sh\nexec "${process.execPath}" "${scriptPath}" --fake-mode ${mode} "$@"\n`, + "utf8", + ); + await fs.chmod(launcherPath, 0o755); +} + +async function removeIfAvailable(target: string): Promise { + try { + await fs.remove(target); + } catch { + // Windows can hold short-lived locks on SQLite/WAL files after spawned + // daemon processes exit. The fixture lives under the OS temp directory. + } +} + +const FAKE_AGENT_SCRIPT = String.raw` +const { spawnSync } = require("node:child_process"); +const fs = require("node:fs"); + +const modeIndex = process.argv.indexOf("--fake-mode"); +const mode = modeIndex === -1 ? "success" : process.argv[modeIndex + 1]; +const cli = process.env.JUMBO_FAKE_AGENT_CLI; +const prompt = process.argv + .filter((_, index) => index !== modeIndex && index !== modeIndex + 1) + .slice(2) + .join(" "); + +fs.writeSync(process.stdout.fd, "fake agent activity: received prompt\n"); + +if (mode === "failure") { + fs.writeSync(process.stderr.fd, "fake agent failure\n"); + process.exit(1); +} + +const command = findCommand(prompt); +if (!command) { + fs.writeSync(process.stderr.fd, "fake agent could not find a Jumbo completion command\n"); + process.exit(1); +} + +const result = spawnSync(process.execPath, [cli, ...command], { + cwd: process.cwd(), + encoding: "utf8", +}); + +if (result.stdout) { + fs.writeSync(process.stdout.fd, result.stdout); +} +if (result.stderr) { + fs.writeSync(process.stderr.fd, result.stderr); +} + +process.exit(result.status === null ? 1 : result.status); + +function findCommand(prompt) { + const commit = /jumbo goal commit --id ([^\s]+)/.exec(prompt); + if (commit) return ["goal", "commit", "--id", commit[1]]; + + const approve = /jumbo goal approve --id ([^\s]+)/.exec(prompt); + if (approve) return ["goal", "approve", "--id", approve[1]]; + + const close = /jumbo goal close --id ([^\s]+)/.exec(prompt); + if (close) return ["goal", "close", "--id", close[1]]; + + return null; +} +`;