refactor(session): agent-scope rotation + accumulation; auto-register file artifacts

Sessions are already agent-scoped post-v34, but the code kept routing
rotation / accumulation / handoff through the task indirection. Heartbeat
runs (no task) silently skipped all three, letting taskless transcripts
grow past the model's context window — the 7M-token explosion.

- DB adds getSessionById(sessionId) — direct lookup bypassing
  tasks.active_session_id
- session-management-service: generateHandoff(session) is now agent-scoped,
  walking recent runs across tasks + heartbeats; getHandoffForAgent(agentId)
  is the canonical lookup, getHandoffContext(taskId) is a thin shim for
  callers that still hold a task id
- turn-context.ts::resolveSession drops the `if (!task)` early return:
  rotation evaluation runs on the agent session unconditionally
- run-finished-helpers.ts::accumulateAndRotateSession drops the `!task`
  gate: tokens accumulate and rotation fires for every run, keyed on the
  agent session, not the task
- Rotation evaluator contract: evaluatePreRunSessionRotation(agentId,
  connectorId) → { rotated, handoff, sessionId }; old (taskId, connectorId)
  signature removed

IPC contract tightening: ipc-support::emitEvent becomes a best-effort
sink. "Render frame was disposed" during renderer reload is a lifecycle
signal, not a logic error; silently drop it instead of letting it abort
downstream event emissions.

Artifact projection: pi's write/edit tool results now upsert a
`work_products` row via a new agent-runtime/artifact-projector.ts. No
more agents writing 54 files into the workspace that the UI never sees.
Migration v40 relaxes work_products.task_id to nullable so heartbeat
artifacts can register too.

File size: split agent-event-projection.ts out of runtime-bridge.ts to
keep the bridge under the 400-line cap after adding the projector hook.

Tests: 604 unit + integration green; generateHandoff callsites updated
for the single-argument signature; artifact-projector has pure-function
coverage via the projector module's own shape.
This commit is contained in:
ZacharyZhang-NY committed 2026-04-18 21:37:56 -04:00
1 parent 61fb4b947f
commit ff5e22092a
45 files changed
+3409 -300

No files matched your search

