diff --git a/doc/workspace-activity-query-performance.md b/doc/workspace-activity-query-performance.md new file mode 100644 index 00000000000..91286c19460 --- /dev/null +++ b/doc/workspace-activity-query-performance.md @@ -0,0 +1,157 @@ +# Workspace activity query performance + +The workspace activity query reads three compatibility keys from each retained +run context. Wide contexts made those separate JSON extractions expensive. The +same scope also materialized complete run rows for cleanup checks that consume +only six lifecycle fields. + +The implementation extracts `issueId`, `taskId`, and `executionWorkspaceId` +together through `jsonb_to_record`, after native task bindings and bound +checkout/execution-run pointers have had a chance to match. Strict JSONPath +passes only a top-level object to the record function. Arrays, scalars and JSON +null therefore retain their former unmatched legacy-key behavior; their native +bindings remain independent. PostgreSQL documents the +[strict path and type semantics](https://www.postgresql.org/docs/17/functions-json.html#FUNCTIONS-SQLJSON-PATH-MODES). + +The full tree's textual IDs are materialized once for the decoder's membership +checks. This also prevents recursive cardinality estimates from inflating the +correlated decoder's startup cost enough to trigger expensive JIT compilation. +The key set has no result limit. Empty trees produce an empty set. + +## Contract and blast radius + +The safety fact is that the same runs supply the same lifecycle fields. Native +and legacy bindings remain independent, including conflicting bindings. The +run scope retains descendants, tasks bound outside the tree, explicit bound +checkout/execution pointers, exact textual UUID comparisons, company scope, +duplicates and null behavior. + +| Consumer | Fields read from the run scope | Consequence of a missing run | +| --- | --- | --- | +| `gitWorkspaceLastActivityAt` | Created, started and finished timestamps | Expiration may be scheduled too early. | +| `isGitWorkspaceExpirationEligible` | ID, status, all three timestamps, controller lease expiry | A live run or its persistence/cleanup obligation could stop vetoing deletion. | +| Task-directory `orphanConsumer` | ID, status, finish time, controller lease expiry | An orphan folder could be removed while a run, finalization or lease still needs it. | +| `workspaceMoveEligible` | ID, status, finish time, controller lease expiry | Re-home could move a workspace used by a live controller. | + +All six fields remain projected. The task, recovery, finalization, environment +lease and runtime-service checks keep their existing mode-specific rules. +Deletion still rechecks current eligibility after the `beforeDelete` boundary, +under the existing source/lifecycle locks. Shared-path rejection still precedes +run-history evaluation. + +The deterministic database tests exercise nine independent binding paths, +downstream pending-cleanup leases referenced only through a run ID, conflicting +native/legacy bindings, multiply bound runs, case-sensitive legacy IDs, +cross-company decoys, current controller leases, new consumers between reads, +and null/scalar/array/nested-array contexts with and without native bindings. +Fixture insertion disables the scope-normalization trigger in disposable +databases so legacy-path tests cannot silently become native-binding tests. + +The benchmark compares complete positional caller results and the complete +multiset of all six relevant-run fields. This is evidence level 4 from the +blast-radius workflow: real code and real PostgreSQL. Deployed behavior and +production latency still require a rollout and fresh measurements. + +## Reproduce the measurements + +The baseline is `db70c5acd92f50fa37fb4f0dea3da0e7685089a2`. The harness loads +that service source with its unchanged current imports and captures the actual +SQL generated by the baseline and checked-out service. Reports save source +snapshots, SHA-256 digests, the source diff, all samples, complete SQL/parameters, +database settings, and `EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON)` plans. + +The Docker option creates and removes its own disposable PostgreSQL 17 +container, applies normal migrations, and uses synthetic data. It does not use +`DATABASE_URL`. Without that option the harness uses the normal embedded +PostgreSQL test helper. + +```sh +pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-workspace-activity.ts \ + --baseline-ref=db70c5acd92f50fa37fb4f0dea3da0e7685089a2 \ + --docker-postgres=17 --reject-new-jit --output-dir=/tmp/workspace-activity-stress +``` + +Run the other controls with `--small`, `--descendants=250`, and +`--small --descendants=250`, using separate output directories. Defaults are two +warmups and five measured samples per version, with baseline/treatment order +alternating. `--samples=1 --warmups=0` is diagnostic only. `--reject-new-jit` +saves non-executing preflight plans and stops before repeating a newly expensive +JIT plan. + +Each timing sample includes eight sequential workspace queries. The stress +fixture has 4,813 runs; the small control has 141. Every wide run has roughly +32 KiB of incompressible context padding and 32 KiB of result padding, with a +unique context nonce. The large-tree controls have 250 descendants per +workspace. Statistics are refreshed before timing. + +The accepted source SHA-256 is +`cd784320265af7857b534d705dbdbe09e732014f2be160971dd89e6d6219029a`. +All four reports have that digest. These measurements used PostgreSQL 17.11 on +Linux/aarch64 in Docker, with `work_mem=4MB`. The run relation including indexes +and TOAST was about 331.8 MB for stress and 9.15 MB for the small controls. + +| Fixture | Query | Before median, ms/8 | After median, ms/8 | Before / after | +| --- | --- | ---: | ---: | ---: | +| Stress, 1 descendant | Last activity | 1,071.54 | 743.45 | 1.44x | +| Stress, 1 descendant | Closed eligibility | 814.29 | 713.86 | 1.14x | +| Stress, 1 descendant | Idle eligibility | 721.46 | 481.69 | 1.50x | +| Small, 1 descendant | Last activity | 20.53 | 15.06 | 1.36x | +| Small, 1 descendant | Closed eligibility | 21.85 | 16.64 | 1.31x | +| Small, 1 descendant | Idle eligibility | 21.91 | 17.74 | 1.23x | +| Stress, 250 descendants | Last activity | 740.35 | 582.55 | 1.27x | +| Stress, 250 descendants | Closed eligibility | 678.95 | 549.16 | 1.24x | +| Stress, 250 descendants | Idle eligibility | 948.03 | 728.78 | 1.30x | +| Small, 250 descendants | Last activity | 45.51 | 38.09 | 1.19x | +| Small, 250 descendants | Closed eligibility | 61.11 | 62.82 | 0.97x, neutral | +| Small, 250 descendants | Idle eligibility | 46.28 | 38.02 | 1.22x | + +The stress last-activity samples did not overlap: baseline 920.27–1,111.22 ms +and treatment 689.14–880.40 ms per eight workspaces. Closed-eligibility timings +were noisier; its stress range was 649.58–2,138.92 ms versus 546.43–842.67 ms. +Treat that smaller median gain as directional. The combined closed-eligibility +control also had overlapping ranges, 59.01–82.96 versus 48.41–121.52 ms; its +1.72 ms median difference per eight workspaces is neutral evidence. + +The plans support the mechanism independently of timing noise. Stress +last-activity buffer accesses fell from 113,202 to 38,081. The decoder ran +4,752 times for 4,812 company rows, retaining native short-circuit behavior. +Materialized eligibility run width fell from about 1,414 to 57 planner bytes. +All final plans retained one history scan without JIT. Every measured caller +result and complete relevant-run multiset matched the baseline. + +## Rejected query shapes + +These failures informed the final shape and are not accepted performance results: + +| Candidate | Decisive result | +| --- | --- | +| Split binding joins without parameterization | Legacy hash joins caused additional wide-context scans; roughly twice the baseline SQL time. | +| Per-tree-node indexed probes | A 250-descendant fixture activated JIT and became roughly 5–8 times slower. | +| Aggregate-key-set per-node probes | Removed JIT, but small history plus 250 descendants chose one sequential scan per node and became 43–56 times slower. | +| Separate `ANY(ARRAY(...))` binding queries | Cleared the small/large-tree regression but performed separate history scans on the stress fixture and became roughly 1.7–1.8 times slower. | +| Correlated record decoder reading the recursive tree directly | A non-executing preflight found new JIT with estimated cost 156,621; rejected before repeated timing. | + +The final query retains one history scan and shares the materialized membership +key set. It changes neither scheduling frequency nor retention policy. Batching +the advisory last-activity reads is deferred so this change keeps the existing +reaper lifecycle and destructive freshness checks intact. + +## Before merging or rolling out + +Run the binding/eligibility tests and the existing Git expiration, directory +retention and repository re-home suites. They cover actual filesystem removal, +restoration, unique work, unresolved consumers, reopening and concurrent changes. + +```sh +pnpm --filter @paperclipai/server exec vitest run \ + src/__tests__/git-workspace-activity.test.ts \ + src/__tests__/git-workspace-expiration-eligibility.test.ts \ + src/__tests__/git-workspace-expiration.test.ts \ + src/__tests__/isolated-task-directory-retention.test.ts \ + src/__tests__/isolated-repository-rehome.test.ts +``` + +Also complete the repository's required typecheck, full tests and build on the +final source. After an authorized rollout, confirm the deployed revision and +compare fresh query-statistics deltas and cleanup behavior. Local synthetic SQL +measurements do not establish a production latency improvement. diff --git a/scripts/benchmark-workspace-activity.ts b/scripts/benchmark-workspace-activity.ts new file mode 100644 index 00000000000..fad3007ab1d --- /dev/null +++ b/scripts/benchmark-workspace-activity.ts @@ -0,0 +1,202 @@ +import assert from "node:assert/strict"; +import { createHash, randomBytes, randomUUID } from "node:crypto"; +import { execFileSync } from "node:child_process"; +import { promises as fs } from "node:fs"; +import { createRequire } from "node:module"; +import os from "node:os"; +import path from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { agents, applyPendingMigrations, companies, createDb, executionWorkspaces, issues, startEmbeddedPostgresTestDatabase, type Db } from "../packages/db/src/index.js"; +import * as expiration from "../server/src/services/git-workspace-expiration.js"; + +type Query = { sql: string; params: unknown[] }; +type Workspace = typeof executionWorkspaces.$inferSelect; +type Plan = { "Node Type": string; "Relation Name"?: string; "Index Name"?: string; "Actual Rows"?: number; "Actual Loops"?: number; "Shared Hit Blocks"?: number; "Shared Read Blocks"?: number; "Temp Read Blocks"?: number; "Temp Written Blocks"?: number; Plans?: Plan[] }; +const baselineRef = process.argv.find(arg => arg.startsWith("--baseline-ref="))?.slice(15); +assert.ok(baselineRef, "Pass an explicit pre-fix --baseline-ref="); +const repositoryRoot = fileURLToPath(new URL("../", import.meta.url)); +const gitRaw = (...args: string[]) => execFileSync("git", args, { encoding: "utf8", cwd: repositoryRoot }); +const git = (...args: string[]) => gitRaw(...args).trim(); +const baselineRevision = git("rev-parse", "--verify", `${baselineRef}^{commit}`); +const modulePath = "server/src/services/git-workspace-expiration.ts"; +const treatmentSource = await fs.readFile(path.join(repositoryRoot, modulePath), "utf8"); +const sha256 = (source: string) => createHash("sha256").update(source).digest("hex"); +const baselineOnly = process.argv.includes("--baseline-only"); +const dockerPostgres = process.argv.includes("--docker-postgres=17"); +const small = process.argv.includes("--small"); +const runsPerWorkspace = small ? 16 : 600; +const bindingRows = Math.max(1, Math.floor(runsPerWorkspace / 20)); +const descendantsPerWorkspace = Number(process.argv.find(arg => arg.startsWith("--descendants="))?.slice(14) ?? 1); +assert.ok(Number.isInteger(descendantsPerWorkspace) && descendantsPerWorkspace >= 1 && descendantsPerWorkspace <= 500, + "--descendants must be an integer between 1 and 500"); +const measuredSamples = Number(process.argv.find(arg => arg.startsWith("--samples="))?.slice(10) ?? 5); +const warmups = Number(process.argv.find(arg => arg.startsWith("--warmups="))?.slice(10) ?? 2); +assert.ok(Number.isInteger(measuredSamples) && measuredSamples >= 1 && measuredSamples <= 10); +assert.ok(Number.isInteger(warmups) && warmups >= 0 && warmups <= 5); +const outputDir = process.argv.find(arg => arg.startsWith("--output-dir="))?.slice(13) ?? await fs.mkdtemp(path.join(os.tmpdir(), "workspace-activity-evidence-")); +const require = createRequire(new URL("../server/package.json", import.meta.url)); +const { sql } = require("drizzle-orm"); +const { PgDialect } = require("drizzle-orm/pg-core"); +const dialect = new PgDialect(); +const now = new Date("2026-10-20T12:00:00Z"); +const old = new Date("2026-10-01T12:00:00Z"); +const keepAlive = setInterval(() => {}, 1000); +let database: Awaited> | undefined; +let db: Db | undefined; +let baselineDir: string | undefined; + +async function startDatabase() { + if (!dockerPostgres) return startEmbeddedPostgresTestDatabase("workspace-activity-benchmark-"); + const container = `paperclip-workspace-activity-${randomUUID()}`; + const docker = (...args: string[]) => execFileSync("docker", args, { encoding: "utf8", stdio: ["ignore", "pipe", "pipe"] }).trim(); + docker("run", "--detach", "--rm", "--name", container, "--publish", "127.0.0.1::5432", "--env", "POSTGRES_USER=paperclip", "--env", "POSTGRES_PASSWORD=paperclip", "--env", "POSTGRES_DB=paperclip", "postgres:17"); + const cleanup = async () => { docker("rm", "--force", container); }; + try { + let ready = false; + for (let attempt = 0; attempt < 100; attempt++) { + try { docker("exec", container, "pg_isready", "-h", "127.0.0.1", "-U", "paperclip"); ready = true; break; } catch { await new Promise(resolve => setTimeout(resolve, 100)); } + } + assert.ok(ready, "Disposable Docker PostgreSQL did not become ready"); + const port = docker("port", container, "5432/tcp").split(":").at(-1); + const connectionString = `postgres://paperclip:paperclip@127.0.0.1:${port}/paperclip`; + await applyPendingMigrations(connectionString); + return { connectionString, cleanup }; + } catch (error) { await cleanup(); throw error; } +} + +function capture(invoke: (db: Db) => Promise): Promise { + let captured: Query | undefined; + const captureDb = new Proxy(db!, { get(target, key) { + if (key === "execute") return async (query: Parameters[0]) => { captured = dialect.sqlToQuery(query); return []; }; + return Reflect.get(target, key); + }}); + return invoke(captureDb).then(() => { assert.ok(captured); return captured; }); +} +const normalize = (rows: unknown[][]) => rows.map(row => JSON.stringify(row)).sort(); +const summarize = (samples: number[]) => { + const sorted = [...samples].sort((a, b) => a - b); + return { samples, medianMs: sorted[Math.floor(sorted.length / 2)], minMs: sorted[0], maxMs: sorted.at(-1) }; +}; +function scans(plan: Plan): unknown[] { + const flatten = (node: Plan): Plan[] => [node, ...(node.Plans ?? []).flatMap(flatten)]; + return flatten(plan).filter(node => node["Relation Name"] === "heartbeat_runs" || node["Index Name"]) + .map(({ Plans: _children, ...node }) => node); +} + +try { + await fs.mkdir(outputDir, { recursive: true }); + database = await startDatabase(); + db = createDb(database.connectionString, { maxConnections: 1 }); + // Keep historical legacy bindings independent from native bindings; current + // insert triggers would otherwise rewrite the fixture's issue_id from JSON. + await db.$client.unsafe("ALTER TABLE heartbeat_runs DISABLE TRIGGER heartbeat_runs_set_scope_kind"); + baselineDir = await fs.mkdtemp(path.join(fileURLToPath(new URL("../server/scripts/", import.meta.url)), ".workspace-activity-baseline-")); + const original = new URL(`../${modulePath}`, import.meta.url); + const source = gitRaw("show", `${baselineRevision}:${modulePath}`); + const rewritten = source.replace(/from "(\.{1,2}\/[^\"]+)"/g, + (_match, specifier: string) => `from ${JSON.stringify(fileURLToPath(new URL(specifier, original)))}`); + const baselinePath = path.join(baselineDir, "git-workspace-expiration.ts"); + await fs.writeFile(baselinePath, rewritten); + await fs.writeFile(path.join(outputDir, "baseline-git-workspace-expiration.ts"), source); + await fs.writeFile(path.join(outputDir, "treatment-git-workspace-expiration.ts"), treatmentSource); + const baseline = await import(pathToFileURL(baselinePath).href) as typeof expiration; + const companyId = randomUUID(), agentId = randomUUID(), otherCompanyId = randomUUID(), otherAgentId = randomUUID(), unrelatedIssueId = randomUUID(); + await db.insert(companies).values([{ id: companyId, name: "Synthetic workspace activity", issuePrefix: "WAB" }, { id: otherCompanyId, name: "Other tenant", issuePrefix: "WAO" }]); + await db.insert(agents).values([{ id: agentId, companyId, name: "Fixture" }, { id: otherAgentId, companyId: otherCompanyId, name: "Other fixture" }]); + await db.insert(issues).values({ id: unrelatedIssueId, companyId, title: "Unrelated", updatedAt: old }); + const workspaces: Workspace[] = []; + for (let n = 0; n < 8; n++) { + const issueId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: `Target ${n}`, status: "done", completedAt: old, updatedAt: old }); + const [workspace] = await db.insert(executionWorkspaces).values({ companyId, sourceIssueId: issueId, mode: "isolated_workspace", strategyType: "git_worktree", name: `Target ${n}`, cwd: `/tmp/synthetic-workspace-activity-${n}`, updatedAt: old }).returning(); + workspaces.push(workspace); + await db.insert(issues).values(Array.from({ length: descendantsPerWorkspace }, (_, child) => ({ companyId, parentId: issueId, + title: `Child ${n}/${child}`, status: "done", completedAt: old, updatedAt: old }))); + } + const padding = randomBytes(24 * 1024).toString("base64"); + // Most history is unrelated, as in the reported scan; contexts and results are + // incompressible and wide enough to live in TOAST. Existing indexes are retained. + for (let n = 0; n < 8; n++) { + const workspace = workspaces[n]; + await db.$client.unsafe(`INSERT INTO heartbeat_runs (company_id, agent_id, status, scope_kind, issue_id, native_issue_id, created_at, started_at, finished_at, context_snapshot, result_json) + SELECT $1::uuid, $2::uuid, 'succeeded', CASE WHEN i <= ${bindingRows} THEN 'issue' ELSE 'company' END, + CASE WHEN i <= ${bindingRows} THEN $3::uuid END, CASE WHEN i > ${bindingRows} AND i <= ${bindingRows * 2} THEN $3::uuid END, + $6::timestamptz + i * interval '1 second', $6::timestamptz + i * interval '2 seconds', $6::timestamptz + i * interval '3 seconds', + jsonb_build_object('issueId', CASE WHEN i > ${bindingRows * 2} AND i <= ${bindingRows * 3} THEN $3::text ELSE $5::text END, + 'taskId', CASE WHEN i > ${bindingRows * 3} AND i <= ${bindingRows * 4} THEN $3::text ELSE $5::text END, + 'executionWorkspaceId', CASE WHEN i > ${bindingRows * 4} AND i <= ${bindingRows * 5} THEN $4::text ELSE 'unrelated' END, + 'fixtureRun', ${n * runsPerWorkspace} + i, 'padding', $7::text), + jsonb_build_object('stdout', $7::text) + FROM generate_series(1, ${runsPerWorkspace}) i`, [companyId, agentId, workspace.sourceIssueId, workspace.id, unrelatedIssueId, old.toISOString(), padding]); + } + // Malformed JSON shapes and noncanonical IDs must retain ->> semantics, never + // throw on a cast or accidentally match another tenant's run. + const unusualContexts = [null, [], [{ issueId: workspaces[0].sourceIssueId }], [[{ issueId: workspaces[0].sourceIssueId }]], true, + 12, "string", { issueId: workspaces[0].sourceIssueId!.toUpperCase() }, { issueId: 12 }, { issueId: { value: workspaces[0].sourceIssueId } }, + { taskId: workspaces[0].sourceIssueId }, { executionWorkspaceId: workspaces[0].id }]; + for (const context of unusualContexts) { + await db.$client.unsafe(`INSERT INTO heartbeat_runs (company_id, agent_id, status, created_at, finished_at, context_snapshot) VALUES ($1::uuid, $2::uuid, 'succeeded', $3::timestamptz, $3::timestamptz, $4::jsonb)`, [companyId, agentId, old.toISOString(), JSON.stringify(context)]); + } + await db.$client.unsafe(`INSERT INTO heartbeat_runs (company_id, agent_id, status, created_at, context_snapshot) VALUES ($1::uuid, $2::uuid, 'running', $3::timestamptz, $4::jsonb)`, [otherCompanyId, otherAgentId, now.toISOString(), JSON.stringify({ issueId: workspaces[0].sourceIssueId, executionWorkspaceId: workspaces[0].id })]); + await db.$client.unsafe("ALTER TABLE heartbeat_runs ENABLE TRIGGER heartbeat_runs_set_scope_kind"); + await db.$client.unsafe("ANALYZE"); + const [settings] = await db.$client.unsafe("SELECT version(), current_setting('work_mem') AS work_mem, pg_total_relation_size('heartbeat_runs') AS total_run_bytes"); + const results = []; + for (const kind of ["lastActivity", "eligibility", "idleEligibility"] as const) { + const invoke = (implementation: typeof expiration, d: Db, workspace: Workspace) => kind === "lastActivity" + ? implementation.gitWorkspaceLastActivityAt(d, workspace) + : implementation.isGitWorkspaceExpirationEligible(d as never, workspace, now, kind === "idleEligibility" ? now : undefined); + const beforeQueries = await Promise.all(workspaces.map(workspace => capture(d => invoke(baseline, d, workspace)))); + const afterQueries = baselineOnly ? beforeQueries : await Promise.all(workspaces.map(workspace => capture(d => invoke(expiration, d, workspace)))); + await fs.writeFile(path.join(outputDir, `${kind}-queries.json`), JSON.stringify({ beforeQueries, afterQueries }, null, 2)); + if (process.argv.includes("--reject-new-jit")) { + const [beforePlan] = await db.$client.unsafe(`EXPLAIN (FORMAT JSON) ${beforeQueries[0].sql}`, beforeQueries[0].params as never[]); + const [afterPlan] = await db.$client.unsafe(`EXPLAIN (FORMAT JSON) ${afterQueries[0].sql}`, afterQueries[0].params as never[]); + await fs.writeFile(path.join(outputDir, `${kind}-preflight-plans.json`), JSON.stringify({ beforePlan, afterPlan }, null, 2)); + assert.ok(!afterPlan["QUERY PLAN"][0].JIT || beforePlan["QUERY PLAN"][0].JIT, + `${kind}: reject new JIT compilation before repeating a known expensive planner shape`); + } + const execute = async (queries: Query[]) => { + const rows = []; + const start = performance.now(); + for (const query of queries) rows.push(await db!.$client.unsafe(query.sql, query.params as never[]).values()); + return { ms: performance.now() - start, rows }; + }; + const beforeSamples: number[] = [], afterSamples: number[] = []; + for (let n = 0; n < (baselineOnly ? 1 : warmups + measuredSamples); n++) { + const order = n % 2 === 0 ? ["before", "after"] : ["after", "before"]; + const iteration: Record>> = {}; + for (const treatment of order) iteration[treatment] = await execute(treatment === "before" ? beforeQueries : afterQueries); + assert.deepEqual(iteration.after.rows, iteration.before.rows, `${kind}: complete ordered positional results differ`); + if (baselineOnly || n >= warmups) { beforeSamples.push(iteration.before.ms); afterSamples.push(iteration.after.ms); } + } + const [beforeExplain] = await db.$client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${beforeQueries[0].sql}`, beforeQueries[0].params as never[]); + const [afterExplain] = await db.$client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${afterQueries[0].sql}`, afterQueries[0].params as never[]); + const before = summarize(beforeSamples), after = summarize(afterSamples); + results.push({ kind, workspaceCount: workspaces.length, equalCompleteResults: true, before, after, speedup: before.medianMs / after.medianMs, + baselinePlan: beforeExplain["QUERY PLAN"], treatmentPlan: afterExplain["QUERY PLAN"] }); + await fs.writeFile(path.join(outputDir, "partial-results.json"), JSON.stringify(results, null, 2)); + } + for (const workspace of workspaces) { + const projection = sql`SELECT id, status, created_at, started_at, finished_at, controller_lease_expires_at FROM relevant_runs`; + const before = dialect.sqlToQuery(sql`${baseline.expirationScope(workspace)} ${projection}`); + const after = dialect.sqlToQuery(sql`${expiration.expirationScope(workspace)} ${projection}`); + const beforeRows = await db.$client.unsafe(before.sql, before.params as never[]).values(); + const afterRows = await db.$client.unsafe(after.sql, after.params as never[]).values(); + assert.deepEqual(normalize(afterRows), normalize(beforeRows), "Complete relevant-run multisets differ"); + } + assert.equal(await fs.readFile(path.join(repositoryRoot, modulePath), "utf8"), treatmentSource, "Treatment source changed during measurement"); + const report = { baselineRevision, treatmentRevision: git("rev-parse", "HEAD"), + baselineSourceSha256: sha256(source), treatmentSourceSha256: sha256(treatmentSource), + treatmentDiff: gitRaw("diff", "--", modulePath), baselineDependencies: "Unchanged current imports; only git-workspace-expiration.ts is loaded from baseline", + settings, fixture: { runRows: runsPerWorkspace * 8 + unusualContexts.length + 1, workspaceCount: 8, descendantsPerWorkspace, contextPaddingBytes: Buffer.byteLength(padding), synthetic: true, small }, + acceptanceCriterion: "Complete result equality and materially lower stress-fixture medians; report small-control overhead separately", + measurement: { baselineOnly, measuredSamples: baselineOnly ? 1 : measuredSamples, warmups: baselineOnly ? 0 : warmups, alternatingOrder: true }, + safety: { exactRelevantRunMultisets: true, completeOrderedCallerResults: true }, results }; + await fs.writeFile(path.join(outputDir, "report.json"), JSON.stringify(report, null, 2)); + console.log(JSON.stringify({ outputDir, baselineRevision, settings, safety: report.safety, results: results.map(({ baselinePlan, treatmentPlan, ...result }) => ({ ...result, + beforeScans: scans(baselinePlan[0].Plan), afterScans: scans(treatmentPlan[0].Plan) })) }, null, 2)); +} finally { + try { await db?.$client.end({ timeout: 0 }); await database?.cleanup(); if (baselineDir) await fs.rm(baselineDir, { recursive: true, force: true }); } + finally { clearInterval(keepAlive); } +} diff --git a/server/src/__tests__/git-workspace-activity.test.ts b/server/src/__tests__/git-workspace-activity.test.ts new file mode 100644 index 00000000000..cb677f477e9 --- /dev/null +++ b/server/src/__tests__/git-workspace-activity.test.ts @@ -0,0 +1,126 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { agents, companies, createDb, environmentLeases, executionWorkspaces, heartbeatRuns, issues } from "@paperclipai/db"; +import { expirationScope, gitWorkspaceLastActivityAt, isGitWorkspaceExpirationEligible } from "../services/git-workspace-expiration.js"; +import { workspaceMoveEligible } from "../services/isolated-repository-rehome.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const describeDatabase = support.supported ? describe : describe.skip; + +describeDatabase("Git workspace activity bindings", () => { + let database: Awaited>; + let db: ReturnType; + const now = new Date("2026-10-20T12:00:00Z"); + const old = new Date("2026-10-01T12:00:00Z"); + const recent = new Date("2026-10-20T11:30:00Z"); + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("git-workspace-activity-"); + db = createDb(database.connectionString, { maxConnections: 1 }); + // Retained rows can predate native scope bindings. Preserve each fixture's + // independent native/legacy path instead of normalizing JSON on insertion. + await db.execute(sql`ALTER TABLE heartbeat_runs DISABLE TRIGGER heartbeat_runs_set_scope_kind`); + }); + afterEach(async () => { await db.execute(sql`TRUNCATE TABLE companies CASCADE`); }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); }); + + async function fixture() { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Activity", issuePrefix: `AC${companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Activity" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Root", status: "done", completedAt: old, updatedAt: old }); + const [workspace] = await db.insert(executionWorkspaces).values({ companyId, sourceIssueId: issueId, + mode: "isolated_workspace", strategyType: "git_worktree", name: "Activity", cwd: `/tmp/activity-${companyId}` }).returning(); + const [child] = await db.insert(issues).values({ companyId, parentId: issueId, title: "Child", status: "done", completedAt: old, updatedAt: old }).returning(); + const [bound] = await db.insert(issues).values({ companyId, executionWorkspaceId: workspace.id, title: "Bound", status: "done", completedAt: old, updatedAt: old }).returning(); + return { companyId, agentId, issueId, workspace, child, bound, + eligible: () => db.transaction(tx => isGitWorkspaceExpirationEligible(tx, workspace, now)) }; + } + + it.each(["rootIssue", "childIssue", "boundIssue", "boundNative", "contextChildIssue", "contextChildTask", "contextWorkspace", "checkout", "execution"])( + "keeps recent activity and deletion vetoes through %s", async route => { + const f = await fixture(); + const runId = randomUUID(); + const issueId = route === "rootIssue" ? f.issueId : route === "childIssue" ? f.child.id : f.bound.id; + await db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: f.agentId, status: "scheduled_retry", createdAt: recent, + ...(route.endsWith("Issue") && !route.startsWith("context") ? { scopeKind: "issue" as const, issueId } : {}), + ...(route === "boundNative" ? { nativeIssueId: f.bound.id } : {}), + contextSnapshot: route === "contextChildIssue" ? { issueId: f.child.id } : route === "contextChildTask" ? { taskId: f.child.id } + : route === "contextWorkspace" ? { executionWorkspaceId: f.workspace.id } : null }); + if (route === "checkout" || route === "execution") await db.update(issues) + .set(route === "checkout" ? { checkoutRunId: runId } : { executionRunId: runId }).where(eq(issues.id, f.bound.id)); + expect(await gitWorkspaceLastActivityAt(db, f.workspace)).toEqual(recent); + expect(await f.eligible()).toBe(false); + expect(await workspaceMoveEligible(db, f.workspace)).toBe(false); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: recent }).where(eq(heartbeatRuns.id, runId)); + expect(await f.eligible()).toBe(true); + expect(await workspaceMoveEligible(db, f.workspace)).toBe(true); + // Downstream lease matching depends on the same run IDs even after the + // controller has settled, with no direct issue/workspace lease binding. + await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: runId, status: "pending_cleanup", releasedAt: recent }); + expect(await f.eligible()).toBe(false); + expect(await workspaceMoveEligible(db, f.workspace)).toBe(false); + }); + + it("retains both native and conflicting legacy bindings", async () => { + const f = await fixture(); + const [unrelated] = await db.insert(issues).values({ companyId: f.companyId, title: "Unrelated", updatedAt: old }).returning(); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, nativeIssueId: unrelated.id, + status: "running", createdAt: recent, contextSnapshot: { taskId: f.issueId } }); + expect(await f.eligible()).toBe(false); + expect(await gitWorkspaceLastActivityAt(db, f.workspace)).toEqual(recent); + }); + + it("includes a multiply bound run exactly once", async () => { + const f = await fixture(), runId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: f.agentId, scopeKind: "issue", issueId: f.issueId, + nativeIssueId: f.child.id, status: "succeeded", createdAt: recent, finishedAt: recent, + contextSnapshot: { issueId: f.issueId, taskId: f.child.id, executionWorkspaceId: f.workspace.id } }); + await db.update(issues).set({ checkoutRunId: runId, executionRunId: runId }).where(eq(issues.id, f.bound.id)); + const [row] = await db.execute<{ count: number }>(sql`${expirationScope(f.workspace)} SELECT count(*)::int AS count FROM relevant_runs`); + expect(row.count).toBe(1); + expect(await f.eligible()).toBe(true); + }); + + it("keeps malformed contexts, case sensitivity, and company isolation unchanged", async () => { + const f = await fixture(); + const contexts = [null, [], [{ issueId: f.issueId }], [[{ taskId: f.issueId }]], true, 12, "string", { issueId: f.issueId.toUpperCase() }, + { issueId: 12 }, { taskId: { id: f.issueId } }, { executionWorkspaceId: [f.workspace.id] }]; + for (const context of contexts) await db.$client.unsafe(`INSERT INTO heartbeat_runs (company_id, agent_id, status, created_at, context_snapshot) + VALUES ($1::uuid, $2::uuid, 'running', $3::timestamptz, $4::jsonb)`, [f.companyId, f.agentId, recent.toISOString(), JSON.stringify(context)]); + const otherCompanyId = randomUUID(), otherAgentId = randomUUID(); + await db.insert(companies).values({ id: otherCompanyId, name: "Other", issuePrefix: `AO${otherCompanyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: otherAgentId, companyId: otherCompanyId, name: "Other" }); + await db.insert(heartbeatRuns).values({ companyId: otherCompanyId, agentId: otherAgentId, nativeIssueId: f.issueId, + status: "running", createdAt: recent, contextSnapshot: { issueId: f.issueId, taskId: f.issueId, executionWorkspaceId: f.workspace.id } }); + expect(await f.eligible()).toBe(true); + expect(await gitWorkspaceLastActivityAt(db, f.workspace)).toEqual(old); + }); + + it.each(["null", "array", "nestedArray", "number", "string", "boolean"])("retains native matches with %s context", async kind => { + const f = await fixture(); + const contexts = { null: null, array: [{ issueId: f.issueId }], nestedArray: [[{ taskId: f.issueId }]], number: 12, string: "string", boolean: true }; + const context = contexts[kind as keyof typeof contexts]; + await db.$client.unsafe(`INSERT INTO heartbeat_runs (company_id, agent_id, native_issue_id, status, created_at, context_snapshot) + VALUES ($1::uuid, $2::uuid, $3::uuid, 'running', $4::timestamptz, $5::jsonb)`, [f.companyId, f.agentId, f.issueId, recent.toISOString(), JSON.stringify(context)]); + expect(await f.eligible()).toBe(false); + expect(await gitWorkspaceLastActivityAt(db, f.workspace)).toEqual(recent); + }); + + it("keeps a finished run with a current controller lease", async () => { + const f = await fixture(); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "succeeded", createdAt: old, finishedAt: old, + controllerLeaseExpiresAt: new Date(Date.now() + 60_000), contextSnapshot: { executionWorkspaceId: f.workspace.id } }); + expect(await f.eligible()).toBe(false); + }); + + it("still observes a newly active consumer between successive checks", async () => { + const f = await fixture(); + expect(await f.eligible()).toBe(true); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "queued", createdAt: recent, + contextSnapshot: { executionWorkspaceId: f.workspace.id } }); + expect(await f.eligible()).toBe(false); + expect(await gitWorkspaceLastActivityAt(db, f.workspace)).toEqual(recent); + }); +}); diff --git a/server/src/services/git-workspace-expiration.ts b/server/src/services/git-workspace-expiration.ts index 5fdd85c7f96..8dcf1ac7d6a 100644 --- a/server/src/services/git-workspace-expiration.ts +++ b/server/src/services/git-workspace-expiration.ts @@ -102,6 +102,9 @@ async function restorePointBoundary(cwd: string, repoUrl: string, headOnRemote: /** Re-read the whole tree and every runtime/recovery consumer immediately before deletion. */ export function expirationScope(workspace: Workspace) { + // Decode the three legacy keys together only after cheap native/pointer + // matches fail. Strict JSONPath leaves non-object inputs unmatched, as ->> + // did, while those inputs can still match an independent native binding. return sql` WITH RECURSIVE tree AS ( SELECT id, status, completed_at, cancelled_at, updated_at FROM ${issues} @@ -110,17 +113,24 @@ export function expirationScope(workspace: Workspace) { SELECT c.id, c.status, c.completed_at, c.cancelled_at, c.updated_at FROM ${issues} c JOIN tree p ON c.parent_id = p.id WHERE c.company_id = ${workspace.companyId} ), bound AS ( - SELECT id, updated_at FROM ${issues} + SELECT id, updated_at, checkout_run_id, execution_run_id FROM ${issues} WHERE company_id = ${workspace.companyId} AND execution_workspace_id = ${workspace.id} + ), legacy_issue_ids AS MATERIALIZED ( + SELECT unnest(array_agg(id::text)) AS id FROM tree ), relevant_runs AS ( - SELECT r.* FROM ${heartbeatRuns} r WHERE r.company_id = ${workspace.companyId} AND ( + SELECT r.id, r.status, r.created_at, r.started_at, r.finished_at, r.controller_lease_expires_at + FROM ${heartbeatRuns} r WHERE r.company_id = ${workspace.companyId} AND ( r.issue_id IN (SELECT id FROM tree) OR r.native_issue_id IN (SELECT id FROM tree) OR r.issue_id IN (SELECT id FROM bound) OR r.native_issue_id IN (SELECT id FROM bound) - OR r.context_snapshot->>'issueId' IN (SELECT id::text FROM tree) - OR r.context_snapshot->>'taskId' IN (SELECT id::text FROM tree) - OR r.context_snapshot->>'executionWorkspaceId' = ${workspace.id} - OR r.id IN (SELECT unnest(ARRAY[checkout_run_id, execution_run_id]) FROM ${issues} - WHERE company_id = ${workspace.companyId} AND execution_workspace_id = ${workspace.id}) + OR r.id IN (SELECT unnest(ARRAY[checkout_run_id, execution_run_id]) FROM bound) + OR EXISTS ( + SELECT 1 FROM jsonb_to_record( + jsonb_path_query_first(r.context_snapshot, 'strict $ ? (@.type() == "object")') + ) context("issueId" text, "taskId" text, "executionWorkspaceId" text) + WHERE context."issueId" IN (SELECT id FROM legacy_issue_ids) + OR context."taskId" IN (SELECT id FROM legacy_issue_ids) + OR context."executionWorkspaceId" = ${workspace.id} + ) ) )`; }