mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
perf(server): make run liveness backfill lookup issue-scoped (#20393)
## Thinking Path
> - Paperclip is the open source app people use to manage AI agents for
work
> - Each finished agent run gets a liveness classification. The task
page shows it, and recovery reads it.
> - Run finalization writes the classification. A backfill repairs runs
that do not have it, for example older runs or runs whose finalization
failed.
> - The task page runs read (`GET /issues/:id/runs`) starts this
backfill on each read, for each task.
> - The candidate lookup reads every unclassified finished run in the
company. For each row, it reads the run context and probes
`activity_log`. Rows of other tasks never match, so each read repeats
the same work.
> - On a hosted test instance, this lookup used approximately 1 s per
call, 2,636 calls, and 3,500 s in total.
> - This pull request makes the lookup read only the rows of the task,
and skips a task after a pass finds nothing left to repair.
> - The benefit is less database work per task page read. The
classification still gets written.
## Linked Issues or Issue Description
**What existing behavior does this improve?**
The run liveness backfill that the task runs read starts
(`scheduleRunLivenessBackfill` in `server/src/services/activity.ts`).
**Subsystem affected**
Server: activity service, task runs read, run liveness backfill.
**Current behavior**
Each task runs read runs this lookup:
```sql
select ... from heartbeat_runs
where company_id = $1 and liveness_state is null and status not in ('queued', 'running')
and (context_snapshot ->> 'issueId' = $2
or exists (select 1 from activity_log where company_id = $3 and entity_type = 'issue'
and entity_id = $4 and run_id = heartbeat_runs.id))
limit 20
```
PostgreSQL reads all unclassified finished runs of the company through
`heartbeat_runs_company_liveness_idx`. It then reads the context of each
row, which can be large, and probes `activity_log` for each row.
Unclassified runs of other tasks, and runs without a task, stay
unclassified, because no read of their own task repairs them. Thus the
cost grows with the company, not with the task. The lookup also runs
again on each read when the task has nothing to repair.
**Proposed behavior**
- The lookup reads only the runs of this task. One branch uses
`heartbeat_runs_company_ctx_issue_created_idx` (context `issueId`). The
other branch uses `activity_log_company_entity_created_idx` (issue
activity rows with a run id). The result keeps the same filter: same
company, liveness missing, status not `queued` or `running`, at most 20
rows.
- When a pass finds fewer than 20 rows, it has repaired all rows of the
task. The service then skips the task for 10 minutes. A full batch of 20
does not start the skip window, so the next read repairs the next batch.
**Reason and benefit**
The lookup is now proportional to the history of one task. Most reads
find nothing, and those reads now do no database work for 10 minutes.
**Breaking changes**
None. No migration, no API change, no new background work.
**Additional context**
The chat bootstrap read already skips the backfill
(`skipLivenessBackfill`). That option still works, and a test covers it.
## What Changed
- `server/src/services/run-liveness-backfill-query.ts`: new
`missingRunLivenessCandidatesQuery`. It replaces the `OR` and the
correlated `exists` with `id in (context branch union all activity
branch)`. Both branches are filtered by company and task.
- `server/src/services/activity.ts`: the backfill uses the new query and
returns the number of rows it found. `scheduleRunLivenessBackfill` keeps
a bounded in-memory map (at most 10,000 tasks, oldest entry is removed
first) of tasks with nothing left to repair. It skips these tasks for 10
minutes.
- `server/src/__tests__/activity-service.test.ts`: four new
real-database tests.
- The backfill repairs runs found by context and runs found only by
issue activity. It does not change a running run, an unclassified run of
another task, or a run of another company that names the same task id.
- After a pass leaves nothing to repair, more reads do not repair a new
unclassified run. After the skip window ends, the next read repairs it.
- A task with 25 unclassified runs (more than one batch) gets all of
them repaired over later reads.
- A read with `skipLivenessBackfill` does not repair. The next normal
read does.
- `scripts/benchmark-run-liveness-backfill.ts`: a local SQL benchmark
that compares the pre-fix query with the new query.
## Verification
Local SQL benchmark (embedded PostgreSQL 18.1, macOS, synthetic data):
```sh
cd server
pnpm exec tsx ../scripts/benchmark-run-liveness-backfill.ts --baseline-ref=02bfe71a6bbd7c5676cb7e72778b361c5bc72765
```
- Baseline: `02bfe71a6b` (pre-fix master). Treatment: `80fcfcf7f1`,
clean tree.
- Fixture assumptions (staging row counts were not available): one
company with 30,000 classified runs, 6,000 unclassified finished runs of
other tasks or of no task, each with an approximately 20 KiB hex
context, 400 other tasks, and 300,000 `activity_log` rows (half with a
run id). The target task has 40 classified runs, 200 activity rows, and
the missing rows of each case. Decoys: one `running` run of the task,
and one run in another company that names the task id. A small control
uses 500 classified and 50 unclassified runs.
- Each arm: 2 warm-up runs and 11 measured runs, in alternating order.
| Case | Candidates | Baseline median (min to max) | New median (min to
max) | Speedup | Shared buffers (baseline to new) |
| --- | --- | --- | --- | --- | --- |
| Task with nothing to repair | 0 | 37.2 ms (31.1 to 44.1) | 1.21 ms
(0.42 to 1.79) | 30.8x | 34,247 to 138 |
| Task with 3 missing | 3 | 37.8 ms (27.2 to 39.6) | 1.16 ms (0.77 to
2.05) | 32.6x | 34,265 to 148 |
| Task with 25 missing | 25 | 37.5 ms (34.6 to 69.8) | 1.63 ms (1.06 to
3.47) | 23.0x | 34,361 to 154 |
| Small control | 3 | 0.79 ms (0.70 to 1.06) | 0.70 ms (0.48 to 1.70) |
1.1x (neutral) | 314 to 18 |
The `EXPLAIN ANALYZE` execution time for the large cases goes from
approximately 35 ms to approximately 0.15 ms. These are local SQL
measurements. The hosted latency improvement is not yet measured.
The benchmark also checks result equality. For each case, it runs both
queries without a limit and compares every column of every row (as
sorted positional rows). It also checks that the limited batch of 20
comes from the baseline candidate set. All cases match.
Beyond the query, the skip window removes the lookup for most reads of a
task with nothing to repair.
Tests:
```sh
cd server
pnpm exec vitest run src/__tests__/activity-service.test.ts # 24 passed
pnpm exec vitest run src/__tests__/activity-routes.test.ts src/__tests__/task-cold-open.test.ts src/__tests__/liveness.test.ts
pnpm exec tsc --noEmit -p .
```
- The skip-window test fails on the pre-fix `activity.ts` (red), and
passes with this change (green).
- `pnpm verify` asks for broader coverage, because the change adds a
script. Required CI supplies that coverage.
## Risks
- A run that misses its classification after a pass finds nothing for
its task gets repaired up to 10 minutes later. Normal runs get their
classification at finalization, so only the repair path is late.
- The skip map is in memory for each server process. A restart clears
it. That only causes one more lookup for each task.
- The new query keeps the same rows, filter, and limit. The query has no
order, as before. Thus the 20 rows that a pass picks can differ when a
task has more than 20 candidates. Later reads repair the rest in both
versions.
- Low risk. No migration and no API change.
## Model Used
- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code
execution in Claude Code.
## Checklist
- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
199670ba99
commit
17014cd7db
4 files changed
+352
-35
No files matched your search
@@ -0,0 +1,176 @@
|
||||
// Local SQL benchmark for the run liveness backfill candidate lookup.
|
||||
// Usage: pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-run-liveness-backfill.ts --baseline-ref=<pre-fix sha>
|
||||
// The baseline query is the pre-fix builder from activity.ts, copied verbatim.
|
||||
import assert from "node:assert/strict";
|
||||
import { execFileSync } from "node:child_process";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { and, eq, isNull, or, sql } from "../server/node_modules/drizzle-orm/index.js";
|
||||
import { activityLog, agents, companies, createDb, heartbeatRuns, issues, startEmbeddedPostgresTestDatabase, type Db }
|
||||
from "../packages/db/src/index.js";
|
||||
import { missingRunLivenessCandidatesQuery } from "../server/src/services/run-liveness-backfill-query.js";
|
||||
|
||||
type Query = { sql: string; params: unknown[] };
|
||||
const repoRoot = fileURLToPath(new URL("../", import.meta.url));
|
||||
const baselineRef = process.argv.find(arg => arg.startsWith("--baseline-ref="))?.slice(15);
|
||||
assert.ok(baselineRef, "Pass a pinned pre-fix --baseline-ref");
|
||||
const git = (...args: string[]) => execFileSync("git", args, { cwd: repoRoot, encoding: "utf8" }).trim();
|
||||
const baselineRevision = git("rev-parse", "--verify", `${baselineRef}^{commit}`);
|
||||
const baselineSource = git("show", `${baselineRevision}:server/src/services/activity.ts`);
|
||||
assert.ok(baselineSource.includes("${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`,\n sql`exists ("),
|
||||
"Baseline revision must contain the pre-fix candidate query");
|
||||
|
||||
function baselineQuery(db: Db, companyId: string, issueId: string, limit: number) {
|
||||
return db
|
||||
.select({
|
||||
id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, status: heartbeatRuns.status,
|
||||
contextSnapshot: heartbeatRuns.contextSnapshot, resultJson: heartbeatRuns.resultJson,
|
||||
stdoutExcerpt: heartbeatRuns.stdoutExcerpt, stderrExcerpt: heartbeatRuns.stderrExcerpt, error: heartbeatRuns.error,
|
||||
errorCode: heartbeatRuns.errorCode, continuationAttempt: heartbeatRuns.continuationAttempt,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.where(and(
|
||||
eq(heartbeatRuns.companyId, companyId),
|
||||
isNull(heartbeatRuns.livenessState),
|
||||
sql`${heartbeatRuns.status} not in ('queued', 'running')`,
|
||||
or(
|
||||
sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`,
|
||||
sql`exists (
|
||||
select 1
|
||||
from ${activityLog}
|
||||
where ${activityLog.companyId} = ${companyId}
|
||||
and ${activityLog.entityType} = 'issue'
|
||||
and ${activityLog.entityId} = ${issueId}
|
||||
and ${activityLog.runId} = ${heartbeatRuns.id}
|
||||
)`,
|
||||
),
|
||||
))
|
||||
.limit(limit);
|
||||
}
|
||||
|
||||
const fixtures = [
|
||||
{ name: "clean issue, many unclassified runs elsewhere", unclassifiedElsewhere: 6000, classified: 30000, activity: 300000, targetMissing: 0 },
|
||||
{ name: "issue with 3 missing, many unclassified runs elsewhere", unclassifiedElsewhere: 6000, classified: 30000, activity: 300000, targetMissing: 3 },
|
||||
{ name: "issue with 25 missing (more than one batch)", unclassifiedElsewhere: 6000, classified: 30000, activity: 300000, targetMissing: 25 },
|
||||
{ name: "small control", unclassifiedElsewhere: 50, classified: 500, activity: 2000, targetMissing: 3 },
|
||||
];
|
||||
|
||||
const keepAlive = setInterval(() => {}, 1000);
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: Db | undefined;
|
||||
try {
|
||||
database = await startEmbeddedPostgresTestDatabase("run-liveness-backfill-benchmark-");
|
||||
db = createDb(database.connectionString);
|
||||
const client = db.$client;
|
||||
const [settings] = await client.unsafe("select version(), current_setting('shared_buffers') shared_buffers, current_setting('work_mem') work_mem, current_setting('jit') jit");
|
||||
const results = [];
|
||||
for (const fixture of fixtures) {
|
||||
await client.unsafe("set client_min_messages = warning");
|
||||
await client.unsafe("truncate activity_log, heartbeat_runs, issues, agents, companies cascade");
|
||||
const companyId = randomUUID(), otherCompanyId = randomUUID(), agentId = randomUUID(), otherAgentId = randomUUID();
|
||||
const issueId = randomUUID();
|
||||
await db.insert(companies).values([
|
||||
{ id: companyId, name: "Fixture", issuePrefix: `L${companyId.replaceAll("-", "").slice(0, 8)}` },
|
||||
{ id: otherCompanyId, name: "Decoy", issuePrefix: `D${otherCompanyId.replaceAll("-", "").slice(0, 8)}` },
|
||||
]);
|
||||
await db.insert(agents).values([
|
||||
{ id: agentId, companyId, name: "Fixture", adapterType: "codex_local" },
|
||||
{ id: otherAgentId, companyId: otherCompanyId, name: "Decoy", adapterType: "codex_local" },
|
||||
]);
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Target", status: "in_progress", assigneeAgentId: agentId });
|
||||
// 400 other issue ids in the same company.
|
||||
await client.unsafe(`insert into issues (id, company_id, title, status)
|
||||
select gen_random_uuid(), $1, 'Other ' || g, 'todo' from generate_series(1, 400) g`, [companyId]);
|
||||
// Wide, mostly incompressible context: hex digests, about 20 KiB each.
|
||||
const blob = `(select string_agg(md5(random()::text || g || s), '') from generate_series(1, 640) s)`;
|
||||
const otherIssues = `(select array_agg(id::text order by id) ids from issues where company_id = $1 and id::text <> $4::text)`;
|
||||
// Classified runs across the company (the common case).
|
||||
await client.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, liveness_state, context_snapshot, created_at)
|
||||
select $1, $2, 'succeeded', 'advanced', jsonb_build_object('issueId', other.ids[1 + g % 400]),
|
||||
now() - (g || ' seconds')::interval
|
||||
from generate_series(1, $3::int) g, ${otherIssues} other`, [companyId, agentId, fixture.classified, issueId]);
|
||||
// Unclassified finished runs of other issues and of no issue. They never match the target issue.
|
||||
await client.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, liveness_state, context_snapshot, created_at)
|
||||
select $1, $2, case when g % 3 = 0 then 'cancelled' when g % 3 = 1 then 'failed' else 'timed_out' end, null,
|
||||
case when g % 4 = 0 then jsonb_build_object('blob', ${blob})
|
||||
else jsonb_build_object('issueId', other.ids[1 + g % 400], 'blob', ${blob}) end,
|
||||
now() - (g || ' minutes')::interval
|
||||
from generate_series(1, $3::int) g, ${otherIssues} other`, [companyId, agentId, fixture.unclassifiedElsewhere, issueId]);
|
||||
// Target issue: classified history, missing rows via context, missing rows linked only by activity,
|
||||
// one running run (excluded), and a cross-company decoy naming the same issue id.
|
||||
await client.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, liveness_state, context_snapshot)
|
||||
select $1, $2, 'succeeded', 'completed', jsonb_build_object('issueId', $3::text, 'blob', ${blob}) from generate_series(1, 40) g`,
|
||||
[companyId, agentId, issueId]);
|
||||
const viaContext = Math.ceil(fixture.targetMissing / 2), viaActivity = fixture.targetMissing - viaContext;
|
||||
await client.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, liveness_state, context_snapshot)
|
||||
select $1, $2, 'failed', null, jsonb_build_object('issueId', $3::text, 'blob', ${blob}) from generate_series(1, $4::int) g`,
|
||||
[companyId, agentId, issueId, viaContext]);
|
||||
await client.unsafe(`with runs as (
|
||||
insert into heartbeat_runs (company_id, agent_id, status, liveness_state, context_snapshot)
|
||||
select $1, $2, 'cancelled', null, jsonb_build_object('blob', ${blob}) from generate_series(1, $4::int) g returning id)
|
||||
insert into activity_log (company_id, actor_id, action, entity_type, entity_id, run_id)
|
||||
select $1, 'agent', 'issue.updated', 'issue', $3, id from runs`, [companyId, agentId, issueId, viaActivity]);
|
||||
await client.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, liveness_state, context_snapshot)
|
||||
values ($1, $2, 'running', null, jsonb_build_object('issueId', $3::text))`, [companyId, agentId, issueId]);
|
||||
await client.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, liveness_state, context_snapshot)
|
||||
values ($1, $2, 'failed', null, jsonb_build_object('issueId', $3::text))`, [otherCompanyId, otherAgentId, issueId]);
|
||||
// Activity log: about half the rows carry a run id of a run that is not the target issue's.
|
||||
// Issue entities spread over the other issues; the target issue has its own history too.
|
||||
await client.unsafe(`insert into activity_log (company_id, actor_id, action, entity_type, entity_id, run_id, created_at)
|
||||
select $1, 'agent', 'issue.updated', 'issue', other.ids[1 + g % 400],
|
||||
case when g % 2 = 0 then runs.ids[1 + g % array_length(runs.ids, 1)] end,
|
||||
now() - (g || ' seconds')::interval
|
||||
from generate_series(1, $2::int) g, ${otherIssues.replace("$4", "$3")} other,
|
||||
(select array_agg(id order by id) ids from heartbeat_runs
|
||||
where company_id = $1 and context_snapshot ->> 'issueId' is distinct from $3) runs`,
|
||||
[companyId, fixture.activity, issueId]);
|
||||
await client.unsafe(`insert into activity_log (company_id, actor_id, action, entity_type, entity_id, run_id)
|
||||
select $1, 'agent', 'issue.comment_added', 'issue', $2, case when g % 2 = 0 then runs.ids[1 + g % 40] end
|
||||
from generate_series(1, 200) g, (select array_agg(id order by id) ids from heartbeat_runs
|
||||
where company_id = $1 and context_snapshot ->> 'issueId' = $2 and liveness_state = 'completed') runs`,
|
||||
[companyId, issueId]);
|
||||
await client.unsafe("vacuum analyze");
|
||||
|
||||
const before = baselineQuery(db, companyId, issueId, 20).toSQL() as Query;
|
||||
const after = missingRunLivenessCandidatesQuery(db, companyId, issueId, 20).toSQL() as Query;
|
||||
const execute = async (query: Query) => client.unsafe(query.sql, query.params as never[]).values();
|
||||
// Equivalence on the complete candidate set (no limit), every column, stable order.
|
||||
const sortRows = (rows: unknown[][]) => rows.map(row => JSON.stringify(row)).sort();
|
||||
const fullBefore = sortRows(await execute(baselineQuery(db, companyId, issueId, 100000).toSQL() as Query));
|
||||
const fullAfter = sortRows(await execute(missingRunLivenessCandidatesQuery(db, companyId, issueId, 100000).toSQL() as Query));
|
||||
assert.deepEqual(fullAfter, fullBefore, "Full candidate set must match on every column");
|
||||
assert.equal(fullAfter.length, fixture.targetMissing);
|
||||
const limitedAfter = sortRows(await execute(after));
|
||||
assert.equal(limitedAfter.length, Math.min(20, fixture.targetMissing));
|
||||
assert.ok(limitedAfter.every(row => fullBefore.includes(row)), "Limited batch must come from the baseline candidate set");
|
||||
|
||||
const samples = { baseline: [] as number[], treatment: [] as number[] };
|
||||
for (let round = 0; round < 13; round++) {
|
||||
for (const arm of round % 2 ? ["treatment", "baseline"] as const : ["baseline", "treatment"] as const) {
|
||||
const start = performance.now();
|
||||
await execute(arm === "baseline" ? before : after);
|
||||
if (round >= 2) samples[arm].push(performance.now() - start);
|
||||
}
|
||||
}
|
||||
const explain = async (query: Query) => {
|
||||
const [row] = await client.unsafe(`explain (analyze, buffers, format json) ${query.sql}`, query.params as never[]);
|
||||
const plan = (row!["QUERY PLAN"] as Array<{ Plan: Record<string, unknown>; "Execution Time": number }>)[0]!;
|
||||
return { executionMs: plan["Execution Time"], sharedHit: plan.Plan["Shared Hit Blocks"], sharedRead: plan.Plan["Shared Read Blocks"] };
|
||||
};
|
||||
const summarize = (values: number[]) => {
|
||||
const sorted = [...values].sort((a, b) => a - b);
|
||||
const median = sorted[Math.floor(sorted.length / 2)]!;
|
||||
return { medianMs: +median.toFixed(3), minMs: +sorted[0]!.toFixed(3), maxMs: +sorted.at(-1)!.toFixed(3) };
|
||||
};
|
||||
const b = summarize(samples.baseline), t = summarize(samples.treatment);
|
||||
results.push({ fixture, candidates: fullAfter.length, fullSetEquality: true,
|
||||
baseline: { ...b, plan: await explain(before) }, treatment: { ...t, plan: await explain(after) },
|
||||
speedup: +(b.medianMs / t.medianMs).toFixed(1) });
|
||||
}
|
||||
console.log(JSON.stringify({ baselineRevision, treatmentRevision: git("rev-parse", "HEAD"),
|
||||
treatmentDirty: git("status", "--porcelain", "--", "server/src/services/activity.ts", "server/src/services/run-liveness-backfill-query.ts") !== "",
|
||||
settings, node: process.version, warmupsPerArm: 2, measuredSamplesPerArm: 11, results }, null, 2));
|
||||
} finally {
|
||||
try { await db?.$client.end({ timeout: 0 }); await database?.cleanup(); }
|
||||
finally { clearInterval(keepAlive); }
|
||||
}
|
||||
@@ -1,6 +1,6 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
activityLog,
|
||||
agentWakeupRequests,
|
||||
@@ -693,4 +693,98 @@ describeEmbeddedPostgres("activity service", () => {
|
||||
lastUsefulActionAt: null,
|
||||
});
|
||||
});
|
||||
|
||||
describe("run liveness backfill", () => {
|
||||
async function seedIssue() {
|
||||
const companyId = randomUUID();
|
||||
const agentId = randomUUID();
|
||||
const issueId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Backfill", issuePrefix: `B${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}` });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Coder", role: "engineer", status: "idle", adapterType: "codex_local" });
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Backfill target", status: "done", assigneeAgentId: agentId });
|
||||
return { companyId, agentId, issueId };
|
||||
}
|
||||
|
||||
async function insertMissingRun(companyId: string, agentId: string, contextSnapshot: Record<string, unknown>, status = "succeeded") {
|
||||
const runId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status, contextSnapshot, livenessState: null });
|
||||
return runId;
|
||||
}
|
||||
|
||||
async function livenessOf(runIds: string[]) {
|
||||
const rows = await db.select({ id: heartbeatRuns.id, livenessState: heartbeatRuns.livenessState }).from(heartbeatRuns);
|
||||
return runIds.map((id) => rows.find((row) => row.id === id)?.livenessState ?? null);
|
||||
}
|
||||
|
||||
async function waitForLiveness(service: ActivityService, companyId: string, issueId: string, runIds: string[]) {
|
||||
const deadline = performance.now() + 3_000;
|
||||
while (performance.now() < deadline) {
|
||||
await service.runsForIssue(companyId, issueId);
|
||||
if ((await livenessOf(runIds)).every((state) => state !== null)) {
|
||||
// Let the pass finish its bookkeeping after the last update.
|
||||
await new Promise((resolve) => setTimeout(resolve, 50));
|
||||
return;
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 25));
|
||||
}
|
||||
throw new Error(`Timed out waiting for liveness: ${JSON.stringify(await livenessOf(runIds))}`);
|
||||
}
|
||||
|
||||
it("backfills only this issue's runs, found by context or by issue activity", async () => {
|
||||
const { companyId, agentId, issueId } = await seedIssue();
|
||||
const viaContext = await insertMissingRun(companyId, agentId, { issueId });
|
||||
const viaActivity = await insertMissingRun(companyId, agentId, {}, "failed");
|
||||
await db.insert(activityLog).values({ companyId, actorType: "agent", actorId: agentId, action: "issue.updated",
|
||||
entityType: "issue", entityId: issueId, runId: viaActivity });
|
||||
const running = await insertMissingRun(companyId, agentId, { issueId }, "running");
|
||||
const otherIssueRun = await insertMissingRun(companyId, agentId, { issueId: randomUUID() });
|
||||
const other = await seedIssue();
|
||||
const otherCompanyRun = await insertMissingRun(other.companyId, other.agentId, { issueId });
|
||||
|
||||
await waitForLiveness(activityService(db), companyId, issueId, [viaContext, viaActivity]);
|
||||
|
||||
expect(await livenessOf([running, otherIssueRun, otherCompanyRun])).toEqual([null, null, null]);
|
||||
});
|
||||
|
||||
it("does not repeat the lookup for a clean issue until the skip window ends", async () => {
|
||||
const { companyId, agentId, issueId } = await seedIssue();
|
||||
const first = await insertMissingRun(companyId, agentId, { issueId });
|
||||
const service = activityService(db);
|
||||
await waitForLiveness(service, companyId, issueId, [first]);
|
||||
|
||||
const late = await insertMissingRun(companyId, agentId, { issueId });
|
||||
for (let read = 0; read < 3; read++) {
|
||||
await service.runsForIssue(companyId, issueId);
|
||||
await new Promise((resolve) => setTimeout(resolve, 50));
|
||||
}
|
||||
expect(await livenessOf([late])).toEqual([null]);
|
||||
|
||||
const now = Date.now();
|
||||
const clock = vi.spyOn(Date, "now").mockReturnValue(now + 11 * 60_000);
|
||||
try {
|
||||
await waitForLiveness(service, companyId, issueId, [late]);
|
||||
} finally {
|
||||
clock.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it("keeps backfilling on later reads while an issue has more than one batch", async () => {
|
||||
const { companyId, agentId, issueId } = await seedIssue();
|
||||
const runIds: string[] = [];
|
||||
for (let index = 0; index < 25; index++) runIds.push(await insertMissingRun(companyId, agentId, { issueId }));
|
||||
|
||||
await waitForLiveness(activityService(db), companyId, issueId, runIds);
|
||||
});
|
||||
|
||||
it("leaves the backfill to the task page when a read skips it", async () => {
|
||||
const { companyId, agentId, issueId } = await seedIssue();
|
||||
const runId = await insertMissingRun(companyId, agentId, { issueId });
|
||||
const service = activityService(db);
|
||||
await service.runsForIssue(companyId, issueId, null, 2, undefined, { skipLivenessBackfill: true });
|
||||
await new Promise((resolve) => setTimeout(resolve, 100));
|
||||
expect(await livenessOf([runId])).toEqual([null]);
|
||||
|
||||
await waitForLiveness(service, companyId, issueId, [runId]);
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -24,6 +24,7 @@ import { hasWorkspaceRestoreFailure, safeWorkspaceRestorePath, ISSUE_CONTINUATIO
|
||||
import { logger } from "../middleware/logger.js";
|
||||
import { visibleIssueCondition } from "./issue-visibility.js";
|
||||
import { classifyRunLiveness } from "./run-liveness.js";
|
||||
import { missingRunLivenessCandidatesQuery } from "./run-liveness-backfill-query.js";
|
||||
|
||||
export interface ActivityFilters {
|
||||
companyId: string;
|
||||
@@ -43,8 +44,27 @@ export function normalizeActivityLimit(limit: number | undefined) {
|
||||
return Math.max(1, Math.min(MAX_ACTIVITY_LIMIT, Math.floor(limit ?? DEFAULT_ACTIVITY_LIMIT)));
|
||||
}
|
||||
|
||||
const RUN_LIVENESS_BACKFILL_BATCH = 20;
|
||||
// Runs finalize with a liveness classification. The backfill only repairs rows
|
||||
// that missed it (older rows or a failed finalization), so a delay is safe.
|
||||
const RUN_LIVENESS_CLEAN_TTL_MS = 10 * 60_000;
|
||||
const RUN_LIVENESS_CLEAN_MAX_ENTRIES = 10_000;
|
||||
|
||||
export function activityService(db: Db) {
|
||||
const scheduledLivenessBackfills = new Set<string>();
|
||||
// Issues whose last backfill pass left no missing liveness, keyed like
|
||||
// `scheduledLivenessBackfills`. Values are the time the skip window ends.
|
||||
// The map keeps insertion order, so the first entry is the oldest.
|
||||
const livenessCleanUntil = new Map<string, number>();
|
||||
function rememberLivenessClean(key: string) {
|
||||
livenessCleanUntil.delete(key);
|
||||
livenessCleanUntil.set(key, Date.now() + RUN_LIVENESS_CLEAN_TTL_MS);
|
||||
while (livenessCleanUntil.size > RUN_LIVENESS_CLEAN_MAX_ENTRIES) {
|
||||
const oldest = livenessCleanUntil.keys().next().value;
|
||||
if (oldest === undefined) break;
|
||||
livenessCleanUntil.delete(oldest);
|
||||
}
|
||||
}
|
||||
const issueIdAsText = sql<string>`${issues.id}::text`;
|
||||
const summarizedUsageJson = sql<Record<string, unknown> | null>`
|
||||
case
|
||||
@@ -192,41 +212,9 @@ export function activityService(db: Db) {
|
||||
}
|
||||
|
||||
async function backfillMissingRunLivenessForIssue(companyId: string, issueId: string) {
|
||||
const runs = await db
|
||||
.select({
|
||||
id: heartbeatRuns.id,
|
||||
companyId: heartbeatRuns.companyId,
|
||||
status: heartbeatRuns.status,
|
||||
contextSnapshot: heartbeatRuns.contextSnapshot,
|
||||
resultJson: heartbeatRuns.resultJson,
|
||||
stdoutExcerpt: heartbeatRuns.stdoutExcerpt,
|
||||
stderrExcerpt: heartbeatRuns.stderrExcerpt,
|
||||
error: heartbeatRuns.error,
|
||||
errorCode: heartbeatRuns.errorCode,
|
||||
continuationAttempt: heartbeatRuns.continuationAttempt,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
and(
|
||||
eq(heartbeatRuns.companyId, companyId),
|
||||
isNull(heartbeatRuns.livenessState),
|
||||
sql`${heartbeatRuns.status} not in ('queued', 'running')`,
|
||||
or(
|
||||
sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`,
|
||||
sql`exists (
|
||||
select 1
|
||||
from ${activityLog}
|
||||
where ${activityLog.companyId} = ${companyId}
|
||||
and ${activityLog.entityType} = 'issue'
|
||||
and ${activityLog.entityId} = ${issueId}
|
||||
and ${activityLog.runId} = ${heartbeatRuns.id}
|
||||
)`,
|
||||
),
|
||||
),
|
||||
)
|
||||
.limit(20);
|
||||
const runs = await missingRunLivenessCandidatesQuery(db, companyId, issueId, RUN_LIVENESS_BACKFILL_BATCH);
|
||||
|
||||
if (runs.length === 0) return;
|
||||
if (runs.length === 0) return 0;
|
||||
|
||||
const issue = await db
|
||||
.select({
|
||||
@@ -357,13 +345,24 @@ export function activityService(db: Db) {
|
||||
})
|
||||
.where(and(eq(heartbeatRuns.id, run.id), isNull(heartbeatRuns.livenessState)));
|
||||
}
|
||||
return runs.length;
|
||||
}
|
||||
|
||||
function scheduleRunLivenessBackfill(companyId: string, issueId: string) {
|
||||
const key = `${companyId}:${issueId}`;
|
||||
if (scheduledLivenessBackfills.has(key)) return;
|
||||
const cleanUntil = livenessCleanUntil.get(key);
|
||||
if (cleanUntil !== undefined) {
|
||||
if (cleanUntil > Date.now()) return;
|
||||
livenessCleanUntil.delete(key);
|
||||
}
|
||||
scheduledLivenessBackfills.add(key);
|
||||
void backfillMissingRunLivenessForIssue(companyId, issueId)
|
||||
.then((found) => {
|
||||
// A partial batch means this pass classified every remaining candidate.
|
||||
// Skip this issue for a while, so page reads do not repeat the lookup.
|
||||
if (found < RUN_LIVENESS_BACKFILL_BATCH) rememberLivenessClean(key);
|
||||
})
|
||||
.catch((err: unknown) => {
|
||||
logger.warn({ err, companyId, issueId }, "run liveness backfill failed");
|
||||
})
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
import { and, eq, isNull, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { activityLog, heartbeatRuns } from "@paperclipai/db";
|
||||
|
||||
/**
|
||||
* Finished runs of one issue that have no liveness classification yet.
|
||||
* A run belongs to the issue when its context names the issue or when an
|
||||
* issue activity row records the run. Each branch reads only this issue's
|
||||
* rows through its own index, so the cost does not grow with the number of
|
||||
* unclassified runs in other issues of the company.
|
||||
*/
|
||||
export function missingRunLivenessCandidatesQuery(db: Db, companyId: string, issueId: string, limit: number) {
|
||||
return db
|
||||
.select({
|
||||
id: heartbeatRuns.id,
|
||||
companyId: heartbeatRuns.companyId,
|
||||
status: heartbeatRuns.status,
|
||||
contextSnapshot: heartbeatRuns.contextSnapshot,
|
||||
resultJson: heartbeatRuns.resultJson,
|
||||
stdoutExcerpt: heartbeatRuns.stdoutExcerpt,
|
||||
stderrExcerpt: heartbeatRuns.stderrExcerpt,
|
||||
error: heartbeatRuns.error,
|
||||
errorCode: heartbeatRuns.errorCode,
|
||||
continuationAttempt: heartbeatRuns.continuationAttempt,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
and(
|
||||
eq(heartbeatRuns.companyId, companyId),
|
||||
isNull(heartbeatRuns.livenessState),
|
||||
sql`${heartbeatRuns.status} not in ('queued', 'running')`,
|
||||
sql`${heartbeatRuns.id} in (
|
||||
select issue_run.id
|
||||
from ${heartbeatRuns} as issue_run
|
||||
where issue_run.company_id = ${companyId}
|
||||
and issue_run.context_snapshot ->> 'issueId' = ${issueId}
|
||||
union all
|
||||
select ${activityLog.runId}
|
||||
from ${activityLog}
|
||||
where ${activityLog.companyId} = ${companyId}
|
||||
and ${activityLog.entityType} = 'issue'
|
||||
and ${activityLog.entityId} = ${issueId}
|
||||
and ${activityLog.runId} is not null
|
||||
)`,
|
||||
),
|
||||
)
|
||||
.limit(limit);
|
||||
}
|
||||
Reference in new issue
Block a user