diff --git a/eslint.config.mjs b/eslint.config.mjs index df4fbccb..43582b4d 100644 --- a/eslint.config.mjs +++ b/eslint.config.mjs @@ -77,6 +77,20 @@ export default tseslint.config( }, }, }, + { + files: ["src/**/*.{ts,tsx}"], + rules: { + "max-lines": ["error", { max: 500, skipBlankLines: true, skipComments: false }], + }, + }, + { + // Locale dictionaries are data tables generated as TypeScript + // modules. The max-lines ceiling applies to executable source. + files: ["src/renderer/i18n/*.ts"], + rules: { + "max-lines": "off", + }, + }, { // C1 part 3/3 — block new top-level methods on AppDatabase. // diff --git a/src/main/agent-runtime/artifact-baseline.ts b/src/main/agent-runtime/artifact-baseline.ts new file mode 100644 index 00000000..9dbbbf26 --- /dev/null +++ b/src/main/agent-runtime/artifact-baseline.ts @@ -0,0 +1,97 @@ +import { relative, resolve } from "node:path"; +import { readdirSync, statSync } from "node:fs"; +import { isGitRepo } from "./artifact-git"; +import { logArtifactFsError, type Logger } from "./artifact-logging"; +import type { WorkspaceBaseline } from "./artifact-types"; + +const MAX_BASELINE_ENTRIES = 20_000; + +export const BASELINE_SKIP_DIR_NAMES: ReadonlySet = new Set([ + ".git", + "node_modules", + ".next", + "dist", + "build", + "out", + ".cache", + ".pnpm-store", +]); + +/** + * Capture a run-start mtime baseline only for non-git workspaces. + * Git workspaces use `git status` as the authoritative diff source. + */ +export function captureBaselineIfRequired( + workspacePath: string | null, + logger?: Logger, +): WorkspaceBaseline | null { + if (!workspacePath) return null; + if (isGitRepo(workspacePath)) return null; + try { + if (!statSync(workspacePath).isDirectory()) return null; + } catch (error) { + logArtifactFsError( + logger, + error, + "artifact.fs.workspace_stat_failed", + "Failed to stat workspace before artifact baseline capture.", + { workspacePath }, + ); + return null; + } + + const mtimeByRelPath = new Map(); + const stack: string[] = [workspacePath]; + let truncated = false; + while (stack.length > 0) { + if (mtimeByRelPath.size >= MAX_BASELINE_ENTRIES) { + truncated = true; + break; + } + const dir = stack.pop(); + if (!dir) break; + let entries: { name: string; isDirectory: () => boolean; isSymbolicLink: () => boolean }[]; + try { + entries = readdirSync(dir, { withFileTypes: true }); + } catch (error) { + logArtifactFsError( + logger, + error, + "artifact.fs.baseline_readdir_failed", + "Failed to read workspace directory during artifact baseline capture.", + { workspacePath, dir }, + ); + continue; + } + for (const entry of entries) { + const abs = resolve(dir, entry.name); + if (entry.isSymbolicLink()) continue; + if (entry.isDirectory()) { + if (BASELINE_SKIP_DIR_NAMES.has(entry.name)) continue; + stack.push(abs); + continue; + } + try { + const stat = statSync(abs); + if (!stat.isFile()) continue; + const rel = relative(workspacePath, abs); + if (rel.length === 0) continue; + mtimeByRelPath.set(rel, stat.mtimeMs); + } catch (error) { + logArtifactFsError( + logger, + error, + "artifact.fs.baseline_stat_failed", + "Failed to stat workspace file during artifact baseline capture.", + { workspacePath, absPath: abs }, + ); + continue; + } + if (mtimeByRelPath.size >= MAX_BASELINE_ENTRIES) { + truncated = true; + break; + } + } + } + return { mtimeByRelPath, capturedAt: Date.now(), truncated }; +} diff --git a/src/main/agent-runtime/artifact-change-detector.ts b/src/main/agent-runtime/artifact-change-detector.ts new file mode 100644 index 00000000..b46dc5a9 --- /dev/null +++ b/src/main/agent-runtime/artifact-change-detector.ts @@ -0,0 +1,139 @@ +import { execFileSync } from "node:child_process"; +import { readdirSync, statSync } from "node:fs"; +import { relative, resolve } from "node:path"; +import { BASELINE_SKIP_DIR_NAMES } from "./artifact-baseline"; +import { isGitRepo } from "./artifact-git"; +import { logArtifactFsError } from "./artifact-logging"; +import type { ArtifactProjectorRunContext, ChangeSet, GitChangeSet, WorkspaceBaseline } from "./artifact-types"; + +export function collectChangedPaths(ctx: ArtifactProjectorRunContext): ChangeSet { + if (!ctx.workspacePath) { + return { source: "none", changed: [], deleted: [], skipReason: "no_workspace" }; + } + if (isGitRepo(ctx.workspacePath)) { + try { + const gitSet = listChangedRelPaths(ctx.workspacePath); + return { source: "git", changed: gitSet.changed, deleted: gitSet.deleted }; + } catch (error) { + logArtifactFsError( + ctx.logger, + error, + "artifact.fs.git_status_failed", + "Failed to read git status during artifact projection.", + { workspacePath: ctx.workspacePath, runId: ctx.runId }, + ); + return { source: "none", changed: [], deleted: [], skipReason: "git_status_failed" }; + } + } + if (ctx.workspaceBaseline) { + const diff = diffAgainstBaseline(ctx.workspacePath, ctx.workspaceBaseline, ctx.logger); + return { + source: "baseline", + changed: diff.changed, + deleted: diff.deleted, + coverageTruncated: ctx.workspaceBaseline.truncated, + }; + } + return { source: "none", changed: [], deleted: [], skipReason: "no_baseline" }; +} + +function diffAgainstBaseline( + workspacePath: string, + baseline: WorkspaceBaseline, + logger: ArtifactProjectorRunContext["logger"], +): GitChangeSet { + const changed: string[] = []; + const seenRels = new Set(); + const stack: string[] = [workspacePath]; + while (stack.length > 0) { + const dir = stack.pop(); + if (!dir) break; + let entries: { name: string; isDirectory: () => boolean; isSymbolicLink: () => boolean }[]; + try { + entries = readdirSync(dir, { withFileTypes: true }); + } catch (error) { + logArtifactFsError( + logger, + error, + "artifact.fs.diff_readdir_failed", + "Failed to read workspace directory during artifact diff.", + { workspacePath, dir }, + ); + continue; + } + for (const entry of entries) { + if (entry.isSymbolicLink()) continue; + const abs = resolve(dir, entry.name); + if (entry.isDirectory()) { + if (BASELINE_SKIP_DIR_NAMES.has(entry.name)) continue; + stack.push(abs); + continue; + } + let mtimeMs: number; + try { + const stat = statSync(abs); + if (!stat.isFile()) continue; + mtimeMs = stat.mtimeMs; + } catch (error) { + logArtifactFsError( + logger, + error, + "artifact.fs.diff_stat_failed", + "Failed to stat workspace file during artifact diff.", + { workspacePath, absPath: abs }, + ); + continue; + } + const rel = relative(workspacePath, abs); + if (rel.length === 0) continue; + seenRels.add(rel); + const previous = baseline.mtimeByRelPath.get(rel); + if (previous === undefined) { + if (!baseline.truncated) changed.push(rel); + continue; + } + if (mtimeMs > previous) changed.push(rel); + } + } + const deleted: string[] = []; + for (const rel of baseline.mtimeByRelPath.keys()) { + if (!seenRels.has(rel)) deleted.push(rel); + } + return { changed, deleted }; +} + +function listChangedRelPaths(workspacePath: string): GitChangeSet { + const raw = execFileSync("git", ["status", "--porcelain=v1", "-z", "--untracked-files=all"], { + cwd: workspacePath, + timeout: 5_000, + encoding: "utf8", + maxBuffer: 1024 * 1024, + }); + if (!raw) return { changed: [], deleted: [] }; + const entries = raw.split("\0").filter(Boolean); + const changed: string[] = []; + const deleted: string[] = []; + for (let i = 0; i < entries.length; i++) { + const entry = entries[i]; + if (entry.length < 3) continue; + const indexStatus = entry[0]; + const worktreeStatus = entry[1]; + const rest = entry.slice(3); + if (indexStatus === "!" && worktreeStatus === "!") continue; + const isRename = indexStatus === "R" || worktreeStatus === "R"; + if (representsDeletion(indexStatus, worktreeStatus)) { + deleted.push(rest); + if (isRename) i += 1; + continue; + } + changed.push(rest); + if (isRename) i += 1; + } + return { changed, deleted }; +} + +function representsDeletion(indexStatus: string, worktreeStatus: string): boolean { + if (worktreeStatus === "D") return true; + if (indexStatus === "D" && worktreeStatus === " ") return true; + return false; +} diff --git a/src/main/agent-runtime/artifact-git.ts b/src/main/agent-runtime/artifact-git.ts new file mode 100644 index 00000000..3cd1b62f --- /dev/null +++ b/src/main/agent-runtime/artifact-git.ts @@ -0,0 +1,14 @@ +import { spawnSync } from "node:child_process"; + +/** + * Ask git whether `workspacePath` sits inside a working tree. This + * covers a plain clone, linked worktree, and submodule. + */ +export function isGitRepo(workspacePath: string): boolean { + const result = spawnSync("git", ["rev-parse", "--is-inside-work-tree"], { + cwd: workspacePath, + timeout: 2_000, + stdio: ["ignore", "ignore", "ignore"], + }); + return result.status === 0; +} diff --git a/src/main/agent-runtime/artifact-logging.ts b/src/main/agent-runtime/artifact-logging.ts new file mode 100644 index 00000000..99386989 --- /dev/null +++ b/src/main/agent-runtime/artifact-logging.ts @@ -0,0 +1,25 @@ +import { logAppError, type Logger } from "../errors/index.js"; + +const DEFAULT_ARTIFACT_LOGGER: Logger = console; + +export type { Logger }; + +export function logArtifactFsError( + logger: Logger | undefined, + error: unknown, + code: string, + message: string, + context: Record, +): void { + logAppError( + logger ?? DEFAULT_ARTIFACT_LOGGER, + error, + { + kind: "recoverable", + code, + message, + context, + }, + "warn", + ); +} diff --git a/src/main/agent-runtime/artifact-path-filter.ts b/src/main/agent-runtime/artifact-path-filter.ts new file mode 100644 index 00000000..64b78a08 --- /dev/null +++ b/src/main/agent-runtime/artifact-path-filter.ts @@ -0,0 +1,52 @@ +import { isAbsolute, relative } from "node:path"; + +/** Paths we never register because they are tool output or build/cache state. */ +const IGNORED_PATH_SEGMENTS: readonly string[] = [ + "/.git/", + "/node_modules/", + "/.next/", + "/dist/", + "/build/", + "/out/", + "/.cache/", + "/.pnpm-store/", + "/.DS_Store", +]; + +const IGNORED_EXACT_NAMES: ReadonlySet = new Set([ + ".DS_Store", + "package-lock.json", + "pnpm-lock.yaml", + "yarn.lock", + ".gitignore", +]); + +const IGNORED_EXTENSIONS: ReadonlySet = new Set([ + ".lock", + ".log", + ".map", + ".min.js", + ".min.css", +]); + +/** Returns true when the path is inside the workspace and passes the artifact filter. */ +export 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; + const prevDot = basename.lastIndexOf(".", dot - 1); + if (prevDot > 0) { + const compound = basename.slice(prevDot).toLowerCase(); + if (IGNORED_EXTENSIONS.has(compound)) return false; + } + } + return true; +} diff --git a/src/main/agent-runtime/artifact-projector.ts b/src/main/agent-runtime/artifact-projector.ts index ae1f9a31..8d4e57a3 100644 --- a/src/main/agent-runtime/artifact-projector.ts +++ b/src/main/agent-runtime/artifact-projector.ts @@ -1,233 +1,38 @@ -// TODO(M-followup by 2026-05-20): decompose — current file exceeds 500 LOC -// ceiling (637 LOC) and mixes fs-walk baseline capture, git-diff projection, -// and work-product reconciliation. Split into baseline / diff / projector -// modules. If 2026-05-20 passes without decomposition, a follow-up session -// MUST revisit and either decompose or justify the ceiling exception. -import { isAbsolute, relative, resolve } from "node:path"; -import { execFileSync, spawnSync } from "node:child_process"; -import { readdirSync, statSync } from "node:fs"; import type { AppDatabase } from "@main/database"; -import { logAppError, type Logger } from "../errors/index.js"; +import { collectChangedPaths } from "./artifact-change-detector"; +import { + projectChangedPaths, + projectSinglePathArtifact, + retireDeletedArtifacts, +} from "./artifact-work-products"; +import type { + ArtifactProjectorRunContext, + ArtifactProjectorToolEvent, + ChangeSet, + ProjectionOutcome, +} from "./artifact-types"; + +export { captureBaselineIfRequired } from "./artifact-baseline"; +export type { + ArtifactProjectorRunContext, + ArtifactProjectorToolEvent, + ProjectionOutcome, + WorkspaceBaseline, +} from "./artifact-types"; /** * Filesystem artifact protocol for pi tool invocations. * - * Every successful destructive tool call (`write`, `edit`, `bash`) is - * projected into the `work_products` table so the desktop UI can - * surface what the agent actually produced without requiring the agent - * to remember to POST /api/documents. + * 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. * - * Projection is idempotent: the natural identity of an artifact is + * Projection is idempotent through the natural identity * `(companyId, workspaceId, provider="filesystem", externalId=)`. - * The first tool call inserts; subsequent calls to the same path go - * through `updateWorkProduct`, which also refreshes - * `last_touched_by_run_id` so every run that modifies the file shows - * up in the deliverable signal and in the DocumentCenter preview. - * - * `write` / `edit` expose the touched path in their args — the - * projection path is trivial. `bash` is opaque: the tool may run any - * command (git commit, npm build, shell redirection). Diffing uses two - * sources of truth, chosen per workspace: - * - * - git workspace → `git status --porcelain` reports modified and - * untracked files relative to HEAD; - * - non-git workspace → an mtime baseline captured at run start - * reports every file whose mtime advanced during the run. - * - * Either way the projector only upserts paths that survive the - * registrability filter (`IGNORED_PATH_SEGMENTS` / extensions), so - * build output and caches never pollute the DocumentCenter. + * Git workspaces use `git status`; non-git workspaces use a run-start + * mtime baseline captured by `captureBaselineIfRequired`. */ -/** 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 = 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 = new Set([ - ".lock", - ".log", - ".map", - ".min.js", - ".min.css", -]); - -/** Bash scan budget: cap file count + byte count per invocation so a - * runaway rebuild cannot flood the artifact table. */ -const MAX_BASH_PROJECTED_FILES = 50; - -export type ArtifactProjectorToolEvent = { - toolName: string; - args: unknown; - result: unknown; - isError: boolean; -}; - -export type ArtifactProjectorRunContext = { - runId: string; - agentId: string; - companyId: string; - taskId: string | null; - workspaceId: string | null; - workspacePath: string | null; - 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. - */ - workspaceBaseline: WorkspaceBaseline | null; -}; - -export interface WorkspaceBaseline { - /** Relative path → mtimeMs at capture. */ - mtimeByRelPath: Map; - /** Unix ms at which the baseline was taken. */ - capturedAt: number; - /** True when the walker stopped early because the cap was hit; the baseline is still usable for the paths that were captured. */ - truncated: boolean; -} - -const MAX_BASELINE_ENTRIES = 20_000; -const BASELINE_SKIP_DIR_NAMES: ReadonlySet = new Set([ - ".git", - "node_modules", - ".next", - "dist", - "build", - "out", - ".cache", - ".pnpm-store", -]); - -const DEFAULT_ARTIFACT_LOGGER: Logger = console; - -function logArtifactFsError( - logger: Logger | undefined, - error: unknown, - code: string, - message: string, - context: Record, -): void { - logAppError( - logger ?? DEFAULT_ARTIFACT_LOGGER, - error, - { - kind: "recoverable", - code, - message, - context, - }, - "warn", - ); -} - -/** - * Capture a run-start mtime baseline only for workspaces that need - * one. Git workspaces return null here because `git status` already - * provides the authoritative diff — paying for an O(tree) walk on - * every run just to throw the result away is a hot-path tax the - * scheduler should not be paying. - * - * For non-git workspaces, walk the workspace and record the mtime of - * every non-directory entry. Directories matching - * `BASELINE_SKIP_DIR_NAMES` are pruned at walk time — the same - * categories the registrability filter rejects — so the baseline - * stays bounded on real project trees. The walker stops early at - * `MAX_BASELINE_ENTRIES` and flags `truncated`; the partial baseline - * still covers every path that was captured while the operator sees - * the cap signal. - */ -export function captureBaselineIfRequired( - workspacePath: string | null, - logger?: Logger, -): WorkspaceBaseline | null { - if (!workspacePath) return null; - if (isGitRepo(workspacePath)) return null; - try { - if (!statSync(workspacePath).isDirectory()) return null; - } catch (error) { - logArtifactFsError( - logger, - error, - "artifact.fs.workspace_stat_failed", - "Failed to stat workspace before artifact baseline capture.", - { workspacePath }, - ); - return null; - } - const mtimeByRelPath = new Map(); - const stack: string[] = [workspacePath]; - let truncated = false; - while (stack.length > 0) { - if (mtimeByRelPath.size >= MAX_BASELINE_ENTRIES) { - truncated = true; - break; - } - const dir = stack.pop(); - if (!dir) break; - let entries: { name: string; isDirectory: () => boolean; isSymbolicLink: () => boolean }[]; - try { - entries = readdirSync(dir, { withFileTypes: true }); - } catch (error) { - logArtifactFsError( - logger, - error, - "artifact.fs.baseline_readdir_failed", - "Failed to read workspace directory during artifact baseline capture.", - { workspacePath, dir }, - ); - continue; - } - for (const entry of entries) { - const abs = resolve(dir, entry.name); - if (entry.isSymbolicLink()) continue; - if (entry.isDirectory()) { - if (BASELINE_SKIP_DIR_NAMES.has(entry.name)) continue; - stack.push(abs); - continue; - } - try { - const stat = statSync(abs); - if (!stat.isFile()) continue; - const rel = relative(workspacePath, abs); - if (rel.length === 0) continue; - mtimeByRelPath.set(rel, stat.mtimeMs); - } catch (error) { - logArtifactFsError( - logger, - error, - "artifact.fs.baseline_stat_failed", - "Failed to stat workspace file during artifact baseline capture.", - { workspacePath, absPath: abs }, - ); - continue; - } - if (mtimeByRelPath.size >= MAX_BASELINE_ENTRIES) { - truncated = true; - break; - } - } - } - return { mtimeByRelPath, capturedAt: Date.now(), truncated }; -} - type WriteArgs = { path: unknown; content?: unknown }; type EditArgs = { path: unknown }; @@ -245,41 +50,10 @@ function extractToolPath(toolName: string, args: unknown): string | 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; - 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 } - | { kind: "retired"; workProductId: string; relativePath: string }; - /** * Upsert work_products rows for a successful destructive tool call. - * For `write` / `edit` the path comes directly from the tool args. - * For `bash` we scan the workspace via `git status --porcelain`. - * - * Always returns an array — `write`/`edit` touch one path so the array - * holds a single outcome; `bash` can surface many. + * `write` / `edit` use their explicit path; `bash` scans the workspace + * change set through git status or the captured baseline. */ export function projectToolExecutionToWorkProduct( db: AppDatabase, @@ -290,7 +64,7 @@ export function projectToolExecutionToWorkProduct( if (!ctx.workspacePath) return [{ kind: "skipped", reason: "no_workspace" }]; if (event.toolName === "write" || event.toolName === "edit") { - return [projectSinglePathArtifact(db, ctx, event.toolName, event.args)]; + return [projectSinglePathArtifact(db, ctx, extractToolPath(event.toolName, event.args))]; } if (event.toolName === "bash") { return projectBashArtifacts(db, ctx); @@ -300,258 +74,28 @@ export function projectToolExecutionToWorkProduct( /** * End-of-run reconciliation sweep. The per-tool `afterToolCall` hook - * registers artifacts eagerly, but individual projections can still - * skip (path on the ignore list, transient `git status` failure, or - * an mtime baseline that was truncated). The sweep is the backstop: - * every file the change-detector still knows about is upserted once - * more before the run's resources are released, so the DocumentCenter - * is never empty just because a single hook call lost a row. - * - * Diffing uses the same dual-source strategy as `projectBashArtifacts`: - * `git status` when a `.git` is present, the captured mtime baseline - * otherwise. Idempotent through `findWorkProductByExternalId`, which - * keys on `(company, workspace, provider, externalId)`. + * registers artifacts eagerly; the sweep catches files missed by a + * transient hook failure or a bash invocation that touched many paths. */ export function reconcileWorkspaceArtifacts( db: AppDatabase, ctx: ArtifactProjectorRunContext, ): ProjectionOutcome[] { - const change = collectChangedPaths(ctx); - if (change.source === "none") { - return [{ kind: "skipped", reason: change.skipReason ?? "no_changes" }]; - } - const outcomes: ProjectionOutcome[] = []; - if (change.changed.length === 0 && change.deleted.length === 0) { - outcomes.push({ kind: "skipped", reason: "no_changes" }); - } else { - outcomes.push(...projectChangedPaths(db, ctx, change.changed)); - outcomes.push(...retireDeletedArtifacts(db, ctx, change.deleted)); - } - if (change.coverageTruncated) { - outcomes.push({ kind: "skipped", reason: "baseline_truncated" }); - } - return outcomes; -} - -function projectChangedPaths( - db: AppDatabase, - ctx: ArtifactProjectorRunContext, - changed: string[], -): ProjectionOutcome[] { - 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) { - const absPath = resolve(workspacePath, rel); - if (!isRegistrablePath(absPath, workspacePath)) { - outcomes.push({ kind: "skipped", reason: "path_ignored" }); - continue; - } - 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; -} - -/** - * Archive `work_products` rows whose filesystem sources the change - * detector reported as removed during this run. The row stays in the - * table as a tombstone (operators can audit who produced what) but - * drops to `status = "archived"` so the DocumentCenter stops - * surfacing it and `openWorkProduct` no longer leads to an empty - * path. `last_touched_by_run_id` carries the retiring run so the - * provenance timeline records the deletion alongside the earlier - * edits. - * - * Every detected deletion is processed — retirement is a bulk - * `UPDATE … IN (…)` chunked by the DB layer, so a large `rm -r` - * retires the full tail instead of leaving the overflow visible. - */ -function retireDeletedArtifacts( - db: AppDatabase, - ctx: ArtifactProjectorRunContext, - deleted: string[], -): ProjectionOutcome[] { - if (deleted.length === 0) return []; - const archived = db.documentsMod.archiveWorkProductsByExternalPaths({ - companyId: ctx.companyId, - workspaceId: ctx.workspaceId, - provider: "filesystem", - paths: deleted, - lastTouchedByRunId: ctx.runId, - }); - return archived.map(({ id, externalId }) => ({ - kind: "retired", - workProductId: id, - relativePath: externalId, - })); -} - -/** - * Compare current workspace mtimes against the run-start baseline. - * Skipped-dir filtering mirrors the baseline walker so the diff and - * the capture stay in sync. - * - * A file is reported as `changed` when either: - * - * - its mtime advanced past the captured value, or - * - it was not in the baseline AND the baseline saw the whole tree - * (`baseline.truncated === false`). - * - * A file is reported as `deleted` when it was in the baseline at - * capture time and is no longer a regular file on disk — the - * retirement path archives the matching `work_products` row so the - * DocumentCenter stops showing a removed file. - * - * A truncated baseline deliberately stops classifying unsampled files - * as new: the walker gave up before reaching them, so we cannot - * distinguish a file that was always there from a file the run just - * wrote. Falsely registering every one of them as a deliverable is - * worse than missing a few that live past the sample cap. Callers - * emit a `baseline_truncated` outcome so operators see the degraded - * signal. - */ -function diffAgainstBaseline( - workspacePath: string, - baseline: WorkspaceBaseline, - logger: Logger | undefined, -): GitChangeSet { - const changed: string[] = []; - const seenRels = new Set(); - const stack: string[] = [workspacePath]; - while (stack.length > 0) { - const dir = stack.pop(); - if (!dir) break; - let entries: { name: string; isDirectory: () => boolean; isSymbolicLink: () => boolean }[]; - try { - entries = readdirSync(dir, { withFileTypes: true }); - } catch (error) { - logArtifactFsError( - logger, - error, - "artifact.fs.diff_readdir_failed", - "Failed to read workspace directory during artifact diff.", - { workspacePath, dir }, - ); - continue; - } - for (const entry of entries) { - if (entry.isSymbolicLink()) continue; - const abs = resolve(dir, entry.name); - if (entry.isDirectory()) { - if (BASELINE_SKIP_DIR_NAMES.has(entry.name)) continue; - stack.push(abs); - continue; - } - let mtimeMs: number; - try { - const stat = statSync(abs); - if (!stat.isFile()) continue; - mtimeMs = stat.mtimeMs; - } catch (error) { - logArtifactFsError( - logger, - error, - "artifact.fs.diff_stat_failed", - "Failed to stat workspace file during artifact diff.", - { workspacePath, absPath: abs }, - ); - continue; - } - const rel = relative(workspacePath, abs); - if (rel.length === 0) continue; - seenRels.add(rel); - const previous = baseline.mtimeByRelPath.get(rel); - if (previous === undefined) { - if (!baseline.truncated) changed.push(rel); - continue; - } - if (mtimeMs > previous) changed.push(rel); - } - } - const deleted: string[] = []; - for (const rel of baseline.mtimeByRelPath.keys()) { - if (!seenRels.has(rel)) deleted.push(rel); - } - return { changed, deleted }; -} - -function projectSinglePathArtifact( - db: AppDatabase, - ctx: ArtifactProjectorRunContext, - toolName: string, - args: unknown, -): ProjectionOutcome { - if (!ctx.workspacePath) return { kind: "skipped", reason: "no_workspace" }; - const rawPath = extractToolPath(toolName, 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" }; - } - return upsertFilesystemArtifact(db, ctx, ctx.workspacePath, absPath); -} - -type ChangeSource = "git" | "baseline" | "none"; - -interface ChangeSet { - source: ChangeSource; - /** Paths, relative to the workspace root, that currently exist on disk with a new or advanced mtime. */ - changed: string[]; - /** Paths, relative to the workspace root, that the detector reports as removed during the run. */ - deleted: string[]; - /** Reason the source chose to skip (only meaningful when source === "none"). */ - skipReason?: string; - /** - * True when the baseline walker stopped before sampling the whole - * tree. Callers surface it as a `baseline_truncated` skip outcome so - * the operator can see that unsampled files are deliberately outside - * the diff — registering them would falsely claim hundreds of - * unchanged files as new deliverables. - */ - coverageTruncated?: boolean; -} - -function collectChangedPaths(ctx: ArtifactProjectorRunContext): ChangeSet { - if (!ctx.workspacePath) { - return { source: "none", changed: [], deleted: [], skipReason: "no_workspace" }; - } - if (isGitRepo(ctx.workspacePath)) { - try { - const gitSet = listChangedRelPaths(ctx.workspacePath); - return { source: "git", changed: gitSet.changed, deleted: gitSet.deleted }; - } catch (error) { - logArtifactFsError( - ctx.logger, - error, - "artifact.fs.git_status_failed", - "Failed to read git status during artifact projection.", - { workspacePath: ctx.workspacePath, runId: ctx.runId }, - ); - return { source: "none", changed: [], deleted: [], skipReason: "git_status_failed" }; - } - } - if (ctx.workspaceBaseline) { - const diff = diffAgainstBaseline(ctx.workspacePath, ctx.workspaceBaseline, ctx.logger); - return { - source: "baseline", - changed: diff.changed, - deleted: diff.deleted, - coverageTruncated: ctx.workspaceBaseline.truncated, - }; - } - return { source: "none", changed: [], deleted: [], skipReason: "no_baseline" }; + return projectCollectedChanges(db, ctx, collectChangedPaths(ctx)); } function projectBashArtifacts( db: AppDatabase, ctx: ArtifactProjectorRunContext, ): ProjectionOutcome[] { - const change = collectChangedPaths(ctx); + return projectCollectedChanges(db, ctx, collectChangedPaths(ctx)); +} + +function projectCollectedChanges( + db: AppDatabase, + ctx: ArtifactProjectorRunContext, + change: ChangeSet, +): ProjectionOutcome[] { if (change.source === "none") { return [{ kind: "skipped", reason: change.skipReason ?? "no_changes" }]; } @@ -567,168 +111,3 @@ function projectBashArtifacts( } return outcomes; } - -/** - * Ask git whether `workspacePath` sits inside a working tree. This - * covers the three forms the `.git` entry takes in real setups: - * - * - a directory (a plain clone), - * - a file with `gitdir: …` (a linked worktree), or - * - a file with `gitdir: …/modules/…` (a submodule). - * - * The previous `existsSync + isDirectory` probe only recognised the - * first form, so linked worktrees and submodules silently degraded - * onto the baseline path — and lost `git status` as the authoritative - * diff source. `rev-parse --is-inside-work-tree` exits 0 whenever git - * itself would accept the directory, which is the semantics we want. - */ -function isGitRepo(workspacePath: string): boolean { - const result = spawnSync("git", ["rev-parse", "--is-inside-work-tree"], { - cwd: workspacePath, - timeout: 2_000, - stdio: ["ignore", "ignore", "ignore"], - }); - return result.status === 0; -} - -interface GitChangeSet { - changed: string[]; - deleted: string[]; -} - -/** - * Enumerate porcelain status entries and split them into "currently on - * disk" (`changed`) and "removed during the run" (`deleted`). The - * retirement path consumes `deleted` to archive the matching - * `work_products` row so the DocumentCenter stops showing a file that - * no longer exists. - * - * Porcelain v1 format per entry: `XY path`, NUL-terminated. For - * renames the new path is the entry and the old path follows as its - * own NUL-terminated trailer — we advance past the trailer without - * emitting it. - */ -function listChangedRelPaths(workspacePath: string): GitChangeSet { - const raw = execFileSync("git", ["status", "--porcelain=v1", "-z", "--untracked-files=all"], { - cwd: workspacePath, - timeout: 5_000, - encoding: "utf8", - maxBuffer: 1024 * 1024, - }); - if (!raw) return { changed: [], deleted: [] }; - const entries = raw.split("\0").filter(Boolean); - const changed: string[] = []; - const deleted: string[] = []; - for (let i = 0; i < entries.length; i++) { - const entry = entries[i]; - if (entry.length < 3) continue; - const indexStatus = entry[0]; - const worktreeStatus = entry[1]; - const rest = entry.slice(3); - if (indexStatus === "!" && worktreeStatus === "!") continue; - const isRename = indexStatus === "R" || worktreeStatus === "R"; - if (representsDeletion(indexStatus, worktreeStatus)) { - deleted.push(rest); - if (isRename) i += 1; - continue; - } - changed.push(rest); - if (isRename) i += 1; - } - return { changed, deleted }; -} - -function representsDeletion(indexStatus: string, worktreeStatus: string): boolean { - // Worktree missing (`Y = D`) always means the file is not on disk. - if (worktreeStatus === "D") return true; - // Staged deletion whose worktree matches (`XD` + Y space) means the - // file was removed from the tree and the worktree is in sync; the - // retirement path needs to hear about it so the corresponding - // `work_products` row can be archived. - if (indexStatus === "D" && worktreeStatus === " ") return true; - return false; -} - -function upsertFilesystemArtifact( - db: AppDatabase, - ctx: ArtifactProjectorRunContext, - workspacePath: string, - absPath: string, -): ProjectionOutcome { - const relativePath = relative(workspacePath, absPath); - - // Defense-in-depth against the change-detectors (git status + the - // baseline diff) reporting a path whose file is no longer a regular - // file on disk: a staged deletion that was undone mid-run, a symlink - // that replaced a file, a path that the agent touched and then - // rm'd. Registering it would plant a ghost row in `work_products` - // whose DocumentCenter "open" link leads nowhere. - let stat: ReturnType; - try { - stat = statSync(absPath); - } catch (error) { - logArtifactFsError( - ctx.logger, - error, - "artifact.fs.path_stat_failed", - "Failed to stat filesystem artifact before projection.", - { - workspacePath, - absPath, - runId: ctx.runId, - }, - ); - return { kind: "skipped", reason: "path_missing" }; - } - if (!stat.isFile()) { - return { kind: "skipped", reason: "not_a_regular_file" }; - } - - const lastSlash = relativePath.lastIndexOf("/"); - const title = lastSlash >= 0 ? relativePath.slice(lastSlash + 1) : relativePath; - - const provider = "filesystem"; - // Reverse-resolve workspaceId from workspacePath when ctx didn't - // carry it. The run may have been created before workspace_id was - // tracked end-to-end, but the projector only ever walks a workspace - // whose local_path is known — that path uniquely identifies the - // workspace row, so it is safe to persist the foreign key instead - // of writing another orphan. - const resolvedWorkspaceId = - ctx.workspaceId ?? db.workspacesMod.findWorkspaceIdByLocalPath(ctx.companyId, workspacePath); - // Identity is scoped to (company, workspace, provider, externalId) so two - // workspaces with the same relative path do not collapse onto one row. - // Every touch refreshes `last_touched_by_run_id` so the deliverable - // signal, the DocumentCenter preview, and the run-provenance readers - // all see the latest run that wrote the file. - const existing = db.documentsMod.findWorkProductByExternalId(ctx.companyId, provider, relativePath, resolvedWorkspaceId); - if (existing) { - // `status: "active"` flips a previously-archived row (earlier run - // retired the path, this run recreated it) back into the visible - // DocumentCenter set. `lastTouchedByRunId` keeps the provenance - // timeline pointed at the run that just wrote the bytes. - // Only include workspaceId in the patch when we actually have one — - // otherwise the UPDATE would clobber a previously-resolved - // workspace_id back to null (the original source of orphan rows). - const patch: Parameters[2] = { - title, - status: "active", - lastTouchedByRunId: ctx.runId, - }; - if (resolvedWorkspaceId) patch.workspaceId = resolvedWorkspaceId; - db.documentsMod.updateWorkProduct(existing.id, ctx.companyId, patch); - return { kind: "updated", workProductId: existing.id, relativePath }; - } - const workProductId = db.documentsMod.createWorkProduct({ - companyId: ctx.companyId, - taskId: ctx.taskId, - workspaceId: resolvedWorkspaceId, - type: "file", - provider, - externalId: relativePath, - title, - url: null, - createdByRunId: ctx.runId, - }); - return { kind: "inserted", workProductId, relativePath }; -} diff --git a/src/main/agent-runtime/artifact-types.ts b/src/main/agent-runtime/artifact-types.ts new file mode 100644 index 00000000..8e42b496 --- /dev/null +++ b/src/main/agent-runtime/artifact-types.ts @@ -0,0 +1,62 @@ +import type { Logger } from "./artifact-logging"; + +export type ArtifactProjectorToolEvent = { + toolName: string; + args: unknown; + result: unknown; + isError: boolean; +}; + +export type ArtifactProjectorRunContext = { + runId: string; + agentId: string; + companyId: string; + taskId: string | null; + workspaceId: string | null; + workspacePath: string | null; + 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. + */ + workspaceBaseline: WorkspaceBaseline | null; +}; + +export interface WorkspaceBaseline { + /** Relative path -> mtimeMs at capture. */ + mtimeByRelPath: Map; + /** Unix ms at which the baseline was taken. */ + capturedAt: number; + /** True when the walker stopped early because the cap was hit; the baseline is still usable for captured paths. */ + truncated: boolean; +} + +export type ProjectionOutcome = + | { kind: "skipped"; reason: string } + | { kind: "inserted"; workProductId: string; relativePath: string } + | { kind: "updated"; workProductId: string; relativePath: string } + | { kind: "retired"; workProductId: string; relativePath: string }; + +export interface GitChangeSet { + changed: string[]; + deleted: string[]; +} + +export type ChangeSource = "git" | "baseline" | "none"; + +export interface ChangeSet { + source: ChangeSource; + /** Paths, relative to the workspace root, that currently exist on disk with a new or advanced mtime. */ + changed: string[]; + /** Paths, relative to the workspace root, that the detector reports as removed during the run. */ + deleted: string[]; + /** Reason the source chose to skip, meaningful when source is "none". */ + skipReason?: string; + /** + * True when the baseline walker stopped before sampling the whole + * tree. Callers surface it as a `baseline_truncated` skip outcome. + */ + coverageTruncated?: boolean; +} diff --git a/src/main/agent-runtime/artifact-work-products.ts b/src/main/agent-runtime/artifact-work-products.ts new file mode 100644 index 00000000..6cf26c69 --- /dev/null +++ b/src/main/agent-runtime/artifact-work-products.ts @@ -0,0 +1,129 @@ +import { statSync } from "node:fs"; +import { isAbsolute, 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, + changed: string[], +): ProjectionOutcome[] { + 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) { + const absPath = resolve(workspacePath, rel); + if (!isRegistrablePath(absPath, workspacePath)) { + outcomes.push({ kind: "skipped", reason: "path_ignored" }); + continue; + } + 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; +} + +export function retireDeletedArtifacts( + db: AppDatabase, + ctx: ArtifactProjectorRunContext, + deleted: string[], +): ProjectionOutcome[] { + if (deleted.length === 0) return []; + const archived = db.documentsMod.archiveWorkProductsByExternalPaths({ + companyId: ctx.companyId, + workspaceId: ctx.workspaceId, + provider: "filesystem", + paths: deleted, + lastTouchedByRunId: ctx.runId, + }); + return archived.map(({ id, externalId }) => ({ + kind: "retired", + workProductId: id, + relativePath: externalId, + })); +} + +function upsertFilesystemArtifact( + db: AppDatabase, + ctx: ArtifactProjectorRunContext, + workspacePath: string, + absPath: string, +): ProjectionOutcome { + const relativePath = relative(workspacePath, absPath); + let stat: ReturnType; + try { + stat = statSync(absPath); + } catch (error) { + logArtifactFsError( + ctx.logger, + error, + "artifact.fs.path_stat_failed", + "Failed to stat filesystem artifact before projection.", + { + workspacePath, + absPath, + runId: ctx.runId, + }, + ); + return { kind: "skipped", reason: "path_missing" }; + } + if (!stat.isFile()) { + return { kind: "skipped", reason: "not_a_regular_file" }; + } + + const lastSlash = relativePath.lastIndexOf("/"); + const title = lastSlash >= 0 ? relativePath.slice(lastSlash + 1) : relativePath; + const provider = "filesystem"; + const resolvedWorkspaceId = + ctx.workspaceId ?? db.workspacesMod.findWorkspaceIdByLocalPath(ctx.companyId, workspacePath); + const existing = db.documentsMod.findWorkProductByExternalId( + ctx.companyId, + provider, + relativePath, + resolvedWorkspaceId, + ); + if (existing) { + const patch: Parameters[2] = { + title, + status: "active", + lastTouchedByRunId: ctx.runId, + }; + if (resolvedWorkspaceId) patch.workspaceId = resolvedWorkspaceId; + db.documentsMod.updateWorkProduct(existing.id, ctx.companyId, patch); + return { kind: "updated", workProductId: existing.id, relativePath }; + } + const workProductId = db.documentsMod.createWorkProduct({ + companyId: ctx.companyId, + taskId: ctx.taskId, + workspaceId: resolvedWorkspaceId, + type: "file", + provider, + externalId: relativePath, + title, + url: null, + createdByRunId: ctx.runId, + }); + return { kind: "inserted", workProductId, relativePath }; +} diff --git a/src/main/agent-runtime/inbound-correlation-registry.ts b/src/main/agent-runtime/inbound-correlation-registry.ts new file mode 100644 index 00000000..929272ca --- /dev/null +++ b/src/main/agent-runtime/inbound-correlation-registry.ts @@ -0,0 +1,75 @@ +import type { AgentMessageRecord } from "@shared/types"; + +export interface InboundCorrelationEntry { + fromAgentId: string; + senderKind: AgentMessageRecord["senderKind"]; + messageId: string; + /** + * Stable pi-message timestamp that the projected user turn carries. + * Reverse-indexed so `message_end` can resolve back to `messageId`. + */ + piTimestamp: number; +} + +interface InboundCorrelationRegistry { + byMessageId: Map; + byPiTimestamp: Map; + lastAllocatedTimestamp: number; +} + +const MAX_INBOUND_CORRELATIONS_PER_RUN = 512; + +export class InboundCorrelationRegistryStore { + private readonly registries = new Map(); + + register(runId: string, record: AgentMessageRecord, baselineTimestamp: number): number { + const registry = this.getOrCreate(runId); + const allocated = Math.max(baselineTimestamp, registry.lastAllocatedTimestamp + 1); + registry.lastAllocatedTimestamp = allocated; + const entry: InboundCorrelationEntry = { + fromAgentId: record.fromAgentId, + senderKind: record.senderKind, + messageId: record.id, + piTimestamp: allocated, + }; + registry.byMessageId.set(record.id, entry); + registry.byPiTimestamp.set(allocated, record.id); + this.evictOldestIfNeeded(registry); + return allocated; + } + + lookup(runId: string, timestamp: number | undefined): InboundCorrelationEntry | null { + if (typeof timestamp !== "number" || !Number.isFinite(timestamp)) return null; + const registry = this.registries.get(runId); + if (!registry) return null; + const messageId = registry.byPiTimestamp.get(timestamp); + if (!messageId) return null; + return registry.byMessageId.get(messageId) ?? null; + } + + deleteRun(runId: string): void { + this.registries.delete(runId); + } + + private getOrCreate(runId: string): InboundCorrelationRegistry { + const existing = this.registries.get(runId); + if (existing) return existing; + const registry: InboundCorrelationRegistry = { + byMessageId: new Map(), + byPiTimestamp: new Map(), + lastAllocatedTimestamp: 0, + }; + this.registries.set(runId, registry); + return registry; + } + + private evictOldestIfNeeded(registry: InboundCorrelationRegistry): void { + while (registry.byMessageId.size > MAX_INBOUND_CORRELATIONS_PER_RUN) { + const oldestMessageId = registry.byMessageId.keys().next().value; + if (!oldestMessageId) break; + const entry = registry.byMessageId.get(oldestMessageId); + registry.byMessageId.delete(oldestMessageId); + if (entry) registry.byPiTimestamp.delete(entry.piTimestamp); + } + } +} diff --git a/src/main/agent-runtime/runtime-bridge.ts b/src/main/agent-runtime/runtime-bridge.ts index 9f1e6ae6..9908535e 100644 --- a/src/main/agent-runtime/runtime-bridge.ts +++ b/src/main/agent-runtime/runtime-bridge.ts @@ -1,21 +1,12 @@ -// TODO(M-followup by 2026-05-20): decompose — current file exceeds 500 LOC -// ceiling (648 LOC). Split into sqlite-session-store wiring, cost/metrics -// routing, and hook-dispatch assembly. If 2026-05-20 passes without -// decomposition, a follow-up session MUST revisit and either decompose or -// justify the ceiling exception. import type { AgentMessage } from "@mariozechner/pi-agent-core"; import type { AppDatabase } from "@main/database"; import type { AgentMessageRecord, DesktopEvent } from "@shared/types"; -import { - ensureAgentCommunicationThread, - ensureTaskThread, -} from "@main/thread-integration/thread-helpers"; import { SqliteSessionStore } from "./sqlite-session-store"; import { CostWriter } from "./cost-writer"; import type { CredentialResolver } from "@main/credentials"; import { getConnectorDefinition } from "@main/connectors"; import { projectInboxMessageToPi } from "./inbox-projection"; -import { projectAgentEventToLogChunk, extractTextBody } from "./agent-event-projection"; +import { projectAgentEventToLogChunk } from "./agent-event-projection"; import { hasAttributableAssistantUsage } from "../run-usage-status"; import { captureBaselineIfRequired, reconcileWorkspaceArtifacts } from "./artifact-projector"; import { @@ -30,6 +21,8 @@ import type { } 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"; +import { RuntimeThreadProjector } from "./runtime-thread-projector"; /** * Side-effect callbacks the internal-tool handlers invoke after a @@ -91,59 +84,23 @@ type RpcMethod = WorkerRpcMethod; * Policy itself lives in `hook-handlers.ts`; this class is a pure IPC * router plus an `ActiveRunContext` cache. */ -interface InboundCorrelationEntry { - fromAgentId: string; - senderKind: AgentMessageRecord["senderKind"]; - messageId: string; - /** - * Stable pi-message timestamp that the projected user turn carries. - * Reverse-indexed so `message_end` (which only surfaces the pi - * message shape) can resolve back to `messageId` without relying on - * wall-clock millisecond precision. - */ - piTimestamp: number; -} - -interface InboundCorrelationRegistry { - byMessageId: Map; - byPiTimestamp: Map; - lastAllocatedTimestamp: number; -} - -/** - * Upper bound on inbound-correlation entries retained per active run. - * - * Every inbound peer message injected during a single pi run allocates - * one entry in both the messageId and piTimestamp maps of the registry. - * A long-lived CEO session that accumulates thousands of mid-run DMs - * would otherwise grow the maps without bound until the session - * compaction fires (which does not clean these structures). Keeping the - * `MAX_INBOUND_CORRELATIONS_PER_RUN` most recent entries is sufficient: - * `message_end` resolves author identity immediately after each turn, - * so the in-memory window only needs to cover pending-to-drain inbound - * messages, not the full run history. - */ -const MAX_INBOUND_CORRELATIONS_PER_RUN = 512; - export class RuntimeBridge { private readonly costWriter: CostWriter; + private readonly threadProjector: RuntimeThreadProjector; private readonly activeRuns = new Map(); private readonly sessionStores = new Map(); private internalToolCallbacks: InternalToolCallbacks | null = null; - /** - * Per-run inbound-message registry. Primary identity is - * `record.id` (the `agent_messages.id`). A monotonically allocated - * `piTimestamp` lives on every projected user turn so - * `message_end` — which only surfaces the pi-message shape — can - * invert the mapping back to the real author. The wall-clock - * `createdAt` is not enough: two messages that land in the same - * millisecond would collide on a timestamp-keyed map and one peer's - * identity would be silently lost in thread-event projection. - */ - private readonly inboundCorrelations = new Map(); + private readonly inboundCorrelations = new InboundCorrelationRegistryStore(); constructor(private readonly deps: RuntimeBridgeDeps) { this.costWriter = new CostWriter({ db: deps.db }); + this.threadProjector = new RuntimeThreadProjector({ + db: deps.db, + logger: deps.logger, + resolveAgentName: (agentId) => this.resolveAgentName(agentId), + lookupInboundCorrelation: (runId, timestamp) => + this.inboundCorrelations.lookup(runId, timestamp), + }); } /** @@ -181,7 +138,7 @@ export class RuntimeBridge { const ctx = this.activeRuns.get(runId); if (ctx) this.reconcileArtifactsForRun(ctx); this.activeRuns.delete(runId); - this.inboundCorrelations.delete(runId); + this.inboundCorrelations.deleteRun(runId); if (ctx) this.sessionStores.delete(ctx.taskSessionId); } @@ -224,53 +181,14 @@ export class RuntimeBridge { * entries when `message_end` fires. */ projectInboundMessage(runId: string, record: AgentMessageRecord): AgentMessage { - const registry = this.getOrCreateInboundRegistry(runId); const base = Date.parse(record.createdAt); const baseline = Number.isFinite(base) ? base : Date.now(); - const allocated = Math.max(baseline, registry.lastAllocatedTimestamp + 1); - registry.lastAllocatedTimestamp = allocated; - const entry: InboundCorrelationEntry = { - fromAgentId: record.fromAgentId, - senderKind: record.senderKind, - messageId: record.id, - piTimestamp: allocated, - }; - registry.byMessageId.set(record.id, entry); - registry.byPiTimestamp.set(allocated, record.id); - this.evictOldestCorrelationsIfNeeded(registry); + const allocated = this.inboundCorrelations.register(runId, record, baseline); return projectInboxMessageToPi(record, { getAgentName: (id) => this.resolveAgentName(id), }, { timestampOverride: allocated }); } - /** - * Bound the per-run correlation maps. JavaScript `Map` preserves - * insertion order, so dropping the head of each map retires the - * oldest entries. Eviction operates symmetrically on both maps so - * the reverse lookups remain consistent. - */ - private evictOldestCorrelationsIfNeeded(registry: InboundCorrelationRegistry): void { - while (registry.byMessageId.size > MAX_INBOUND_CORRELATIONS_PER_RUN) { - const oldestMessageId = registry.byMessageId.keys().next().value; - if (!oldestMessageId) break; - const entry = registry.byMessageId.get(oldestMessageId); - registry.byMessageId.delete(oldestMessageId); - if (entry) registry.byPiTimestamp.delete(entry.piTimestamp); - } - } - - private getOrCreateInboundRegistry(runId: string): InboundCorrelationRegistry { - const existing = this.inboundCorrelations.get(runId); - if (existing) return existing; - const registry: InboundCorrelationRegistry = { - byMessageId: new Map(), - byPiTimestamp: new Map(), - lastAllocatedTimestamp: 0, - }; - this.inboundCorrelations.set(runId, registry); - return registry; - } - /** Returns the active run context for a given agent, if any. Used by * main-side services (inbox dispatcher) to decide whether a new inbox * row should route to `agent.steer()` instead of waiting for the next @@ -385,168 +303,7 @@ export class RuntimeBridge { ); } } - this.recordThreadEvent(envelope, messageRaw); - } - } - - /** - * Project the pi `message_end` event into a `conversation_events` row - * so renderer thread views stay live. Resolves the thread from the run - * context: task runs write into the task thread; taskless runs — used - * by communication wakes — write into a per-agent general thread so - * peer conversations remain visible in the UI. - * - * User-role messages whose pi timestamp matches a registered inbound - * injection are attributed back to the original author via the - * correlation map; without that, the envelope's `agentId` (the - * recipient) would be recorded as the sender and every peer message - * would collapse into a self-reply row. - * - * Best-effort: thread persistence failures must not interrupt the run. - */ - private recordThreadEvent( - envelope: { runId: string; agentId: string; companyId: string; taskId: string | null }, - message: { role?: string; content?: unknown; toolCallId?: string; timestamp?: number } | undefined, - ): void { - if (!message?.role) return; - - const thread = this.resolveThreadForRun(envelope); - if (!thread) return; - - try { - const body = extractTextBody(message.content); - const actor = this.resolveThreadEventActor(envelope, message); - this.deps.db.threads.addThreadEvent({ - threadId: thread.id, - actorId: actor.actorId, - actorType: actor.actorType, - eventType: message.role === "toolResult" ? "system_note" : "message", - body: body.length > 4000 ? `${body.slice(0, 3997)}...` : body, - metadataJson: JSON.stringify({ - runId: envelope.runId, - role: message.role, - sourceMessageId: actor.sourceMessageId ?? undefined, - }), - }); - } catch (error) { - this.deps.logger.warn( - `[runtime-bridge] failed to project agent-event into thread ${thread.id}: ${ - error instanceof Error ? error.message : String(error) - }`, - ); - } - } - - private resolveThreadForRun( - envelope: { runId: string; agentId: string; companyId: string; taskId: string | null }, - ) { - if (envelope.taskId) { - try { - const task = this.deps.db.entityCrud.getTask(envelope.taskId); - return ensureTaskThread(this.deps.db, envelope.taskId, envelope.companyId, task.title); - } catch (error) { - this.deps.logger.warn( - `[runtime-bridge] getTask failed for ${envelope.taskId}: ${error instanceof Error ? error.message : String(error)}`, - ); - return null; - } - } - try { - const agentLabel = this.resolveAgentName(envelope.agentId); - return ensureAgentCommunicationThread( - this.deps.db, - envelope.agentId, - envelope.companyId, - agentLabel, - ); - } catch (error) { - this.deps.logger.warn( - `[runtime-bridge] ensureAgentCommunicationThread failed for ${envelope.agentId}: ${error instanceof Error ? error.message : String(error)}`, - ); - return null; - } - } - - private resolveThreadEventActor( - envelope: { runId: string; agentId: string }, - message: { role?: string; timestamp?: number }, - ): { actorId: string; actorType: "agent" | "user" | "system"; sourceMessageId: string | null } { - if (message.role === "user") { - const correlation = this.lookupInboundCorrelation(envelope.runId, message.timestamp); - if (correlation) { - const actorType: "agent" | "user" | "system" = - correlation.senderKind === "human" - ? "user" - : correlation.senderKind === "system" - ? "system" - : "agent"; - return { - actorId: correlation.fromAgentId, - actorType, - sourceMessageId: correlation.messageId, - }; - } - return { actorId: "user", actorType: "user", sourceMessageId: null }; - } - if (message.role === "assistant") { - return { actorId: envelope.agentId, actorType: "agent", sourceMessageId: null }; - } - return { actorId: envelope.agentId, actorType: "system", sourceMessageId: null }; - } - - private lookupInboundCorrelation( - runId: string, - timestamp: number | undefined, - ): InboundCorrelationEntry | null { - if (typeof timestamp !== "number" || !Number.isFinite(timestamp)) return null; - const registry = this.inboundCorrelations.get(runId); - if (!registry) return null; - const messageId = registry.byPiTimestamp.get(timestamp); - if (!messageId) return null; - return registry.byMessageId.get(messageId) ?? null; - } - - /** - * Write the conversation-event row for an inbound peer message that - * was projected onto the recipient's transcript as a prior-turn (via - * `prepareRun`). priorMessages never fire `message_end`, so this is - * the only path that surfaces them in the thread view. Mid-run steer - * injections still go through `recordThreadEvent` off `message_end` - * and resolve their author via `inboundCorrelations`. - */ - private recordInboundMessageAsThreadEvent( - envelope: { runId: string; agentId: string; companyId: string; taskId: string | null }, - record: AgentMessageRecord, - ): void { - const thread = this.resolveThreadForRun(envelope); - if (!thread) return; - try { - const body = record.body.length > 4000 ? `${record.body.slice(0, 3997)}...` : record.body; - const actorType: "agent" | "user" | "system" = - record.senderKind === "human" - ? "user" - : record.senderKind === "system" - ? "system" - : "agent"; - this.deps.db.threads.addThreadEvent({ - threadId: thread.id, - actorId: record.fromAgentId, - actorType, - eventType: "message", - body, - metadataJson: JSON.stringify({ - runId: envelope.runId, - role: "user", - sourceMessageId: record.id, - projection: "prepare_run", - }), - }); - } catch (error) { - this.deps.logger.warn( - `[runtime-bridge] inbound thread-event write failed for message ${record.id}: ${ - error instanceof Error ? error.message : String(error) - }`, - ); + this.threadProjector.recordThreadEvent(envelope, messageRaw); } } @@ -689,7 +446,7 @@ export class RuntimeBridge { // to distinct authors at `message_end`. const projected = this.projectInboundMessage(runId, record); store.appendMessage(projected); - this.recordInboundMessageAsThreadEvent( + this.threadProjector.recordInboundMessageAsThreadEvent( { runId, agentId: payload.agentId, diff --git a/src/main/agent-runtime/runtime-thread-projector.ts b/src/main/agent-runtime/runtime-thread-projector.ts new file mode 100644 index 00000000..c79eb357 --- /dev/null +++ b/src/main/agent-runtime/runtime-thread-projector.ts @@ -0,0 +1,170 @@ +import type { AppDatabase } from "@main/database"; +import type { AgentMessageRecord } from "@shared/types"; +import { + ensureAgentCommunicationThread, + ensureTaskThread, +} from "@main/thread-integration/thread-helpers"; +import { extractTextBody } from "./agent-event-projection"; +import type { InboundCorrelationEntry } from "./inbound-correlation-registry"; + +type RuntimeThreadEnvelope = { + runId: string; + agentId: string; + companyId: string; + taskId: string | null; +}; + +type RuntimeThreadMessage = { + role?: string; + content?: unknown; + toolCallId?: string; + timestamp?: number; +}; + +interface RuntimeThreadProjectorDeps { + db: AppDatabase; + logger: { + warn: (...args: unknown[]) => void; + }; + resolveAgentName: (agentId: string) => string; + lookupInboundCorrelation: ( + runId: string, + timestamp: number | undefined, + ) => InboundCorrelationEntry | null; +} + +export class RuntimeThreadProjector { + constructor(private readonly deps: RuntimeThreadProjectorDeps) {} + + /** + * Project the pi `message_end` event into a `conversation_events` row + * so renderer thread views stay live. + */ + recordThreadEvent( + envelope: RuntimeThreadEnvelope, + message: RuntimeThreadMessage | undefined, + ): void { + if (!message?.role) return; + + const thread = this.resolveThreadForRun(envelope); + if (!thread) return; + + try { + const body = extractTextBody(message.content); + const actor = this.resolveThreadEventActor(envelope, message); + this.deps.db.threads.addThreadEvent({ + threadId: thread.id, + actorId: actor.actorId, + actorType: actor.actorType, + eventType: message.role === "toolResult" ? "system_note" : "message", + body: body.length > 4000 ? `${body.slice(0, 3997)}...` : body, + metadataJson: JSON.stringify({ + runId: envelope.runId, + role: message.role, + sourceMessageId: actor.sourceMessageId ?? undefined, + }), + }); + } catch (error) { + this.deps.logger.warn( + `[runtime-bridge] failed to project agent-event into thread ${thread.id}: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + } + } + + /** + * Persist an inbound peer message projected as a prior turn during + * `prepareRun`; priorMessages do not emit `message_end`. + */ + recordInboundMessageAsThreadEvent( + envelope: RuntimeThreadEnvelope, + record: AgentMessageRecord, + ): void { + const thread = this.resolveThreadForRun(envelope); + if (!thread) return; + try { + const body = record.body.length > 4000 ? `${record.body.slice(0, 3997)}...` : record.body; + const actorType: "agent" | "user" | "system" = + record.senderKind === "human" + ? "user" + : record.senderKind === "system" + ? "system" + : "agent"; + this.deps.db.threads.addThreadEvent({ + threadId: thread.id, + actorId: record.fromAgentId, + actorType, + eventType: "message", + body, + metadataJson: JSON.stringify({ + runId: envelope.runId, + role: "user", + sourceMessageId: record.id, + projection: "prepare_run", + }), + }); + } catch (error) { + this.deps.logger.warn( + `[runtime-bridge] inbound thread-event write failed for message ${record.id}: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + } + } + + private resolveThreadForRun(envelope: RuntimeThreadEnvelope) { + if (envelope.taskId) { + try { + const task = this.deps.db.entityCrud.getTask(envelope.taskId); + return ensureTaskThread(this.deps.db, envelope.taskId, envelope.companyId, task.title); + } catch (error) { + this.deps.logger.warn( + `[runtime-bridge] getTask failed for ${envelope.taskId}: ${error instanceof Error ? error.message : String(error)}`, + ); + return null; + } + } + try { + const agentLabel = this.deps.resolveAgentName(envelope.agentId); + return ensureAgentCommunicationThread( + this.deps.db, + envelope.agentId, + envelope.companyId, + agentLabel, + ); + } catch (error) { + this.deps.logger.warn( + `[runtime-bridge] ensureAgentCommunicationThread failed for ${envelope.agentId}: ${error instanceof Error ? error.message : String(error)}`, + ); + return null; + } + } + + private resolveThreadEventActor( + envelope: { runId: string; agentId: string }, + message: { role?: string; timestamp?: number }, + ): { actorId: string; actorType: "agent" | "user" | "system"; sourceMessageId: string | null } { + if (message.role === "user") { + const correlation = this.deps.lookupInboundCorrelation(envelope.runId, message.timestamp); + if (correlation) { + const actorType: "agent" | "user" | "system" = + correlation.senderKind === "human" + ? "user" + : correlation.senderKind === "system" + ? "system" + : "agent"; + return { + actorId: correlation.fromAgentId, + actorType, + sourceMessageId: correlation.messageId, + }; + } + return { actorId: "user", actorType: "user", sourceMessageId: null }; + } + if (message.role === "assistant") { + return { actorId: envelope.agentId, actorType: "agent", sourceMessageId: null }; + } + return { actorId: envelope.agentId, actorType: "system", sourceMessageId: null }; + } +} diff --git a/src/main/database/documents-work-products.ts b/src/main/database/documents-work-products.ts new file mode 100644 index 00000000..92ec4884 --- /dev/null +++ b/src/main/database/documents-work-products.ts @@ -0,0 +1,311 @@ +import { randomUUID } from "node:crypto"; +import { statSync } from "node:fs"; +import { isAbsolute, resolve as resolvePath, relative as relativePath } from "node:path"; +import type { DatabaseSync, SQLInputValue } from "node:sqlite"; +import type { ReadAll } from "./helpers"; +import { nowIso } from "./helpers"; + +function resolveWorkspaceRoot( + getDb: () => DatabaseSync, + companyId: string, + workspaceId: string | null, + taskId: string | null, + runId: string | null, +): string | null { + if (workspaceId) { + const row = getDb().prepare( + "SELECT local_path AS localPath FROM workspaces WHERE id = ? AND company_id = ?", + ).get(workspaceId, companyId) as { localPath: string } | undefined; + if (row?.localPath) return row.localPath; + } + if (taskId) { + const row = getDb().prepare( + "SELECT w.local_path AS localPath FROM tasks t JOIN workspaces w ON w.id = t.workspace_id WHERE t.id = ? AND t.company_id = ?", + ).get(taskId, companyId) as { localPath: string } | undefined; + if (row?.localPath) return row.localPath; + } + if (runId) { + const row = getDb().prepare( + "SELECT w.local_path AS localPath FROM runs r JOIN workspaces w ON w.id = r.workspace_id WHERE r.id = ? AND r.company_id = ?", + ).get(runId, companyId) as { localPath: string } | undefined; + if (row?.localPath) return row.localPath; + } + return null; +} + +function joinWorkspaceRelative(root: string, rel: string): string | null { + if (isAbsolute(rel)) return null; + const joined = resolvePath(root, rel); + const within = relativePath(root, joined); + if (within.startsWith("..") || isAbsolute(within)) return null; + return joined; +} + +export function buildWorkProductMethods( + getDb: () => DatabaseSync, + readAll: ReadAll, +) { + return { + createWorkProduct(input: { + companyId: string; taskId: string | null; projectId?: string | null; + workspaceId?: string | null; + type: string; provider: string; externalId?: string | null; + title: string; url?: string | null; status?: string; reviewState?: string; + isPrimary?: boolean; summary?: string | null; metadataJson?: string | null; + createdByRunId?: string | null; + }): string { + const id = randomUUID(); + const ts = nowIso(); + if (input.isPrimary && input.taskId) { + getDb().prepare( + "UPDATE work_products SET is_primary = 0, updated_at = ? WHERE company_id = ? AND task_id = ? AND type = ?", + ).run(ts, input.companyId, input.taskId, input.type); + } + getDb().prepare( + `INSERT INTO work_products (id, company_id, project_id, task_id, workspace_id, type, provider, external_id, title, url, status, review_state, is_primary, summary, metadata_json, created_by_run_id, last_touched_by_run_id, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ).run( + id, input.companyId, input.projectId ?? null, input.taskId ?? null, + input.workspaceId ?? null, + input.type, input.provider, input.externalId ?? null, + input.title, input.url ?? null, input.status ?? "active", + input.reviewState ?? "none", input.isPrimary ? 1 : 0, + input.summary ?? null, input.metadataJson ?? null, + input.createdByRunId ?? null, input.createdByRunId ?? null, ts, ts, + ); + return id; + }, + + findWorkProductByExternalId( + companyId: string, + provider: string, + externalId: string, + workspaceId: string | null, + ): { id: string; taskId: string | null } | null { + if (workspaceId) { + const row = getDb() + .prepare( + "SELECT id, task_id as taskId FROM work_products WHERE company_id = ? AND workspace_id = ? AND provider = ? AND external_id = ? LIMIT 1", + ) + .get(companyId, workspaceId, provider, externalId) as + | { id: string; taskId: string | null } + | undefined; + return row ?? null; + } + const row = getDb() + .prepare( + "SELECT id, task_id as taskId FROM work_products WHERE company_id = ? AND workspace_id IS NULL AND provider = ? AND external_id = ? LIMIT 1", + ) + .get(companyId, provider, externalId) as + | { id: string; taskId: string | null } + | undefined; + return row ?? null; + }, + + updateWorkProduct( + id: string, + companyId: string, + patch: { + type?: string; + provider?: string; + externalId?: string | null; + workspaceId?: string | null; + title?: string; + url?: string | null; + status?: string; + reviewState?: string; + isPrimary?: boolean; + summary?: string | null; + metadataJson?: string | null; + lastTouchedByRunId?: string | null; + }, + ): number { + const ts = nowIso(); + const sets: string[] = ["updated_at = ?"]; + const values: SQLInputValue[] = [ts]; + const map: Array<[keyof typeof patch, string, (v: unknown) => SQLInputValue]> = [ + ["type", "type", (v) => (v as string) ?? null], + ["provider", "provider", (v) => (v as string) ?? null], + ["externalId", "external_id", (v) => (v as string | null) ?? null], + ["workspaceId", "workspace_id", (v) => (v as string | null) ?? null], + ["title", "title", (v) => (v as string) ?? null], + ["url", "url", (v) => (v as string | null) ?? null], + ["status", "status", (v) => (v as string) ?? null], + ["reviewState", "review_state", (v) => (v as string) ?? null], + ["isPrimary", "is_primary", (v) => (v ? 1 : 0)], + ["summary", "summary", (v) => (v as string | null) ?? null], + ["metadataJson", "metadata_json", (v) => (v as string | null) ?? null], + ["lastTouchedByRunId", "last_touched_by_run_id", (v) => (v as string | null) ?? null], + ]; + for (const [key, col, coerce] of map) { + if (key in patch) { + sets.push(`${col} = ?`); + values.push(coerce(patch[key])); + } + } + values.push(id, companyId); + const result = getDb() + .prepare(`UPDATE work_products SET ${sets.join(", ")} WHERE id = ? AND company_id = ?`) + .run(...values); + return Number(result.changes ?? 0); + }, + + resolveWorkProductTarget( + workProductId: string, + companyId: string, + ): + | { kind: "url" | "path"; value: string } + | { kind: "error"; reason: "not_found" | "archived" | "no_target" | "workspace_unresolved" | "path_escape"; detail: string } { + const row = getDb().prepare( + "SELECT provider, external_id, url, status, task_id, created_by_run_id, last_touched_by_run_id, workspace_id FROM work_products WHERE id = ? AND company_id = ?", + ).get(workProductId, companyId) as + | { provider: string; external_id: string | null; url: string | null; status: string; task_id: string | null; created_by_run_id: string | null; last_touched_by_run_id: string | null; workspace_id: string | null } + | undefined; + if (!row) return { kind: "error", reason: "not_found", detail: `No work_product row with id=${workProductId}` }; + if (row.status === "archived") return { kind: "error", reason: "archived", detail: "Work product is archived (retirement tombstone)" }; + + if (row.url && row.url.length > 0) return { kind: "url", value: row.url }; + if (row.provider !== "filesystem" || !row.external_id) { + return { kind: "error", reason: "no_target", detail: `provider=${row.provider} external_id=${row.external_id ?? "null"} url=null` }; + } + + const workspaceRoot = resolveWorkspaceRoot( + getDb, + companyId, + row.workspace_id, + row.task_id, + row.last_touched_by_run_id ?? row.created_by_run_id, + ); + if (workspaceRoot) { + const absolute = joinWorkspaceRelative(workspaceRoot, row.external_id); + if (!absolute) { + return { kind: "error", reason: "path_escape", detail: `external_id=${row.external_id} escapes workspace root ${workspaceRoot}` }; + } + return { kind: "path", value: absolute }; + } + + const observation = getDb() + .prepare( + "SELECT local_path as localPath FROM workspaces WHERE company_id = ? AND local_path IS NOT NULL ORDER BY is_primary DESC, created_at ASC LIMIT 1", + ) + .get(companyId) as { localPath: string } | undefined; + if (observation?.localPath) { + const candidate = joinWorkspaceRelative(observation.localPath, row.external_id); + if (candidate) { + try { + if (statSync(candidate).isFile()) { + return { kind: "path", value: candidate }; + } + } catch { + // fall through to error below + } + } + } + return { + kind: "error", + reason: "workspace_unresolved", + detail: `workspace_id=${row.workspace_id ?? "null"} task_id=${row.task_id ?? "null"} run_id=${row.last_touched_by_run_id ?? row.created_by_run_id ?? "null"} — none led to a workspace with a local_path, and ${row.external_id} is not present in the company's observation workspace${observation?.localPath ? ` (${observation.localPath})` : " (no observation workspace exists)"}`, + }; + }, + + listWorkProductsForTask(taskId: string, companyId: string) { + return readAll<{ + id: string; company_id: string; project_id: string | null; task_id: string; + workspace_id: string | null; + type: string; provider: string; external_id: string | null; title: string; + url: string | null; status: string; review_state: string; is_primary: number; + summary: string | null; metadata_json: string | null; + created_by_run_id: string | null; last_touched_by_run_id: string | null; + created_at: string; updated_at: string; + }>( + "SELECT * FROM work_products WHERE task_id = ? AND company_id = ? AND status <> 'archived' ORDER BY is_primary DESC, updated_at DESC", + taskId, companyId, + ).map((r) => ({ + id: r.id, companyId: r.company_id, projectId: r.project_id, taskId: r.task_id, + workspaceId: r.workspace_id, + type: r.type, provider: r.provider, externalId: r.external_id, title: r.title, + url: r.url, status: r.status, reviewState: r.review_state, + isPrimary: Boolean(r.is_primary), summary: r.summary, + metadataJson: r.metadata_json, createdByRunId: r.created_by_run_id, + lastTouchedByRunId: r.last_touched_by_run_id, + createdAt: r.created_at, updatedAt: r.updated_at, + })); + }, + + listWorkProductsForCompany(companyId: string, limit = 500) { + return readAll<{ + id: string; company_id: string; project_id: string | null; task_id: string | null; + workspace_id: string | null; + type: string; provider: string; external_id: string | null; title: string; + url: string | null; status: string; review_state: string; is_primary: number; + summary: string | null; metadata_json: string | null; + created_by_run_id: string | null; last_touched_by_run_id: string | null; + created_at: string; updated_at: string; + }>( + "SELECT * FROM work_products WHERE company_id = ? AND status <> 'archived' ORDER BY updated_at DESC LIMIT ?", + companyId, limit, + ).map((r) => ({ + id: r.id, companyId: r.company_id, projectId: r.project_id, taskId: r.task_id, + workspaceId: r.workspace_id, + type: r.type, provider: r.provider, externalId: r.external_id, title: r.title, + url: r.url, status: r.status, reviewState: r.review_state, + isPrimary: Boolean(r.is_primary), summary: r.summary, + metadataJson: r.metadata_json, createdByRunId: r.created_by_run_id, + lastTouchedByRunId: r.last_touched_by_run_id, + createdAt: r.created_at, updatedAt: r.updated_at, + })); + }, + + deleteWorkProduct(id: string, companyId: string): number { + const result = getDb() + .prepare("DELETE FROM work_products WHERE id = ? AND company_id = ?") + .run(id, companyId); + return Number(result.changes ?? 0); + }, + + archiveWorkProductsByExternalPaths(input: { + companyId: string; + workspaceId: string | null; + provider: string; + paths: string[]; + lastTouchedByRunId: string | null; + }): Array<{ id: string; externalId: string }> { + if (input.paths.length === 0) return []; + const ts = nowIso(); + const CHUNK = 500; + const workspaceClause = input.workspaceId + ? "AND workspace_id = ?" + : "AND workspace_id IS NULL"; + const results: Array<{ id: string; externalId: string }> = []; + for (let i = 0; i < input.paths.length; i += CHUNK) { + const chunk = input.paths.slice(i, i + CHUNK); + const placeholders = chunk.map(() => "?").join(", "); + const params: SQLInputValue[] = [ + input.lastTouchedByRunId ?? null, + ts, + input.companyId, + input.provider, + ]; + if (input.workspaceId) params.push(input.workspaceId); + params.push(...chunk); + const rows = readAll<{ id: string; external_id: string }>( + `UPDATE work_products + SET status = 'archived', + last_touched_by_run_id = ?, + updated_at = ? + WHERE company_id = ? + AND provider = ? + ${workspaceClause} + AND external_id IN (${placeholders}) + AND status <> 'archived' + RETURNING id, external_id`, + ...params, + ); + for (const row of rows) { + results.push({ id: row.id, externalId: row.external_id }); + } + } + return results; + }, + }; +} diff --git a/src/main/database/documents.ts b/src/main/database/documents.ts index dedd4122..6f40f7a5 100644 --- a/src/main/database/documents.ts +++ b/src/main/database/documents.ts @@ -1,52 +1,9 @@ import { randomUUID } from "node:crypto"; -import { statSync } from "node:fs"; -import { isAbsolute, resolve as resolvePath, relative as relativePath } from "node:path"; -import type { DatabaseSync, SQLInputValue } from "node:sqlite"; +import type { DatabaseSync } from "node:sqlite"; import type { ReadAll, Transaction } from "./helpers"; import { ensureId, nowIso } from "./helpers"; import { buildConfigRevisionMethods } from "./config-revisions"; - -function resolveWorkspaceRoot( - getDb: () => DatabaseSync, - companyId: string, - workspaceId: string | null, - taskId: string | null, - runId: string | null, -): string | null { - if (workspaceId) { - const row = getDb().prepare( - "SELECT local_path AS localPath FROM workspaces WHERE id = ? AND company_id = ?", - ).get(workspaceId, companyId) as { localPath: string } | undefined; - if (row?.localPath) return row.localPath; - } - if (taskId) { - const row = getDb().prepare( - "SELECT w.local_path AS localPath FROM tasks t JOIN workspaces w ON w.id = t.workspace_id WHERE t.id = ? AND t.company_id = ?", - ).get(taskId, companyId) as { localPath: string } | undefined; - if (row?.localPath) return row.localPath; - } - if (runId) { - const row = getDb().prepare( - "SELECT w.local_path AS localPath FROM runs r JOIN workspaces w ON w.id = r.workspace_id WHERE r.id = ? AND r.company_id = ?", - ).get(runId, companyId) as { localPath: string } | undefined; - if (row?.localPath) return row.localPath; - } - return null; -} - -/** - * Joins a workspace root with a relative path, refusing paths that escape - * the workspace (parent-dir traversal or absolute substitution). Returns - * null when the relative fragment would land outside `root`; the caller - * treats that as "cannot resolve" and does not open anything. - */ -function joinWorkspaceRelative(root: string, rel: string): string | null { - if (isAbsolute(rel)) return null; - const joined = resolvePath(root, rel); - const within = relativePath(root, joined); - if (within.startsWith("..") || isAbsolute(within)) return null; - return joined; -} +import { buildWorkProductMethods } from "./documents-work-products"; /** * Document row shape returned by listDocuments — used by @@ -98,6 +55,7 @@ export function buildDocumentMethods( ) { return { ...buildConfigRevisionMethods(getDb, readAll, transaction), + ...buildWorkProductMethods(getDb, readAll), saveDocument(input: { id?: string | null; companyId: string; type: string; title: string; content?: string; authorAgentId?: string | null; reviewerAgentId?: string | null; projectId?: string | null; @@ -226,325 +184,6 @@ export function buildDocumentMethods( })); }, - // ── Work Products ── - - createWorkProduct(input: { - companyId: string; taskId: string | null; projectId?: string | null; - workspaceId?: string | null; - type: string; provider: string; externalId?: string | null; - title: string; url?: string | null; status?: string; reviewState?: string; - isPrimary?: boolean; summary?: string | null; metadataJson?: string | null; - createdByRunId?: string | null; - }): string { - const id = randomUUID(); - const ts = nowIso(); - if (input.isPrimary && input.taskId) { - getDb().prepare( - "UPDATE work_products SET is_primary = 0, updated_at = ? WHERE company_id = ? AND task_id = ? AND type = ?", - ).run(ts, input.companyId, input.taskId, input.type); - } - getDb().prepare( - `INSERT INTO work_products (id, company_id, project_id, task_id, workspace_id, type, provider, external_id, title, url, status, review_state, is_primary, summary, metadata_json, created_by_run_id, last_touched_by_run_id, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, - ).run( - id, input.companyId, input.projectId ?? null, input.taskId ?? null, - input.workspaceId ?? null, - input.type, input.provider, input.externalId ?? null, - input.title, input.url ?? null, input.status ?? "active", - input.reviewState ?? "none", input.isPrimary ? 1 : 0, - input.summary ?? null, input.metadataJson ?? null, - input.createdByRunId ?? null, input.createdByRunId ?? null, ts, ts, - ); - return id; - }, - - /** - * Look up a work product by its natural identity. The v45 schema - * pinned identity to `(companyId, workspaceId, provider, externalId)` - * so two workspaces with the same relative path no longer collide. - * When `workspaceId` is null we fall back to the legacy - * `(companyId, provider, externalId)` scope so historical rows that - * pre-date the workspace backfill can still be found and updated. - */ - findWorkProductByExternalId( - companyId: string, - provider: string, - externalId: string, - workspaceId: string | null, - ): { id: string; taskId: string | null } | null { - if (workspaceId) { - const row = getDb() - .prepare( - "SELECT id, task_id as taskId FROM work_products WHERE company_id = ? AND workspace_id = ? AND provider = ? AND external_id = ? LIMIT 1", - ) - .get(companyId, workspaceId, provider, externalId) as - | { id: string; taskId: string | null } - | undefined; - return row ?? null; - } - const row = getDb() - .prepare( - "SELECT id, task_id as taskId FROM work_products WHERE company_id = ? AND workspace_id IS NULL AND provider = ? AND external_id = ? LIMIT 1", - ) - .get(companyId, provider, externalId) as - | { id: string; taskId: string | null } - | undefined; - return row ?? null; - }, - - updateWorkProduct( - id: string, - companyId: string, - patch: { - type?: string; - provider?: string; - externalId?: string | null; - workspaceId?: string | null; - title?: string; - url?: string | null; - status?: string; - reviewState?: string; - isPrimary?: boolean; - summary?: string | null; - metadataJson?: string | null; - lastTouchedByRunId?: string | null; - }, - ): number { - const ts = nowIso(); - const sets: string[] = ["updated_at = ?"]; - const values: SQLInputValue[] = [ts]; - const map: Array<[keyof typeof patch, string, (v: unknown) => SQLInputValue]> = [ - ["type", "type", (v) => (v as string) ?? null], - ["provider", "provider", (v) => (v as string) ?? null], - ["externalId", "external_id", (v) => (v as string | null) ?? null], - ["workspaceId", "workspace_id", (v) => (v as string | null) ?? null], - ["title", "title", (v) => (v as string) ?? null], - ["url", "url", (v) => (v as string | null) ?? null], - ["status", "status", (v) => (v as string) ?? null], - ["reviewState", "review_state", (v) => (v as string) ?? null], - ["isPrimary", "is_primary", (v) => (v ? 1 : 0)], - ["summary", "summary", (v) => (v as string | null) ?? null], - ["metadataJson", "metadata_json", (v) => (v as string | null) ?? null], - ["lastTouchedByRunId", "last_touched_by_run_id", (v) => (v as string | null) ?? null], - ]; - for (const [key, col, coerce] of map) { - if (key in patch) { - sets.push(`${col} = ?`); - values.push(coerce(patch[key])); - } - } - values.push(id, companyId); - const result = getDb() - .prepare(`UPDATE work_products SET ${sets.join(", ")} WHERE id = ? AND company_id = ?`) - .run(...values); - return Number(result.changes ?? 0); - }, - - /** - * Resolves the absolute filesystem (or URL) target for a work product. - * - * Filesystem artifacts are stored as `external_id = ` to stay portable across workspace-root moves; `shell.openPath` - * needs an absolute managed path, so the renderer can't just forward - * the external id. This method joins the work-product row back to its - * originating task → workspace, or — for taskless runs — through the - * creating run, so the caller gets either a URL (non-filesystem - * artifact) or a concrete absolute path that downstream - * `canOpenPath` validates. - */ - resolveWorkProductTarget( - workProductId: string, - companyId: string, - ): - | { kind: "url" | "path"; value: string } - | { kind: "error"; reason: "not_found" | "archived" | "no_target" | "workspace_unresolved" | "path_escape"; detail: string } { - const row = getDb().prepare( - "SELECT provider, external_id, url, status, task_id, created_by_run_id, last_touched_by_run_id, workspace_id FROM work_products WHERE id = ? AND company_id = ?", - ).get(workProductId, companyId) as - | { provider: string; external_id: string | null; url: string | null; status: string; task_id: string | null; created_by_run_id: string | null; last_touched_by_run_id: string | null; workspace_id: string | null } - | undefined; - if (!row) return { kind: "error", reason: "not_found", detail: `No work_product row with id=${workProductId}` }; - if (row.status === "archived") return { kind: "error", reason: "archived", detail: "Work product is archived (retirement tombstone)" }; - - if (row.url && row.url.length > 0) { - return { kind: "url", value: row.url }; - } - if (row.provider !== "filesystem" || !row.external_id) { - return { kind: "error", reason: "no_target", detail: `provider=${row.provider} external_id=${row.external_id ?? "null"} url=null` }; - } - - const workspaceRoot = resolveWorkspaceRoot( - getDb, - companyId, - row.workspace_id, - row.task_id, - row.last_touched_by_run_id ?? row.created_by_run_id, - ); - if (workspaceRoot) { - const absolute = joinWorkspaceRelative(workspaceRoot, row.external_id); - if (!absolute) { - return { kind: "error", reason: "path_escape", detail: `external_id=${row.external_id} escapes workspace root ${workspaceRoot}` }; - } - return { kind: "path", value: absolute }; - } - - // Fourth-tier heal: orphan rows created before the projector tracked - // workspace_id end-to-end lose the link through all three joins. If - // the externalId resolves to a real file inside the company's - // observation (primary/oldest) workspace, we treat that as the - // canonical root for this row. No guessing across all workspaces — - // just one deterministic fallback, and it only fires when the file - // actually exists on disk so the shell never sees a phantom path. - const observation = getDb() - .prepare( - "SELECT local_path as localPath FROM workspaces WHERE company_id = ? AND local_path IS NOT NULL ORDER BY is_primary DESC, created_at ASC LIMIT 1", - ) - .get(companyId) as { localPath: string } | undefined; - if (observation?.localPath) { - const candidate = joinWorkspaceRelative(observation.localPath, row.external_id); - if (candidate) { - try { - if (statSync(candidate).isFile()) { - return { kind: "path", value: candidate }; - } - } catch { - // fall through to error below - } - } - } - return { - kind: "error", - reason: "workspace_unresolved", - detail: `workspace_id=${row.workspace_id ?? "null"} task_id=${row.task_id ?? "null"} run_id=${row.last_touched_by_run_id ?? row.created_by_run_id ?? "null"} — none led to a workspace with a local_path, and ${row.external_id} is not present in the company's observation workspace${observation?.localPath ? ` (${observation.localPath})` : " (no observation workspace exists)"}`, - }; - }, - - // Task-scoped visible listing. Archived rows stay in the table as - // retirement tombstones; the task API and the thread view only - // surface currently-active deliverables so a deleted file stops - // appearing in the task's artifact strip. - listWorkProductsForTask(taskId: string, companyId: string) { - return readAll<{ - id: string; company_id: string; project_id: string | null; task_id: string; - workspace_id: string | null; - type: string; provider: string; external_id: string | null; title: string; - url: string | null; status: string; review_state: string; is_primary: number; - summary: string | null; metadata_json: string | null; - created_by_run_id: string | null; last_touched_by_run_id: string | null; - created_at: string; updated_at: string; - }>( - "SELECT * FROM work_products WHERE task_id = ? AND company_id = ? AND status <> 'archived' ORDER BY is_primary DESC, updated_at DESC", - taskId, companyId, - ).map((r) => ({ - id: r.id, companyId: r.company_id, projectId: r.project_id, taskId: r.task_id, - workspaceId: r.workspace_id, - type: r.type, provider: r.provider, externalId: r.external_id, title: r.title, - url: r.url, status: r.status, reviewState: r.review_state, - isPrimary: Boolean(r.is_primary), summary: r.summary, - metadataJson: r.metadata_json, createdByRunId: r.created_by_run_id, - lastTouchedByRunId: r.last_touched_by_run_id, - createdAt: r.created_at, updatedAt: r.updated_at, - })); - }, - - /** - * Company-scoped work-product listing, newest first. Used by the - * renderer snapshot so the DocumentCenter can surface filesystem - * artifacts (type = "file") the agents produce via bash/write/edit. - * Archived rows are excluded so retired files stop showing in the - * operator UI; callers that need the full audit trail go through - * `listArchivedWorkProductsForCompany`. - */ - listWorkProductsForCompany(companyId: string, limit = 500) { - return readAll<{ - id: string; company_id: string; project_id: string | null; task_id: string | null; - workspace_id: string | null; - type: string; provider: string; external_id: string | null; title: string; - url: string | null; status: string; review_state: string; is_primary: number; - summary: string | null; metadata_json: string | null; - created_by_run_id: string | null; last_touched_by_run_id: string | null; - created_at: string; updated_at: string; - }>( - "SELECT * FROM work_products WHERE company_id = ? AND status <> 'archived' ORDER BY updated_at DESC LIMIT ?", - companyId, limit, - ).map((r) => ({ - id: r.id, companyId: r.company_id, projectId: r.project_id, taskId: r.task_id, - workspaceId: r.workspace_id, - type: r.type, provider: r.provider, externalId: r.external_id, title: r.title, - url: r.url, status: r.status, reviewState: r.review_state, - isPrimary: Boolean(r.is_primary), summary: r.summary, - metadataJson: r.metadata_json, createdByRunId: r.created_by_run_id, - lastTouchedByRunId: r.last_touched_by_run_id, - createdAt: r.created_at, updatedAt: r.updated_at, - })); - }, - - deleteWorkProduct(id: string, companyId: string): number { - const result = getDb() - .prepare("DELETE FROM work_products WHERE id = ? AND company_id = ?") - .run(id, companyId); - return Number(result.changes ?? 0); - }, - - /** - * Archive every `work_products` row whose natural identity matches - * one of `paths` under the given `(companyId, workspaceId, - * provider)` scope. The change detectors walk the whole workspace - * delta, so every detected deletion has to be retired — previously - * the projector capped the retire list at a hardcoded budget and - * the tail stayed visible in the DocumentCenter. - * - * Runs as a single `UPDATE … WHERE external_id IN (…)` per chunk - * so bulk deletions cost one SQL round-trip per 500 paths instead - * of one per path. `RETURNING` carries the archived row ids back - * to the projector so it can emit `retired` outcomes without a - * follow-up SELECT. - */ - archiveWorkProductsByExternalPaths(input: { - companyId: string; - workspaceId: string | null; - provider: string; - paths: string[]; - lastTouchedByRunId: string | null; - }): Array<{ id: string; externalId: string }> { - if (input.paths.length === 0) return []; - const ts = nowIso(); - const CHUNK = 500; - const workspaceClause = input.workspaceId - ? "AND workspace_id = ?" - : "AND workspace_id IS NULL"; - const results: Array<{ id: string; externalId: string }> = []; - for (let i = 0; i < input.paths.length; i += CHUNK) { - const chunk = input.paths.slice(i, i + CHUNK); - const placeholders = chunk.map(() => "?").join(", "); - const params: SQLInputValue[] = [ - input.lastTouchedByRunId ?? null, - ts, - input.companyId, - input.provider, - ]; - if (input.workspaceId) params.push(input.workspaceId); - params.push(...chunk); - const rows = readAll<{ id: string; external_id: string }>( - `UPDATE work_products - SET status = 'archived', - last_touched_by_run_id = ?, - updated_at = ? - WHERE company_id = ? - AND provider = ? - ${workspaceClause} - AND external_id IN (${placeholders}) - AND status <> 'archived' - RETURNING id, external_id`, - ...params, - ); - for (const row of rows) { - results.push({ id: row.id, externalId: row.external_id }); - } - } - return results; - }, - // ── Task-Document Links ── linkDocumentToTask(taskId: string, documentId: string, companyId: string, key: string): string {