diff --git a/src/main/agent-runtime/agent-hooks.ts b/src/main/agent-runtime/agent-hooks.ts index 8a499b7d..eca2236c 100644 --- a/src/main/agent-runtime/agent-hooks.ts +++ b/src/main/agent-runtime/agent-hooks.ts @@ -78,8 +78,8 @@ export interface AgentHookBundle { * - `beforeToolCall`: fail-closed RPC. A thrown error travels up into * pi's loop and surfaces as a tool-error result — destructive tools * must never silently bypass main's approval gate. - * - `afterToolCall`: fail-open RPC. Projection is observational; if - * main cannot be reached the tool result stands unchanged. + * - `afterToolCall`: fail-closed RPC. The tool has already run, so + * main must durably acknowledge the artifact projection result. * - `transformContext`: worker-local compaction. The transcript stays * inside the worker; main receives telemetry only. */ @@ -120,10 +120,7 @@ export function buildAgentHooks(deps: AgentHooksDeps): AgentHookBundle { }; const response = await rpc.invoke(runId, "afterToolCall", payload); if (!response.ok) { - // After-hook failure is informational: the tool already ran, - // projection is a best-effort write. Drop the hook rather than - // corrupting the tool result. - return undefined; + throw new Error(`afterToolCall RPC failed: ${response.error}`); } const result = response.result as AfterToolCallRpcResult; const override: AfterToolCallResult = {}; diff --git a/src/main/agent-runtime/artifact-projection-events.ts b/src/main/agent-runtime/artifact-projection-events.ts new file mode 100644 index 00000000..e940d1f4 --- /dev/null +++ b/src/main/agent-runtime/artifact-projection-events.ts @@ -0,0 +1,105 @@ +import type { AppDatabase } from "@main/database"; +import type { ProjectionOutcome } from "./artifact-types"; + +export type ArtifactProjectionPhase = "after_tool_call" | "completion_reconcile"; + +export interface ArtifactProjectionEventInput { + companyId: string; + runId: string; + agentId: string; + phase: ArtifactProjectionPhase; + workspaceId: string | null; + workspaceMode: "exclusive" | "shared" | "none"; + toolName?: string | null; + toolCallId?: string | null; + outcomes: ProjectionOutcome[]; +} + +export interface ArtifactProjectionFailureInput { + companyId: string; + runId: string; + agentId: string; + phase: ArtifactProjectionPhase; + workspaceId: string | null; + workspaceMode: "exclusive" | "shared" | "none"; + toolName?: string | null; + toolCallId?: string | null; + errorMessage: string; +} + +interface ProjectionCounts { + inserted: number; + updated: number; + retired: number; + skipped: number; +} + +function countOutcomes(outcomes: ProjectionOutcome[]): ProjectionCounts { + const counts: ProjectionCounts = { + inserted: 0, + updated: 0, + retired: 0, + skipped: 0, + }; + for (const outcome of outcomes) { + counts[outcome.kind] += 1; + } + return counts; +} + +function summarizeOutcomes(phase: ArtifactProjectionPhase, counts: ProjectionCounts): string { + return [ + `Artifact projection ${phase} acknowledged`, + `${counts.inserted} inserted`, + `${counts.updated} updated`, + `${counts.retired} retired`, + `${counts.skipped} skipped`, + ].join("; "); +} + +export function appendArtifactProjectionEvent( + db: AppDatabase, + input: ArtifactProjectionEventInput, +): ProjectionCounts { + const counts = countOutcomes(input.outcomes); + db.core.runs.appendRunEvent({ + companyId: input.companyId, + runId: input.runId, + agentId: input.agentId, + eventType: "artifact_projection", + level: "info", + message: summarizeOutcomes(input.phase, counts), + payload: { + phase: input.phase, + workspaceId: input.workspaceId, + workspaceMode: input.workspaceMode, + toolName: input.toolName ?? null, + toolCallId: input.toolCallId ?? null, + counts, + outcomes: input.outcomes, + }, + }); + return counts; +} + +export function appendArtifactProjectionFailureEvent( + db: AppDatabase, + input: ArtifactProjectionFailureInput, +): void { + db.core.runs.appendRunEvent({ + companyId: input.companyId, + runId: input.runId, + agentId: input.agentId, + eventType: "artifact_projection", + level: "error", + message: `Artifact projection ${input.phase} failed: ${input.errorMessage}`, + payload: { + phase: input.phase, + workspaceId: input.workspaceId, + workspaceMode: input.workspaceMode, + toolName: input.toolName ?? null, + toolCallId: input.toolCallId ?? null, + errorMessage: input.errorMessage, + }, + }); +} diff --git a/src/main/agent-runtime/artifact-projector.ts b/src/main/agent-runtime/artifact-projector.ts index fde232ec..8d4e57a3 100644 --- a/src/main/agent-runtime/artifact-projector.ts +++ b/src/main/agent-runtime/artifact-projector.ts @@ -81,9 +81,6 @@ export function reconcileWorkspaceArtifacts( db: AppDatabase, ctx: ArtifactProjectorRunContext, ): ProjectionOutcome[] { - if (ctx.workspaceMode !== "exclusive") { - return [{ kind: "skipped", reason: "workspace_not_exclusive" }]; - } return projectCollectedChanges(db, ctx, collectChangedPaths(ctx)); } @@ -91,9 +88,6 @@ function projectBashArtifacts( db: AppDatabase, ctx: ArtifactProjectorRunContext, ): ProjectionOutcome[] { - if (ctx.workspaceMode !== "exclusive") { - return [{ kind: "skipped", reason: "workspace_not_exclusive" }]; - } return projectCollectedChanges(db, ctx, collectChangedPaths(ctx)); } diff --git a/src/main/agent-runtime/artifact-work-products.ts b/src/main/agent-runtime/artifact-work-products.ts index 804760b0..3dd8577f 100644 --- a/src/main/agent-runtime/artifact-work-products.ts +++ b/src/main/agent-runtime/artifact-work-products.ts @@ -51,9 +51,14 @@ export function retireDeletedArtifacts( deleted: string[], ): ProjectionOutcome[] { if (deleted.length === 0) return []; + const resolvedWorkspaceId = + ctx.workspaceId ?? + (ctx.workspacePath + ? db.core.workspaces.findWorkspaceIdByLocalPath(ctx.companyId, ctx.workspacePath) + : null); const archived = db.core.documents.archiveWorkProductsByExternalPaths({ companyId: ctx.companyId, - workspaceId: ctx.workspaceId, + workspaceId: resolvedWorkspaceId, provider: "filesystem", paths: deleted, lastTouchedByRunId: ctx.runId, diff --git a/src/main/agent-runtime/hook-handlers.ts b/src/main/agent-runtime/hook-handlers.ts index ddcc4eeb..f02aae75 100644 --- a/src/main/agent-runtime/hook-handlers.ts +++ b/src/main/agent-runtime/hook-handlers.ts @@ -23,6 +23,10 @@ import { projectToolExecutionToWorkProduct, type ArtifactProjectorRunContext, } from "./artifact-projector"; +import { + appendArtifactProjectionEvent, + appendArtifactProjectionFailureEvent, +} from "./artifact-projection-events"; import { isCofounderMandatoryApproval, isOperationalApproval, @@ -144,12 +148,10 @@ function checkOperationalApprovalBlock( * through this hook; as further projections (browser actions, social * posts) migrate off the event-observer path they plug in here. * - * Projection is treated as a primary persistence action: failures are - * logged at ERROR so they stay visible in the desktop log tail, and the - * `run-finished` reconciliation sweep (`RuntimeBridge`) revisits the - * workspace to register anything the hook missed. Silent swallow was - * the failure mode that left `work_products` empty while the workspace - * accumulated files the UI never surfaced. + * Projection is treated as a primary persistence action: the filesystem + * mutation and its run-event acknowledgement commit together. A failure + * returns an RPC error to the worker, which lets pi surface the hook + * failure as a deterministic tool error. */ export function handleAfterToolCall( deps: HookHandlerDeps, @@ -168,16 +170,30 @@ export function handleAfterToolCall( workspaceBaseline: ctx.workspaceBaseline, }; try { - const outcomes = projectToolExecutionToWorkProduct( - deps.db, - runCtx, - { + const outcomes = deps.db.withTransaction(() => { + const projectionOutcomes = projectToolExecutionToWorkProduct( + deps.db, + runCtx, + { + toolName: payload.toolName, + args: payload.args, + result: payload.resultContent, + isError: payload.isError, + }, + ); + appendArtifactProjectionEvent(deps.db, { + companyId: ctx.companyId, + runId: ctx.runId, + agentId: ctx.agentId, + phase: "after_tool_call", + workspaceId: ctx.workspaceId, + workspaceMode: ctx.workspaceMode, toolName: payload.toolName, - args: payload.args, - result: payload.resultContent, - isError: payload.isError, - }, - ); + toolCallId: payload.toolCallId, + outcomes: projectionOutcomes, + }); + return projectionOutcomes; + }); const registered = outcomes.filter((o) => o.kind === "inserted" || o.kind === "updated").length; const retired = outcomes.filter((o) => o.kind === "retired").length; if (registered > 0 || retired > 0) { @@ -186,9 +202,28 @@ export function handleAfterToolCall( ); } } catch (error) { + const errorMessage = error instanceof Error ? error.message : String(error); + try { + appendArtifactProjectionFailureEvent(deps.db, { + companyId: ctx.companyId, + runId: ctx.runId, + agentId: ctx.agentId, + phase: "after_tool_call", + workspaceId: ctx.workspaceId, + workspaceMode: ctx.workspaceMode, + toolName: payload.toolName, + toolCallId: payload.toolCallId, + errorMessage, + }); + } catch (ackError) { + deps.logger.error( + `[hook-handlers] artifact projection failure ack failed for run ${ctx.runId} tool ${payload.toolName}: ${ackError instanceof Error ? ackError.message : String(ackError)}`, + ); + } deps.logger.error( - `[hook-handlers] artifact projection threw for run ${ctx.runId} tool ${payload.toolName}: ${error instanceof Error ? error.message : String(error)}`, + `[hook-handlers] artifact projection threw for run ${ctx.runId} tool ${payload.toolName}: ${errorMessage}`, ); + throw error; } return NO_OP_AFTER; } diff --git a/src/main/agent-runtime/runtime-artifact-reconciler.ts b/src/main/agent-runtime/runtime-artifact-reconciler.ts new file mode 100644 index 00000000..26872e53 --- /dev/null +++ b/src/main/agent-runtime/runtime-artifact-reconciler.ts @@ -0,0 +1,79 @@ +import type { AppDatabase } from "@main/database"; +import type { ActiveRunContext } from "./hook-handlers"; +import { reconcileWorkspaceArtifacts } from "./artifact-projector"; +import { + appendArtifactProjectionEvent, + appendArtifactProjectionFailureEvent, +} from "./artifact-projection-events"; + +export type RunResourceReleaseResult = + | { ok: true } + | { ok: false; errorMessage: string }; + +export interface RuntimeArtifactReconcilerDeps { + db: AppDatabase; + logger: { + info: (...args: unknown[]) => void; + error: (...args: unknown[]) => void; + }; +} + +export function reconcileArtifactsForActiveRun( + deps: RuntimeArtifactReconcilerDeps, + ctx: ActiveRunContext, +): RunResourceReleaseResult { + try { + const outcomes = deps.db.withTransaction(() => { + const projectionOutcomes = reconcileWorkspaceArtifacts(deps.db, { + runId: ctx.runId, + agentId: ctx.agentId, + companyId: ctx.companyId, + taskId: ctx.taskId, + workspaceId: ctx.workspaceId, + workspacePath: ctx.workspacePath, + workspaceMode: ctx.workspaceMode, + logger: deps.logger, + workspaceBaseline: ctx.workspaceBaseline, + }); + appendArtifactProjectionEvent(deps.db, { + companyId: ctx.companyId, + runId: ctx.runId, + agentId: ctx.agentId, + phase: "completion_reconcile", + workspaceId: ctx.workspaceId, + workspaceMode: ctx.workspaceMode, + outcomes: projectionOutcomes, + }); + return projectionOutcomes; + }); + const inserted = outcomes.filter((o) => o.kind === "inserted").length; + const retired = outcomes.filter((o) => o.kind === "retired").length; + if (inserted > 0 || retired > 0) { + deps.logger.info( + `[runtime-bridge] reconciliation for run ${ctx.runId}: ${inserted} registered, ${retired} retired`, + ); + } + return { ok: true }; + } catch (error) { + const errorMessage = error instanceof Error ? error.message : String(error); + try { + appendArtifactProjectionFailureEvent(deps.db, { + companyId: ctx.companyId, + runId: ctx.runId, + agentId: ctx.agentId, + phase: "completion_reconcile", + workspaceId: ctx.workspaceId, + workspaceMode: ctx.workspaceMode, + errorMessage, + }); + } catch (ackError) { + deps.logger.error( + `[runtime-bridge] artifact reconciliation failure ack failed for run ${ctx.runId}: ${ackError instanceof Error ? ackError.message : String(ackError)}`, + ); + } + deps.logger.error( + `[runtime-bridge] artifact reconciliation failed for run ${ctx.runId}: ${errorMessage}`, + ); + return { ok: false, errorMessage }; + } +} diff --git a/src/main/agent-runtime/runtime-bridge.ts b/src/main/agent-runtime/runtime-bridge.ts index 51425ab8..87a4487c 100644 --- a/src/main/agent-runtime/runtime-bridge.ts +++ b/src/main/agent-runtime/runtime-bridge.ts @@ -8,7 +8,11 @@ import { getConnectorDefinition } from "@main/connectors"; import { projectInboxMessageToPi } from "./inbox-projection"; import { projectAgentEventToLogChunk } from "./agent-event-projection"; import { hasAttributableAssistantUsage } from "../run-usage-status"; -import { captureBaselineIfRequired, reconcileWorkspaceArtifacts } from "./artifact-projector"; +import { captureBaselineIfRequired } from "./artifact-projector"; +import { + reconcileArtifactsForActiveRun, + type RunResourceReleaseResult, +} from "./runtime-artifact-reconciler"; import { handleBeforeToolCall, handleAfterToolCall, @@ -134,40 +138,21 @@ export class RuntimeBridge { * when individual projections skipped (bash against a non-git repo, * path-ignore false positives, transient `git status` failures). */ - releaseRunResources(runId: string): void { + releaseRunResources(runId: string): RunResourceReleaseResult { const ctx = this.activeRuns.get(runId); - if (ctx) this.reconcileArtifactsForRun(ctx); - this.activeRuns.delete(runId); - this.inboundCorrelations.deleteRun(runId); - if (ctx) this.sessionStores.delete(ctx.taskSessionId); + let result: RunResourceReleaseResult = { ok: true }; + try { + if (ctx) result = this.reconcileArtifactsForRun(ctx); + return result; + } finally { + this.activeRuns.delete(runId); + this.inboundCorrelations.deleteRun(runId); + if (ctx) this.sessionStores.delete(ctx.taskSessionId); + } } - private reconcileArtifactsForRun(ctx: ActiveRunContext): void { - if (!ctx.workspacePath) return; - try { - const outcomes = reconcileWorkspaceArtifacts(this.deps.db, { - runId: ctx.runId, - agentId: ctx.agentId, - companyId: ctx.companyId, - taskId: ctx.taskId, - workspaceId: ctx.workspaceId, - workspacePath: ctx.workspacePath, - workspaceMode: ctx.workspaceMode, - logger: this.deps.logger, - workspaceBaseline: ctx.workspaceBaseline, - }); - const inserted = outcomes.filter((o) => o.kind === "inserted").length; - const retired = outcomes.filter((o) => o.kind === "retired").length; - if (inserted > 0 || retired > 0) { - this.deps.logger.info( - `[runtime-bridge] reconciliation for run ${ctx.runId}: ${inserted} registered, ${retired} retired`, - ); - } - } catch (error) { - this.deps.logger.error( - `[runtime-bridge] artifact reconciliation failed for run ${ctx.runId}: ${error instanceof Error ? error.message : String(error)}`, - ); - } + private reconcileArtifactsForRun(ctx: ActiveRunContext): RunResourceReleaseResult { + return reconcileArtifactsForActiveRun(this.deps, ctx); } /** @@ -397,11 +382,7 @@ export class RuntimeBridge { workspacePath: string | null; workspaceMode: "exclusive" | "shared" | "none"; }; - // Baseline scans are safe only while this run owns the workspace lock. - const baseline = - payload.workspaceMode === "exclusive" - ? captureBaselineIfRequired(payload.workspacePath, this.deps.logger) - : null; + const baseline = captureBaselineIfRequired(payload.workspacePath, this.deps.logger); this.registerActiveRun({ runId, agentId: payload.agentId, diff --git a/src/main/database/runs.ts b/src/main/database/runs.ts index 086ce293..53d53e4b 100644 --- a/src/main/database/runs.ts +++ b/src/main/database/runs.ts @@ -258,6 +258,17 @@ export function buildRunMethods( return { budgetHardStopped }; }, + markRunCompletionFailure(runId: string, errorMessage: string): void { + const ts = nowIso(); + transaction(() => { + getDb() + .prepare( + "update runs set status = 'failed', summary = ?, error_message = ?, usage_status = 'error_stop', finished_at = coalesce(finished_at, ?), updated_at = ? where id = ?", + ) + .run("Run completion blocked by artifact projection failure.", errorMessage, ts, ts, runId); + }); + }, + cancelRunForDeletion(runId: string, taskId: string | null, workspaceId: string | null, summary: string) { const ts = nowIso(); getDb() diff --git a/src/main/worker/entry/message-handler.ts b/src/main/worker/entry/message-handler.ts index f5612cc6..48ad061c 100644 --- a/src/main/worker/entry/message-handler.ts +++ b/src/main/worker/entry/message-handler.ts @@ -7,6 +7,7 @@ import { releaseCompletedRunWorkspaceLock } from "../handlers/run-finished-helpe import { handleContextCompaction } from "../handlers/context-compaction-handler"; import type { HandlerContext, HandlerDeps } from "../handlers/handler-context"; import type { RunFinishedContext } from "../handlers/run-finished-helpers"; +import type { RunResourceReleaseResult } from "../../agent-runtime/runtime-artifact-reconciler"; export async function handleWorkerMessage( message: unknown, @@ -50,10 +51,21 @@ export async function handleWorkerMessage( return; case "run-finished": { let completion: RunFinishedContext | null = null; + let runtimeResourcesReleased = false; + let runtimeReleaseResult: RunResourceReleaseResult = { ok: true }; + const releaseRuntimeResources = (): RunResourceReleaseResult => { + if (runtimeResourcesReleased) return runtimeReleaseResult; + runtimeResourcesReleased = true; + runtimeReleaseResult = + deps.getRuntimeBridge?.()?.releaseRunResources(event.runId) ?? { ok: true }; + return runtimeReleaseResult; + }; try { - completion = handleRunFinished(event, deps, context); + completion = handleRunFinished(event, deps, context, { + beforePostCompletionAutomation: releaseRuntimeResources, + }); } finally { - deps.getRuntimeBridge?.()?.releaseRunResources(event.runId); + releaseRuntimeResources(); } releaseCompletedRunWorkspaceLock(completion, deps); if (!worker) { diff --git a/src/main/worker/handlers/run-finished-handler.ts b/src/main/worker/handlers/run-finished-handler.ts index 97f44a38..9ccd2707 100644 --- a/src/main/worker/handlers/run-finished-handler.ts +++ b/src/main/worker/handlers/run-finished-handler.ts @@ -8,17 +8,39 @@ import { import type { HandlerContext, HandlerDeps, WorkerMessageEvent } from "./handler-context"; import type { RunFinishedContext } from "./run-finished-helpers"; +export interface RunFinishedHooks { + beforePostCompletionAutomation?: (context: RunFinishedContext) => + | void + | { ok: true } + | { ok: false; errorMessage: string }; +} + export function handleRunFinished( event: Extract, deps: HandlerDeps, context: HandlerContext, + hooks: RunFinishedHooks = {}, ): RunFinishedContext | null { const rfCtx = finalizeRun(event, deps, context); if (!rfCtx) return null; accumulateAndRotateSession(event, deps, rfCtx); - emitAndNotifyCompletion(event, deps, rfCtx); - const { hadDeferredWakes } = runPostCompletionAutomation(event, deps, context, rfCtx); - triggerSelfWake(event, deps, context, rfCtx, hadDeferredWakes); + const boundaryResult = hooks.beforePostCompletionAutomation?.(rfCtx); + const effectiveEvent = + boundaryResult && !boundaryResult.ok + ? { + ...event, + status: "failed" as const, + summary: null, + errorMessage: boundaryResult.errorMessage, + usageStatus: "error_stop" as const, + } + : event; + if (boundaryResult && !boundaryResult.ok) { + deps.db.core.runs.markRunCompletionFailure(event.runId, boundaryResult.errorMessage); + } + emitAndNotifyCompletion(effectiveEvent, deps, rfCtx); + const { hadDeferredWakes } = runPostCompletionAutomation(effectiveEvent, deps, context, rfCtx); + triggerSelfWake(effectiveEvent, deps, context, rfCtx, hadDeferredWakes); return rfCtx; } diff --git a/tests/integration/artifact-projection-ack.spec.ts b/tests/integration/artifact-projection-ack.spec.ts new file mode 100644 index 00000000..b48c369e --- /dev/null +++ b/tests/integration/artifact-projection-ack.spec.ts @@ -0,0 +1,177 @@ +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { RuntimeBridge } from "../../src/main/agent-runtime"; +import { AppDatabase } from "../../src/main/database"; +import type { CredentialResolver } from "../../src/main/credentials"; + +const tempDirs: string[] = []; + +async function createTempDir(prefix: string): Promise { + const dir = await mkdtemp(join(tmpdir(), prefix)); + tempDirs.push(dir); + return dir; +} + +afterEach(async () => { + await Promise.all(tempDirs.splice(0).map((dir) => rm(dir, { recursive: true, force: true }))); +}); + +async function createHarness() { + const profileDir = await createTempDir("agentcompany-artifact-ack-"); + const workspacePath = await createTempDir("agentcompany-artifact-workspace-"); + const db = new AppDatabase(profileDir); + db.init(); + const companyId = db.core.company.saveCompany({ + name: "Artifact Ack Co", + description: "Durable artifact projection test", + status: "active", + }); + const workspaceId = db.core.workspaces.saveWorkspace({ + companyId, + name: "Primary", + localPath: workspacePath, + repoUrl: "", + repoRef: "", + isPrimary: true, + }); + const agentId = db.core.agents.saveAgent({ + companyId, + name: "Artifact Agent", + role: "Engineer", + title: "Engineer", + status: "active", + reportsTo: null, + connectorId: "openai-codex", + workspaceId, + model: null, + capabilities: "Writes files", + budgetMonthlyUsd: 100, + }); + const taskId = db.core.tasks.saveTask({ + companyId, + title: "Produce artifact", + description: "", + assigneeAgentId: agentId, + workspaceId, + priority: "medium", + status: "in_progress", + }); + const run = db.core.runs.createRun({ + companyId, + taskId, + agentId, + workspaceId: null, + connectorId: "openai-codex", + }); + const bridge = new RuntimeBridge({ + db, + emitEvent: vi.fn(), + logger: { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }, + credentialResolver: { + resolve: vi.fn(async () => ({ apiKey: "test-api-key", extraEnv: {} })), + } as unknown as CredentialResolver, + }); + + const prepare = await bridge.handleRpc({ + rpcId: "prepare", + method: "prepareRun", + runId: run.id, + payload: { + connectorId: "openai-codex", + taskSessionId: "task-session-artifact-ack", + agentId, + companyId, + taskId, + modelId: "gpt-5.3-codex", + workspaceId: null, + workspacePath, + workspaceMode: "none", + }, + }); + expect(prepare.ok).toBe(true); + + return { + db, + bridge, + companyId, + workspacePath, + agentId, + taskId, + runId: run.id, + }; +} + +describe("artifact projection durable ack", () => { + it("persists afterToolCall projection outcomes to the run event tail", async () => { + const harness = await createHarness(); + try { + await mkdir(join(harness.workspacePath, "reports"), { recursive: true }); + await writeFile(join(harness.workspacePath, "reports", "hook.md"), "hook artifact", "utf8"); + + const response = await harness.bridge.handleRpc({ + rpcId: "after-tool", + method: "afterToolCall", + runId: harness.runId, + payload: { + toolName: "write", + toolCallId: "tool-call-hook", + args: { path: "reports/hook.md" }, + resultContent: [], + resultDetails: null, + isError: false, + }, + }); + + expect(response.ok).toBe(true); + const event = harness.db.core.runs.listRunEvents(harness.runId, harness.companyId)[0]; + expect(event).toMatchObject({ + eventType: "artifact_projection", + level: "info", + }); + expect(event?.payload).toMatchObject({ + phase: "after_tool_call", + toolName: "write", + toolCallId: "tool-call-hook", + workspaceMode: "none", + counts: { inserted: 1, updated: 0, retired: 0, skipped: 0 }, + }); + } finally { + harness.db.close(); + } + }); + + it("runs completion reconciliation before resource release for non-exclusive observation workspaces", async () => { + const harness = await createHarness(); + try { + await mkdir(join(harness.workspacePath, "reports"), { recursive: true }); + await writeFile(join(harness.workspacePath, "reports", "completion.md"), "completion artifact", "utf8"); + + harness.bridge.releaseRunResources(harness.runId); + + expect( + harness.db.projection.snapshot.listSnapshot().workProducts.some( + (entry) => entry.externalId === "reports/completion.md", + ), + ).toBe(true); + const events = harness.db.core.runs.listRunEvents(harness.runId, harness.companyId); + expect(events).toHaveLength(1); + expect(events[0]).toMatchObject({ + eventType: "artifact_projection", + level: "info", + }); + expect(events[0]?.payload).toMatchObject({ + phase: "completion_reconcile", + workspaceMode: "none", + counts: { inserted: 1, updated: 0, retired: 0, skipped: 0 }, + }); + } finally { + harness.db.close(); + } + }); +}); diff --git a/tests/integration/work-products-lifecycle.spec.ts b/tests/integration/work-products-lifecycle.spec.ts index 758fa6a9..d2c44b92 100644 --- a/tests/integration/work-products-lifecycle.spec.ts +++ b/tests/integration/work-products-lifecycle.spec.ts @@ -5,6 +5,7 @@ import { tmpdir } from "node:os"; import { afterEach, describe, expect, it } from "vitest"; import { AppDatabase } from "../../src/main/database"; import { + projectToolExecutionToWorkProduct, reconcileWorkspaceArtifacts, type ArtifactProjectorRunContext, } from "../../src/main/agent-runtime/artifact-projector"; @@ -236,6 +237,85 @@ describe("work_products archived visibility", () => { }); describe("work_products bulk retirement", () => { + it("registers bash artifacts from non-exclusive observation workspaces", async () => { + const harness = await createHarness(); + try { + const ctx: ArtifactProjectorRunContext = { + runId: harness.runId, + agentId: harness.agentId, + companyId: harness.companyId, + taskId: harness.taskId, + workspaceId: null, + workspacePath: harness.workspacePath, + workspaceMode: "none", + workspaceBaseline: { + mtimeByRelPath: new Map(), + capturedAt: Date.now(), + truncated: false, + }, + }; + await mkdir(join(harness.workspacePath, "reports"), { recursive: true }); + await writeFile(join(harness.workspacePath, "reports", "non-exclusive.md"), "observed", "utf8"); + + const outcomes = projectToolExecutionToWorkProduct(harness.db, ctx, { + toolName: "bash", + args: { command: "printf observed > reports/non-exclusive.md" }, + result: null, + isError: false, + }); + + expect(outcomes).toEqual([ + expect.objectContaining({ + kind: "inserted", + relativePath: "reports/non-exclusive.md", + }), + ]); + expect( + harness.db.projection.snapshot.listSnapshot().workProducts.some( + (entry) => entry.externalId === "reports/non-exclusive.md", + ), + ).toBe(true); + } finally { + harness.db.close(); + } + }); + + it("retires deleted artifacts from non-exclusive observation workspaces", async () => { + const harness = await createHarness(); + try { + const relPath = "reports/non-exclusive-retired.md"; + const workProductId = registerArtifact(harness, relPath); + await unlink(join(harness.workspacePath, relPath)); + + const outcomes = reconcileWorkspaceArtifacts(harness.db, { + runId: harness.runId, + agentId: harness.agentId, + companyId: harness.companyId, + taskId: harness.taskId, + workspaceId: null, + workspacePath: harness.workspacePath, + workspaceMode: "none", + workspaceBaseline: { + mtimeByRelPath: new Map([[relPath, Date.now()]]), + capturedAt: Date.now(), + truncated: false, + }, + }); + + expect(outcomes).toEqual([ + expect.objectContaining({ + kind: "retired", + relativePath: relPath, + }), + ]); + expect( + harness.db.core.documents.resolveWorkProductTarget(workProductId, harness.companyId), + ).toMatchObject({ kind: "error", reason: "archived" }); + } finally { + harness.db.close(); + } + }); + it("archives every deletion past the previous 50-path budget", async () => { const harness = await createHarness(); try { diff --git a/tests/unit/agent-hooks.test.ts b/tests/unit/agent-hooks.test.ts new file mode 100644 index 00000000..1eb2fb69 --- /dev/null +++ b/tests/unit/agent-hooks.test.ts @@ -0,0 +1,28 @@ +import { describe, expect, it, vi } from "vitest"; +import { buildAgentHooks } from "../../src/main/agent-runtime/agent-hooks"; +import type { WorkerRpcClient } from "../../src/main/worker/entry/rpc"; + +function createRpc(response: { ok: true; result: unknown } | { ok: false; error: string }) { + return { + invoke: vi.fn(async () => response), + } as unknown as WorkerRpcClient & { invoke: ReturnType }; +} + +describe("agent hook RPC policy", () => { + it("fails afterToolCall when main cannot acknowledge artifact projection", async () => { + const rpc = createRpc({ ok: false, error: "RuntimeBridge unavailable" }); + const hooks = buildAgentHooks({ + runId: "11111111-1111-4111-8111-111111111111", + rpc, + }); + + await expect( + hooks.afterToolCall({ + toolCall: { name: "write", id: "tool-call-1" }, + args: { path: "report.md" }, + result: { content: [], details: null }, + isError: false, + } as unknown as Parameters[0]), + ).rejects.toThrow("afterToolCall RPC failed: RuntimeBridge unavailable"); + }); +}); diff --git a/tests/unit/artifact-projector-errors.test.ts b/tests/unit/artifact-projector-errors.test.ts index 97102920..80a34482 100644 --- a/tests/unit/artifact-projector-errors.test.ts +++ b/tests/unit/artifact-projector-errors.test.ts @@ -86,7 +86,7 @@ describe("artifact projector error taxonomy", () => { expect(logger.error).not.toHaveBeenCalled(); }); - it("skips broad bash projection when the run has no exclusive workspace lock", async () => { + it("skips broad bash projection when the run has no change source", async () => { const workspacePath = await createTempDir("agentcompany-shared-artifacts-"); const ctx: ArtifactProjectorRunContext = { runId: "run-shared", @@ -105,9 +105,9 @@ describe("artifact projector error taxonomy", () => { args: { command: "touch report.md" }, result: null, isError: false, - })).toEqual([{ kind: "skipped", reason: "workspace_not_exclusive" }]); + })).toEqual([{ kind: "skipped", reason: "no_baseline" }]); expect(reconcileWorkspaceArtifacts({} as AppDatabase, ctx)).toEqual([ - { kind: "skipped", reason: "workspace_not_exclusive" }, + { kind: "skipped", reason: "no_baseline" }, ]); }); }); diff --git a/tests/unit/worker-completion-boundary.test.ts b/tests/unit/worker-completion-boundary.test.ts index 043e8f38..9b57ae9b 100644 --- a/tests/unit/worker-completion-boundary.test.ts +++ b/tests/unit/worker-completion-boundary.test.ts @@ -3,6 +3,7 @@ import { describe, it, expect, vi, afterEach } from "vitest"; import { handleWorkerMessage } from "@main/worker/entry/message-handler"; import { workerMessageSchema } from "@main/worker/entry/types"; import type { RunJob } from "@main/worker/runtime/helpers"; +import type { RunResourceReleaseResult } from "@main/agent-runtime/runtime-artifact-reconciler"; const RUN_ID = "00000000-0000-4000-8000-000000000001"; const NEXT_RUN_ID = "00000000-0000-4000-8000-000000000002"; @@ -31,7 +32,11 @@ function createHandlerDeps(calls: string[]) { }); const rotateSession = vi.fn(() => calls.push("rotateSession")); const processRunCompletion = vi.fn(() => calls.push("processRunCompletion")); - const releaseRunResources = vi.fn(() => calls.push("releaseRunResources")); + const releaseRunResources = vi.fn((): RunResourceReleaseResult => { + calls.push("releaseRunResources"); + return { ok: true as const }; + }); + const markRunCompletionFailure = vi.fn(() => calls.push("markRunCompletionFailure")); const db = { core: { @@ -44,6 +49,7 @@ function createHandlerDeps(calls: string[]) { workspaceId: "workspace-1", })), finishRun, + markRunCompletionFailure, }, workspaces: { releaseWorkspaceLock }, activity: { addActivity: vi.fn(() => calls.push("addActivity")) }, @@ -104,6 +110,7 @@ function createHandlerDeps(calls: string[]) { rotateSession, processRunCompletion, releaseRunResources, + markRunCompletionFailure, }; } @@ -137,7 +144,7 @@ describe("worker completion boundary", () => { Reflect.deleteProperty(process, "parentPort"); }); - it("releases the workspace lock after session, automation, and runtime resource release", async () => { + it("runs artifact reconciliation before automation and releases the workspace lock last", async () => { const calls: string[] = []; const { deps, @@ -167,6 +174,9 @@ describe("worker completion boundary", () => { expect(releaseWorkspaceLock.mock.invocationCallOrder[0]).toBeGreaterThan( rotateSession.mock.invocationCallOrder[0]!, ); + expect(processRunCompletion.mock.invocationCallOrder[0]).toBeGreaterThan( + releaseRunResources.mock.invocationCallOrder[0]!, + ); expect(releaseWorkspaceLock.mock.invocationCallOrder[0]).toBeGreaterThan( processRunCompletion.mock.invocationCallOrder[0]!, ); @@ -175,6 +185,48 @@ describe("worker completion boundary", () => { ); }); + it("marks run completion failed and blocks success automation when artifact reconciliation fails", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + processRunCompletion, + releaseRunResources, + markRunCompletionFailure, + } = createHandlerDeps(calls); + releaseRunResources.mockImplementationOnce(() => { + calls.push("releaseRunResources"); + return { ok: false as const, errorMessage: "completion_reconcile failed" }; + }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(markRunCompletionFailure).toHaveBeenCalledWith(RUN_ID, "completion_reconcile failed"); + expect(processRunCompletion).toHaveBeenCalledWith( + expect.objectContaining({ id: RUN_ID }), + expect.objectContaining({ + status: "failed", + errorMessage: "completion_reconcile failed", + usageStatus: "error_stop", + }), + ); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + it("keeps same-workspace runs queued until parent acknowledges completion", async () => { const parentPort = new EventEmitter() as EventEmitter & { postMessage: ReturnType;