fix: make artifact projection durable

This commit is contained in:
ZacharyZhang-NY committed 2026-04-23 21:50:49 -04:00
1 parent 84c00e0011
commit 88c905c381
15 files changed
+654 -76

No files matched your search

+3 -6
View File
@@ -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 = {};
@@ -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,
},
});
}
@@ -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));
}
@@ -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,
+51 -16
View File
@@ -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;
}
@@ -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 };
}
}
+18 -37
View File
@@ -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,
+11
View File
@@ -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()
+14 -2
View File
@@ -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) {
@@ -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<WorkerMessageEvent, { type: "run-finished" }>,
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;
}