refactor(artifacts): reconcile workspace artifacts at completion

This commit is contained in:
ZacharyZhang-NY committed 2026-04-24 00:58:07 -04:00
1 parent 2ddc67cc3d
commit fd86fc6668
17 files changed
+114 -436

No files matched your search

+8 -36
View File
@@ -1,11 +1,11 @@
/**
* Aggregated factory for the three pi-agent-core tool-execution hooks
* Aggregated factory for the pi-agent-core runtime hooks
* installed by `pi-agent-factory.createPiAgent`.
*
* Before this module the factory wired `beforeToolCall` / `afterToolCall`
* / `transformContext` as three inline closures, each re-establishing
* Before this module the factory wired `beforeToolCall` and
* `transformContext` as inline closures, each re-establishing
* its own RPC envelope shape and its own error policy. A new hook (or
* a signature change in pi) meant touching the factory in three places,
* a signature change in pi) meant touching the factory in multiple places,
* and the "what does each hook do on failure?" matrix lived only in
* reader memory.
*
@@ -17,20 +17,17 @@
* so the full transcript never crosses the worker-main boundary.
*
* The signatures below are pinned to pi via `pi-hook-contract`: any
* drift in `AgentOptions.beforeToolCall` / `afterToolCall` /
* `transformContext` breaks the build on the contract file, so the
* drift in `AgentOptions.beforeToolCall` / `transformContext` breaks
* the build on the contract file, so the
* aggregation cannot silently drift from upstream.
*/
import type {
AfterToolCallResult,
AgentMessage,
AgentOptions,
BeforeToolCallResult,
} from "@mariozechner/pi-agent-core";
import type {
AfterToolCallRpcPayload,
AfterToolCallRpcResult,
BeforeToolCallRpcPayload,
BeforeToolCallRpcResult,
} from "./hook-rpc";
@@ -65,12 +62,11 @@ export interface AgentHooksDeps {
*/
export interface AgentHookBundle {
beforeToolCall: NonNullable<AgentOptions["beforeToolCall"]>;
afterToolCall: NonNullable<AgentOptions["afterToolCall"]>;
transformContext: NonNullable<AgentOptions["transformContext"]>;
}
/**
* Build the three pi-agent-core hooks as a single bundle.
* Build the pi-agent-core hooks as a single bundle.
*
* Each hook's behavior matches the pre-aggregation inline closures in
* `pi-agent-factory`:
@@ -78,8 +74,6 @@ 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-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.
*/
@@ -107,28 +101,6 @@ export function buildAgentHooks(deps: AgentHooksDeps): AgentHookBundle {
return { block: true, reason: result.reason ?? undefined };
};
const afterToolCall: AgentHookBundle["afterToolCall"] = async (
ctx,
): Promise<AfterToolCallResult | undefined> => {
const payload: AfterToolCallRpcPayload = {
toolName: ctx.toolCall.name,
toolCallId: ctx.toolCall.id,
args: ctx.args,
resultContent: ctx.result.content,
resultDetails: ctx.result.details,
isError: ctx.isError,
};
const response = await rpc.invoke(runId, "afterToolCall", payload);
if (!response.ok) {
throw new Error(`afterToolCall RPC failed: ${response.error}`);
}
const result = response.result as AfterToolCallRpcResult;
const override: AfterToolCallResult = {};
if (result.contentOverride) override.content = result.contentOverride;
if (result.isErrorOverride !== null) override.isError = result.isErrorOverride;
return Object.keys(override).length === 0 ? undefined : override;
};
const transformContext: AgentHookBundle["transformContext"] = async (
messages,
): Promise<AgentMessage[]> => {
@@ -152,5 +124,5 @@ export function buildAgentHooks(deps: AgentHooksDeps): AgentHookBundle {
}
};
return { beforeToolCall, afterToolCall, transformContext };
return { beforeToolCall, transformContext };
}
@@ -1,7 +1,7 @@
import type { AppDatabase } from "@main/database";
import type { ProjectionOutcome } from "./artifact-types";
export type ArtifactProjectionPhase = "after_tool_call" | "completion_reconcile";
export type ArtifactProjectionPhase = "completion_reconcile";
export interface ArtifactProjectionEventInput {
companyId: string;
@@ -10,8 +10,6 @@ export interface ArtifactProjectionEventInput {
phase: ArtifactProjectionPhase;
workspaceId: string | null;
workspaceMode: "exclusive" | "shared" | "none";
toolName?: string | null;
toolCallId?: string | null;
outcomes: ProjectionOutcome[];
}
@@ -22,8 +20,6 @@ export interface ArtifactProjectionFailureInput {
phase: ArtifactProjectionPhase;
workspaceId: string | null;
workspaceMode: "exclusive" | "shared" | "none";
toolName?: string | null;
toolCallId?: string | null;
errorMessage: string;
}
@@ -73,8 +69,6 @@ export function appendArtifactProjectionEvent(
phase: input.phase,
workspaceId: input.workspaceId,
workspaceMode: input.workspaceMode,
toolName: input.toolName ?? null,
toolCallId: input.toolCallId ?? null,
counts,
outcomes: input.outcomes,
},
@@ -97,8 +91,6 @@ export function appendArtifactProjectionFailureEvent(
phase: input.phase,
workspaceId: input.workspaceId,
workspaceMode: input.workspaceMode,
toolName: input.toolName ?? null,
toolCallId: input.toolCallId ?? null,
errorMessage: input.errorMessage,
},
});
+6 -56
View File
@@ -2,12 +2,10 @@ import type { AppDatabase } from "@main/database";
import { collectChangedPaths } from "./artifact-change-detector";
import {
projectChangedPaths,
projectSinglePathArtifact,
retireDeletedArtifacts,
} from "./artifact-work-products";
import type {
ArtifactProjectorRunContext,
ArtifactProjectorToolEvent,
ChangeSet,
ProjectionOutcome,
} from "./artifact-types";
@@ -15,17 +13,16 @@ import type {
export { captureBaselineIfRequired } from "./artifact-baseline";
export type {
ArtifactProjectorRunContext,
ArtifactProjectorToolEvent,
ProjectionOutcome,
WorkspaceBaseline,
} from "./artifact-types";
/**
* Filesystem artifact protocol for pi tool invocations.
* End-of-run filesystem artifact protocol.
*
* Successful destructive tool calls (`write`, `edit`, `bash`) are
* projected into `work_products`, so the desktop UI can surface what
* the agent produced without relying on the agent to POST documents.
* Completion reconciliation projects changed and deleted workspace
* paths into `work_products`, so the desktop UI can surface what the
* agent produced without relying on the agent to POST documents.
*
* Projection is idempotent through the natural identity
* `(companyId, workspaceId, provider="filesystem", externalId=<workspace-relative-path>)`.
@@ -33,49 +30,9 @@ export type {
* mtime baseline captured by `captureBaselineIfRequired`.
*/
type WriteArgs = { path: unknown; content?: unknown };
type EditArgs = { path: unknown };
/** Normalises pi tool-call args to the file path the agent touched. */
function extractToolPath(toolName: string, args: unknown): string | null {
if (!args || typeof args !== "object") return null;
if (toolName === "write") {
const p = (args as WriteArgs).path;
return typeof p === "string" && p.length > 0 ? p : null;
}
if (toolName === "edit") {
const p = (args as EditArgs).path;
return typeof p === "string" && p.length > 0 ? p : null;
}
return null;
}
/**
* Upsert work_products rows for a successful destructive tool call.
* `write` / `edit` use their explicit path; `bash` scans the workspace
* change set through git status or the captured baseline.
*/
export function projectToolExecutionToWorkProduct(
db: AppDatabase,
ctx: ArtifactProjectorRunContext,
event: ArtifactProjectorToolEvent,
): ProjectionOutcome[] {
if (event.isError) return [{ kind: "skipped", reason: "tool_error" }];
if (!ctx.workspacePath) return [{ kind: "skipped", reason: "no_workspace" }];
if (event.toolName === "write" || event.toolName === "edit") {
return [projectSinglePathArtifact(db, ctx, extractToolPath(event.toolName, event.args))];
}
if (event.toolName === "bash") {
return projectBashArtifacts(db, ctx);
}
return [{ kind: "skipped", reason: "unsupported_tool" }];
}
/**
* End-of-run reconciliation sweep. The per-tool `afterToolCall` hook
* registers artifacts eagerly; the sweep catches files missed by a
* transient hook failure or a bash invocation that touched many paths.
* End-of-run reconciliation sweep. This is the single artifact
* projection boundary for files produced or removed during a run.
*/
export function reconcileWorkspaceArtifacts(
db: AppDatabase,
@@ -84,13 +41,6 @@ export function reconcileWorkspaceArtifacts(
return projectCollectedChanges(db, ctx, collectChangedPaths(ctx));
}
function projectBashArtifacts(
db: AppDatabase,
ctx: ArtifactProjectorRunContext,
): ProjectionOutcome[] {
return projectCollectedChanges(db, ctx, collectChangedPaths(ctx));
}
function projectCollectedChanges(
db: AppDatabase,
ctx: ArtifactProjectorRunContext,
+2 -10
View File
@@ -1,12 +1,5 @@
import type { Logger } from "./artifact-logging";
export type ArtifactProjectorToolEvent = {
toolName: string;
args: unknown;
result: unknown;
isError: boolean;
};
export type ArtifactProjectorRunContext = {
runId: string;
agentId: string;
@@ -18,9 +11,8 @@ export type ArtifactProjectorRunContext = {
logger?: Logger;
/**
* mtime snapshot captured at run start for workspaces that are NOT
* a git repo. Diff-against-baseline is how `projectBashArtifacts`
* and `reconcileWorkspaceArtifacts` find files the agent wrote
* without a `.git` to fall back on.
* a git repo. Diff-against-baseline is how completion reconciliation
* finds files the agent wrote without a `.git` to fall back on.
*/
workspaceBaseline: WorkspaceBaseline | null;
};
@@ -1,27 +1,10 @@
import { statSync } from "node:fs";
import { isAbsolute, relative, resolve } from "node:path";
import { relative, resolve } from "node:path";
import type { AppDatabase } from "@main/database";
import { logArtifactFsError } from "./artifact-logging";
import { isRegistrablePath } from "./artifact-path-filter";
import type { ArtifactProjectorRunContext, ProjectionOutcome } from "./artifact-types";
const MAX_BASH_PROJECTED_FILES = 50;
export function projectSinglePathArtifact(
db: AppDatabase,
ctx: ArtifactProjectorRunContext,
rawPath: string | null,
): ProjectionOutcome {
if (!ctx.workspacePath) return { kind: "skipped", reason: "no_workspace" };
if (!rawPath) return { kind: "skipped", reason: "missing_path" };
const absPath = isAbsolute(rawPath) ? rawPath : resolve(ctx.workspacePath, rawPath);
if (!isRegistrablePath(absPath, ctx.workspacePath)) {
return { kind: "skipped", reason: "path_ignored" };
}
return upsertFilesystemArtifact(db, ctx, ctx.workspacePath, absPath);
}
export function projectChangedPaths(
db: AppDatabase,
ctx: ArtifactProjectorRunContext,
@@ -30,8 +13,7 @@ export function projectChangedPaths(
const workspacePath = ctx.workspacePath;
if (!workspacePath) return [{ kind: "skipped", reason: "no_workspace" }];
const outcomes: ProjectionOutcome[] = [];
const budget = changed.slice(0, MAX_BASH_PROJECTED_FILES);
for (const rel of budget) {
for (const rel of changed) {
const absPath = resolve(workspacePath, rel);
if (!isRegistrablePath(absPath, workspacePath)) {
outcomes.push({ kind: "skipped", reason: "path_ignored" });
@@ -39,9 +21,6 @@ export function projectChangedPaths(
}
outcomes.push(upsertFilesystemArtifact(db, ctx, workspacePath, absPath));
}
if (changed.length > MAX_BASH_PROJECTED_FILES) {
outcomes.push({ kind: "skipped", reason: "bash_scan_cap_exceeded" });
}
return outcomes;
}
+6 -108
View File
@@ -1,32 +1,20 @@
/**
* Main-side implementations of pi's tool-execution hooks. Invoked by
* `RuntimeBridge.handleRpc` when the worker proxies `beforeToolCall`
* or `afterToolCall`.
* Main-side implementation of pi's tool-execution approval hook.
* Invoked by `RuntimeBridge.handleRpc` when the worker proxies
* `beforeToolCall`.
*
* Each handler is deliberately isolated to one concern:
* - `beforeToolCall`: operational-approval gate (blocks destructive
* tools only while an approval that gates EXECUTION is pending)
* - `afterToolCall`: filesystem artifact projection (bash/write/edit
* outputs become `work_products`)
* The handler is deliberately isolated to the operational-approval
* gate: it blocks destructive tools while an approval that gates
* EXECUTION is pending.
*
* Policy lives here; `RuntimeBridge` stays a pure IPC router.
*/
import type { AppDatabase } from "@main/database";
import type {
AfterToolCallRpcPayload,
AfterToolCallRpcResult,
BeforeToolCallRpcPayload,
BeforeToolCallRpcResult,
} from "./hook-rpc";
import {
projectToolExecutionToWorkProduct,
type ArtifactProjectorRunContext,
} from "./artifact-projector";
import {
appendArtifactProjectionEvent,
appendArtifactProjectionFailureEvent,
} from "./artifact-projection-events";
/**
* Lifecycle phase of an active pi run as seen by main.
@@ -70,11 +58,6 @@ export interface HookHandlerDeps {
};
}
const NO_OP_AFTER: AfterToolCallRpcResult = {
contentOverride: null,
isErrorOverride: null,
};
/**
* Tools that can mutate the workspace or execute host-level effects.
* Every other tool (read/grep/glob) is always permitted — gating them
@@ -132,88 +115,3 @@ function checkOperationalApprovalBlock(
};
}
}
/**
* Post-execution projection. Today only filesystem artifacts flow
* 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: 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,
ctx: ActiveRunContext,
payload: AfterToolCallRpcPayload,
): AfterToolCallRpcResult {
const runCtx: ArtifactProjectorRunContext = {
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,
};
try {
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,
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) {
deps.logger.info(
`[hook-handlers] projection for run ${ctx.runId} via ${payload.toolName}: ${registered} registered, ${retired} retired`,
);
}
} 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}: ${errorMessage}`,
);
throw error;
}
return NO_OP_AFTER;
}
+8 -32
View File
@@ -1,17 +1,16 @@
/**
* Hook RPC contracts — the parent↔worker half of pi-agent-core's
* `beforeToolCall` / `afterToolCall` hooks.
* `beforeToolCall` hook.
*
* The Agent instance lives in the utility process. Policy (approval,
* filesystem projection) lives in main. The hooks are therefore proxied:
* the worker invokes an RPC that main answers from the same
* `RuntimeBridge` that already owns agent-event and `persistMessage`
* handling. Context compaction runs in the worker process and reports
* telemetry through a worker message.
* The Agent instance lives in the utility process. Approval policy
* lives in main. The hook is therefore proxied: the worker invokes an
* RPC that main answers from the same `RuntimeBridge` that already owns
* agent-event and `persistMessage` handling. Context compaction runs in
* the worker process and reports telemetry through a worker message.
*
* Payloads intentionally exclude pi's `AgentContext` (systemPrompt,
* messages, tools) because main policy never needs it. Stripping it
* keeps each hook IPC well under the structured-clone budget even for
* messages, tools) because main policy only needs the tool call. This
* keeps the hook IPC well under the structured-clone budget even for
* large transcripts.
*/
// ─── beforeToolCall ───────────────────────────────────────────────────────
@@ -26,26 +25,3 @@ export interface BeforeToolCallRpcResult {
block: boolean;
reason: string | null;
}
// ─── afterToolCall ────────────────────────────────────────────────────────
export interface AfterToolCallRpcPayload {
toolName: string;
toolCallId: string;
args: unknown;
/** Raw tool result content list as emitted by pi's tool executor. */
resultContent: unknown;
/** Structured `details` payload if the tool exposed one. */
resultDetails: unknown;
isError: boolean;
}
/**
* Optional override for the executed tool result, mirroring pi's
* `AfterToolCallResult` shape. Main returns every field as `null` when
* the hook performs observation only (current projection use-case).
*/
export interface AfterToolCallRpcResult {
contentOverride: { type: "text"; text: string }[] | null;
isErrorOverride: boolean | null;
}
+7 -27
View File
@@ -15,14 +15,10 @@ import {
} from "./runtime-artifact-reconciler";
import {
handleBeforeToolCall,
handleAfterToolCall,
type ActiveRunContext,
type ActiveRunLifecycleState,
} from "./hook-handlers";
import type {
AfterToolCallRpcPayload,
BeforeToolCallRpcPayload,
} from "./hook-rpc";
import type { BeforeToolCallRpcPayload } from "./hook-rpc";
import { routeInternalToolCall } from "./internal-tools/router";
import type { InvokeInternalToolRpcPayload } from "@main/worker/runtime/internal-tools/internal-tool-rpc";
import { InboundCorrelationRegistryStore } from "./inbound-correlation-registry";
@@ -81,9 +77,8 @@ type RpcMethod = WorkerRpcMethod;
* - Forwards `agent-event` from the worker to the renderer (push channel).
* - Persists each `message_end` message via `SqliteSessionStore`.
* - Calls `CostWriter` for assistant messages with usage.
* - Proxies pi's tool-execution hooks (`beforeToolCall` / `afterToolCall`)
* into policy handlers: approval-gate checks and filesystem artifact
* projection.
* - Proxies pi's `beforeToolCall` hook into the approval-gate policy handler.
* - Reconciles filesystem artifacts during run resource release.
*
* Policy itself lives in `hook-handlers.ts`; this class is a pure IPC
* router plus an `ActiveRunContext` cache.
@@ -132,11 +127,10 @@ export class RuntimeBridge {
* `run-finished` worker message so the maps stay bounded by concurrent-
* run count rather than process lifetime.
*
* Before eviction, sweeps the workspace to register any filesystem
* artifacts the per-tool `afterToolCall` hook missed. This is the
* backstop that keeps the DocumentCenter consistent with the disk even
* when individual projections skipped (bash against a non-git repo,
* path-ignore false positives, transient `git status` failures).
* Before eviction, sweeps the workspace to register filesystem
* artifacts produced during the run. This completion boundary keeps
* the DocumentCenter consistent with disk through one durable
* reconciliation acknowledgement.
*/
releaseRunResources(runId: string): RunResourceReleaseResult {
const ctx = this.activeRuns.get(runId);
@@ -311,8 +305,6 @@ export class RuntimeBridge {
return await this.handlePrepareRunRpc(request.runId, request.payload);
case "beforeToolCall":
return this.handleBeforeToolCallRpc(request.runId, request.payload as BeforeToolCallRpcPayload);
case "afterToolCall":
return this.handleAfterToolCallRpc(request.runId, request.payload as AfterToolCallRpcPayload);
case "invokeInternalTool":
return this.handleInvokeInternalToolRpc(request.runId, request.payload as InvokeInternalToolRpcPayload);
default: {
@@ -465,16 +457,4 @@ export class RuntimeBridge {
return { ok: true, result };
}
private handleAfterToolCallRpc(
runId: string,
payload: AfterToolCallRpcPayload,
): { ok: true; result: unknown } | { ok: false; error: string } {
const ctx = this.activeRuns.get(runId);
if (!ctx) {
return { ok: false, error: `afterToolCall: no active run ${runId}` };
}
const result = handleAfterToolCall(this.deps, ctx, payload);
return { ok: true, result };
}
}