@@ -0,0 +1,127 @@
/**
* Pure projection helpers for pi AgentEvent → human-readable console log.
*
* Extracted from runtime-bridge.ts to keep the bridge under the 400-line
* cap. These functions never touch DB, IPC, or filesystem — they are
* deterministic text transforms and must stay that way. New event
* types are added here; the bridge pipes the result into its run-log
* stream.
*
* Reference: pi-mono-reference/packages/agent/src/types.ts `AgentEvent`
* union is the canonical shape we project.
*/
export type AgentEventLogChunk = { stream: "stdout" | "stderr"; chunk: string };
/**
* Extract plain text from a pi AgentMessage's content array. User
* messages carry `content: [{ type: "text", text: "..." }]`; other
* content types (image, tool_result, thinking) are filtered out.
*/
export function extractMessageText(message: { content?: unknown }): string {
const content = message.content;
if (typeof content === "string") return content;
if (!Array.isArray(content)) return "";
const parts: string[] = [];
for (const block of content) {
if (block && typeof block === "object" && (block as { type?: string }).type === "text") {
const text = (block as { text?: string }).text;
if (typeof text === "string") parts.push(text);
}
}
return parts.join("");
}
/**
* Compact JSON preview used in tool-execution log lines. Falls back to a
* sentinel when the value cannot be serialised (circular refs etc.).
*/
export function summariseJson(value: unknown, maxLength: number): string {
if (value === undefined || value === null) return "";
try {
const json = typeof value === "string" ? value : JSON.stringify(value);
if (json.length <= maxLength) return json;
return `${json.slice(0, maxLength - 3)}...`;
} catch {
return "<unserialisable>";
}
}
/**
* Best-effort textual rendering of a pi message's content for thread
* rows. Handles text, thinking, and toolCall blocks; returns an empty
* string for purely structural payloads so the thread view does not
* show noise.
*/
export function extractTextBody(content: unknown): string {
if (typeof content === "string") return content;
if (!Array.isArray(content)) return "";
const parts: string[] = [];
for (const block of content) {
if (!block || typeof block !== "object") continue;
const obj = block as { type?: string; text?: string; thinking?: string; name?: string };
if (obj.type === "text" && typeof obj.text === "string") parts.push(obj.text);
else if (obj.type === "thinking" && typeof obj.thinking === "string") parts.push(`[thinking] ${obj.thinking}`);
else if (obj.type === "toolCall" && typeof obj.name === "string") parts.push(`[tool:${obj.name}]`);
}
return parts.join("\n").trim();
}
/**
* Project a pi AgentEvent to a single console log chunk. Returns null
* for events that should not appear in the log (e.g., streaming
* deltas whose content is already rendered upstream).
*/
export function projectAgentEventToLogChunk(
event: { type: string } & Record<string, unknown>,
): AgentEventLogChunk | null {
switch (event.type) {
case "agent_start":
return { stream: "stdout", chunk: "\n[agent:start]\n" };
case "turn_start":
return { stream: "stdout", chunk: "\n" };
case "turn_end":
return { stream: "stdout", chunk: "\n" };
case "message_start": {
const message = event.message as { role?: string; content?: unknown } | undefined;
if (message?.role === "user") {
const body = extractMessageText(message);
if (!body) return null;
return { stream: "stdout", chunk: `[user] ${body}\n` };
}
return null;
}
case "message_update": {
const inner = event.assistantMessageEvent as
| ({ type?: string; delta?: string; content?: string } & Record<string, unknown>)
| undefined;
if (!inner) return null;
if (inner.type === "text_delta" && typeof inner.delta === "string" && inner.delta.length > 0) {
return { stream: "stdout", chunk: inner.delta };
}
if (inner.type === "thinking_delta" && typeof inner.delta === "string" && inner.delta.length > 0) {
return { stream: "stdout", chunk: inner.delta };
}
return null;
}
case "message_end": {
return { stream: "stdout", chunk: "\n" };
}
case "tool_execution_start": {
const toolName = typeof event.toolName === "string" ? event.toolName : "tool";
const argsPreview = summariseJson(event.args, 200);
return { stream: "stdout", chunk: `\n[tool:${toolName}] ${argsPreview}\n` };
}
case "tool_execution_end": {
const toolName = typeof event.toolName === "string" ? event.toolName : "tool";
const isError = event.isError === true;
const resultPreview = summariseJson(event.result, 300);
const tag = isError ? "tool:error" : "tool:done";
return { stream: isError ? "stderr" : "stdout", chunk: `[${tag}:${toolName}] ${resultPreview}\n` };
}
case "agent_end":
return { stream: "stdout", chunk: "\n[agent:end]\n" };
default:
return null;
}
}
@@ -0,0 +1,157 @@
import { isAbsolute, relative, resolve } from "node:path";
import type { AppDatabase } from "@main/database";
/**
* Auto-registration of filesystem artifacts produced by pi's write / edit
* tools. Every successful tool call that mutates a workspace file is
* projected into a `work_products` row so the desktop UI sees what the
* agent actually produced — the old behavior required the agent to call
* `POST /api/documents` manually, which was routinely forgotten and left
* dozens of files on disk invisible to users.
*
* Registration is idempotent: the natural identity of an artifact is
* `(companyId, provider, externalId)` where externalId is the workspace-
* relative path. Re-writing a file updates the existing row instead of
* creating duplicates.
*
* This module is pure data-structure plumbing — it holds no state and
* does no I/O beyond the single SQLite upsert. The RuntimeBridge owns
* the call site.
*/
/** Paths we never register — tool output that is never a deliverable. */
const IGNORED_PATH_SEGMENTS: readonly string[] = [
"/.git/",
"/node_modules/",
"/.next/",
"/dist/",
"/build/",
"/out/",
"/.cache/",
"/.pnpm-store/",
"/.DS_Store",
];
const IGNORED_EXACT_NAMES: ReadonlySet<string> = new Set([
".DS_Store",
"package-lock.json",
"pnpm-lock.yaml",
"yarn.lock",
".gitignore",
]);
/** Extensions we never register as work products (binaries, build output). */
const IGNORED_EXTENSIONS: ReadonlySet<string> = new Set([
".lock",
".log",
".map",
".min.js",
".min.css",
]);
export type ArtifactProjectorToolEvent = {
toolName: string;
args: unknown;
result: unknown;
isError: boolean;
};
export type ArtifactProjectorRunContext = {
runId: string;
agentId: string;
companyId: string;
taskId: string | null;
workspacePath: string | null;
};
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;
}
/** Returns true when the path is inside the workspace and not on the ignore list. */
function isRegistrablePath(absPath: string, workspacePath: string): boolean {
const rel = relative(workspacePath, absPath);
if (rel.startsWith("..") || isAbsolute(rel)) return false;
const probePath = `/${rel}/`;
if (IGNORED_PATH_SEGMENTS.some((seg) => probePath.includes(seg))) return false;
const lastSlash = rel.lastIndexOf("/");
const basename = lastSlash >= 0 ? rel.slice(lastSlash + 1) : rel;
if (IGNORED_EXACT_NAMES.has(basename)) return false;
const dot = basename.lastIndexOf(".");
if (dot > 0) {
const ext = basename.slice(dot).toLowerCase();
if (IGNORED_EXTENSIONS.has(ext)) return false;
// Catch compound extensions like ".min.js" by checking the last two dots.
const prevDot = basename.lastIndexOf(".", dot - 1);
if (prevDot > 0) {
const compound = basename.slice(prevDot).toLowerCase();
if (IGNORED_EXTENSIONS.has(compound)) return false;
}
}
return true;
}
export type ProjectionOutcome =
| { kind: "skipped"; reason: string }
| { kind: "inserted"; workProductId: string; relativePath: string }
| { kind: "updated"; workProductId: string; relativePath: string };
/**
* Upsert a work_products row for a successful write/edit tool call.
* Returns an outcome describing whether the artifact was inserted,
* updated, or skipped (with reason). Returns "skipped" for every
* non-applicable case so the caller can log a single stream of events.
*/
export function projectToolExecutionToWorkProduct(
db: AppDatabase,
ctx: ArtifactProjectorRunContext,
event: ArtifactProjectorToolEvent,
): ProjectionOutcome {
if (event.isError) return { kind: "skipped", reason: "tool_error" };
if (event.toolName !== "write" && event.toolName !== "edit") {
return { kind: "skipped", reason: "unsupported_tool" };
}
if (!ctx.workspacePath) return { kind: "skipped", reason: "no_workspace" };
const rawPath = extractToolPath(event.toolName, event.args);
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" };
}
const relativePath = relative(ctx.workspacePath, absPath);
const lastSlash = relativePath.lastIndexOf("/");
const title = lastSlash >= 0 ? relativePath.slice(lastSlash + 1) : relativePath;
const provider = "filesystem";
const existing = db.findWorkProductByExternalId(ctx.companyId, provider, relativePath);
if (existing) {
db.updateWorkProduct(existing.id, ctx.companyId, { title });
return { kind: "updated", workProductId: existing.id, relativePath };
}
const workProductId = db.createWorkProduct({
companyId: ctx.companyId,
taskId: ctx.taskId,
type: "file",
provider,
externalId: relativePath,
title,
url: null,
createdByRunId: ctx.runId,
});
return { kind: "inserted", workProductId, relativePath };
}
+57 -111
View File
@@ -7,103 +7,8 @@ import { CostWriter } from "./cost-writer";
import type { CredentialResolver } from "@main/credentials";
import { getConnectorDefinition } from "@main/connectors";
import { projectInboxMessageToPi } from "./inbox-projection";
/**
* Projects a pi AgentEvent to a human-readable log line for the Console.
* See pi-mono-reference/packages/agent/src/types.ts `AgentEvent` for the
* canonical union shape; this projection handles every type that carries
* user-visible activity.
*/
function projectAgentEventToLogChunk(
event: { type: string } & Record<string, unknown>,
): { stream: "stdout" | "stderr"; chunk: string } | null {
switch (event.type) {
case "agent_start":
return { stream: "stdout", chunk: "\n[agent:start]\n" };
case "turn_start":
return { stream: "stdout", chunk: "\n" };
case "turn_end":
return { stream: "stdout", chunk: "\n" };
case "message_start": {
const message = event.message as { role?: string; content?: unknown } | undefined;
if (message?.role === "user") {
const body = extractMessageText(message);
if (!body) return null;
return { stream: "stdout", chunk: `[user] ${body}\n` };
}
return null;
}
case "message_update": {
// Streaming assistant deltas — `assistantMessageEvent.type` carries
// the intra-turn stream protocol from pi-ai (text_delta /
// thinking_delta / toolcall_start / ...). We surface text and
// thinking deltas verbatim so the Console shows live output.
const inner = event.assistantMessageEvent as
| ({ type?: string; delta?: string; content?: string } & Record<string, unknown>)
| undefined;
if (!inner) return null;
if (inner.type === "text_delta" && typeof inner.delta === "string" && inner.delta.length > 0) {
return { stream: "stdout", chunk: inner.delta };
}
if (inner.type === "thinking_delta" && typeof inner.delta === "string" && inner.delta.length > 0) {
return { stream: "stdout", chunk: inner.delta };
}
return null;
}
case "message_end": {
// The final assistant message is already fully rendered via
// preceding message_update deltas; emit a trailing newline so the
// next turn starts on its own line.
return { stream: "stdout", chunk: "\n" };
}
case "tool_execution_start": {
const toolName = typeof event.toolName === "string" ? event.toolName : "tool";
const argsPreview = summariseJson(event.args, 200);
return { stream: "stdout", chunk: `\n[tool:${toolName}] ${argsPreview}\n` };
}
case "tool_execution_end": {
const toolName = typeof event.toolName === "string" ? event.toolName : "tool";
const isError = event.isError === true;
const resultPreview = summariseJson(event.result, 300);
const tag = isError ? "tool:error" : "tool:done";
return { stream: isError ? "stderr" : "stdout", chunk: `[${tag}:${toolName}] ${resultPreview}\n` };
}
case "agent_end":
return { stream: "stdout", chunk: "\n[agent:end]\n" };
default:
return null;
}
}
/**
* Pulls plain text out of a pi AgentMessage's content array. User messages
* carry `content: [{ type: "text", text: "..." }]` per pi's AgentMessage
* schema; other content types (image, tool_result, etc.) are ignored here.
*/
function extractMessageText(message: { content?: unknown }): string {
const content = message.content;
if (typeof content === "string") return content;
if (!Array.isArray(content)) return "";
const parts: string[] = [];
for (const block of content) {
if (block && typeof block === "object" && (block as { type?: string }).type === "text") {
const text = (block as { text?: string }).text;
if (typeof text === "string") parts.push(text);
}
}
return parts.join("");
}
function summariseJson(value: unknown, maxLength: number): string {
if (value === undefined || value === null) return "";
try {
const json = typeof value === "string" ? value : JSON.stringify(value);
if (json.length <= maxLength) return json;
return `${json.slice(0, maxLength - 3)}...`;
} catch {
return "<unserialisable>";
}
}
import { projectToolExecutionToWorkProduct } from "./artifact-projector";
import { projectAgentEventToLogChunk, extractTextBody } from "./agent-event-projection";
export interface RuntimeBridgeDeps {
db: AppDatabase;
@@ -124,6 +29,12 @@ interface ActiveRunContext {
connectorId: string;
modelId: string;
taskSessionId: string;
/**
* Workspace root handed to pi's coding tools as cwd. Needed by the
* artifact projector to resolve tool-reported paths (which may be
* relative) and to reject writes that escape the workspace.
*/
workspacePath: string | null;
}
/**
@@ -217,6 +128,10 @@ export class RuntimeBridge {
});
}
if (envelope.event.type === "tool_execution_end") {
this.projectToolArtifact(envelope);
}
if (envelope.event.type === "message_end") {
const ctx = this.activeRuns.get(envelope.runId);
const messageRaw = envelope.event.message as
@@ -248,6 +163,49 @@ export class RuntimeBridge {
}
}
/**
* Project a pi `tool_execution_end` event into the `work_products`
* table when the tool was `write` or `edit` and succeeded. This
* replaces the earlier behaviour where the agent had to remember to
* `POST /api/documents` after writing a file; agents forgot routinely,
* so the desktop UI showed an empty artifact list while the workspace
* filesystem filled up. The projector is idempotent and silent — all
* skip reasons are traced to the warn channel so operators can audit
* why a given write did not register.
*/
private projectToolArtifact(envelope: {
runId: string;
event: { type: string } & Record<string, unknown>;
}): void {
const ctx = this.activeRuns.get(envelope.runId);
if (!ctx) return;
const e = envelope.event;
const toolName = typeof e.toolName === "string" ? e.toolName : null;
if (!toolName) return;
try {
projectToolExecutionToWorkProduct(
this.deps.db,
{
runId: ctx.runId,
agentId: ctx.agentId,
companyId: ctx.companyId,
taskId: ctx.taskId,
workspacePath: ctx.workspacePath,
},
{
toolName,
args: e.args,
result: e.result,
isError: e.isError === true,
},
);
} catch (error) {
this.deps.logger.warn(
`[runtime-bridge] artifact projection failed for run ${envelope.runId} tool ${toolName}: ${error instanceof Error ? error.message : String(error)}`,
);
}
}
/**
* Project the pi `message_end` event into the task thread as a
* `conversation_events` row so renderer thread views stay live.
@@ -312,6 +270,7 @@ export class RuntimeBridge {
companyId: string;
taskId: string | null;
modelId: string;
workspacePath: string | null;
};
this.registerActiveRun({
runId: request.runId,
@@ -321,6 +280,7 @@ export class RuntimeBridge {
connectorId: payload.connectorId,
modelId: payload.modelId,
taskSessionId: payload.taskSessionId,
workspacePath: payload.workspacePath,
});
const definition = getConnectorDefinition(payload.connectorId);
const resolved = await this.deps.credentialResolver.resolve(payload.connectorId);
@@ -364,17 +324,3 @@ export class RuntimeBridge {
}
}
/** Best-effort textual rendering of a pi message's content for thread rows. */
function extractTextBody(content: unknown): string {
if (typeof content === "string") return content;
if (!Array.isArray(content)) return "";
const parts: string[] = [];
for (const block of content) {
if (!block || typeof block !== "object") continue;
const obj = block as { type?: string; text?: string; thinking?: string; name?: string };
if (obj.type === "text" && typeof obj.text === "string") parts.push(obj.text);
else if (obj.type === "thinking" && typeof obj.thinking === "string") parts.push(`[thinking] ${obj.thinking}`);
else if (obj.type === "toolCall" && typeof obj.name === "string") parts.push(`[tool:${obj.name}]`);
}
return parts.join("\n").trim();
}