Jannik/PaperClipAI
1
0 Fork 0
mirror of https://github.com/paperclipai/paperclip.git synced 2026-11-12 06:10:44 +01:00
Code Issues Packages Projects Releases Wiki Activity

perf(recovery): probe run bindings per comment in the lost-wake check (#20412)

Browse Source 
## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - The ownership reconciler finds user comments whose wake a restart
lost, and it saves a wake for them (plan invariant 4)
> - Its lost-wake check matched a task's runs with one OR across the
legacy context ID and the native issue column
> - With that OR, PostgreSQL can join runs only by company. It then
tests each recent comment against every run of its company and reads
each run's context
> - On a hosted staging database, this statement used about 23 s per
call on average (384 calls, about 8,900 s in total). One call ran for 35
s. Other requests waited behind it
> - This pull request reads the check in three stages and probes the two
run bindings separately with existing indexes
> - The benefit is the same result with bounded, index-driven work for
each recent comment

## Linked Issues or Issue Description

**What happened?**

The reconciler's user-comment lost-wake check was one of the most
expensive statements on a hosted staging database. Query statistics
showed 384 calls with a mean of about 23 s. One active call was seen at
35 s. The database compute was busy, and short authentication reads
waited behind long statements.

The statement matched runs with `(context_snapshot->>'issueId' = issue
OR native_issue_id = issue)` in three subqueries. The OR gives
PostgreSQL no join key except `company_id`. A local plan on synthetic
data showed the cost: a hash join of 273 comments against all 38,002
runs, which took 1.0 s of a 1.3 s statement. The work grows with recent
comments multiplied by the company's run count, and each test reads a
run context.

**Expected behavior**

Each recent comment uses a small number of indexed probes. The check
returns the same comments in the same order, with the same limit and
grace logic.

**Steps to reproduce**

1. Seed a company with a deep run history and a few hundred recent user
comments.
2. Let some recent comments have no wake (for example, after a restart
lost the wake of a steered comment).
3. Run the check. PostgreSQL can pick a hash join on `company_id` with
the OR as a join filter. With nested loops off, the old statement takes
seconds. The local benchmark below shows this.

**Paperclip version or commit**

Baseline: `02bfe71a6b`.

**Deployment mode**

Server with PostgreSQL. Local proof uses disposable PostgreSQL 18.1.
Production and staging use hosted PostgreSQL.

Related: the earlier change that split native and legacy bindings for
other ownership reads (see the "Execution recovery query performance"
section in `doc/DATABASE.md`). A search found no open pull request for
this check.

## What Changed

- `checkUserCommentsWithoutWake` reads in three materialized stages:
1. Recent user comments on assigned, non-conversation tasks that no wake
request carries. This stage keeps the same issue, status and
agent-closed rules.
2. Of those, comments with no live or later run and with at least one
earlier run.
3. Of those, comments that no live run acknowledged in an item event
(the steer check). Only this stage reads event payloads.
- Each run lookup is two probes: one on the legacy context ID and one on
the native issue column. Both use existing indexes. `exists (A or B)`
becomes `exists (A) or exists (B)`, which is logically the same.
- Each probe ends with `offset 0`. This keeps it a per-row index lookup.
Without it, PostgreSQL can build a hashed subplan over all runs, or a
hash join over all item events of the company. The repository already
uses this fence in `execution-blocker.ts`.
- The final `order by c.created_at limit N`, the projection, the grace
window and the reported reason do not change.
- Add `scripts/benchmark-comment-wake-check.ts`. It loads the check from
an explicit baseline commit and from the working tree, and it compares
the complete service result and the complete positional SQL rows before
it records time.
- Add a regression test for each run binding on its own: native only,
legacy only, live run, later run, steered run, an acknowledgement before
the comment, a log-only mention, a run that ended before the comment,
and a run in another company.
- Document the check and the benchmark command in `doc/DATABASE.md`.
- No schema change, index, migration, or new startup work.

## Verification

- `pnpm exec vitest run
server/src/__tests__/ownership-reconciliation-report.test.ts
--maxWorkers=1`: 34 passed. This includes the existing lost-wake,
restart, steered and agent-closed tests and the new binding test.
- `pnpm --filter @paperclipai/server typecheck`: passed.
- `pnpm --filter @paperclipai/server exec tsx
../scripts/benchmark-comment-wake-check.ts --baseline-ref=02bfe71a6b
--samples=7`

Benchmark setup: PostgreSQL 18.1 (embedded, local), `shared_buffers` 128
MB, `work_mem` 4 MB. All data is synthetic, with md5-hex filler so
contexts and payloads do not compress well. There are two companies per
fixture. Two warmup rounds, then 7 timed samples for each arm, in
alternating order. Each sample runs the real check function. Before any
timing, each fixture asserts that the complete check result and every
positional SQL row are equal for baseline and treatment, and that the
fixture reports at least one owed comment. Every timed sample also
asserts the same result.

| Fixture | Runs / recent comments / owed | Baseline median | Treatment
median | Speedup | Hash-join plan, baseline → treatment |
| --- | --- | --- | --- | --- | --- |
| small | 2,541 / 50 / 6 | 8.5 ms | 2.6 ms | 3.2x | 19 ms → 3 ms |
| representative | 75,502 / 750 / 75 | 26.4 ms | 23.3 ms | 1.1x | 2,262
ms → 17 ms |
| lost steer wakes | 38,002 / 750 / 75 | 454.7 ms | 66.9 ms | 6.8x |
2,717 ms → 68 ms |
| deep history | 150,750 / 1,125 / 114 | 26.9 ms | 18.8 ms | 1.4x |
4,006 ms → 34 ms |
| busy hour | 128,750 / 5,625 / 200 (limit) | 79.2 ms | 47.4 ms | 1.7x |
20,047 ms → 75 ms |
| long steered runs (stress) | 19,100 / 600 / 60 | 626.4 ms | 457.4 ms |
1.4x | 5,871 ms → 502 ms |

- The table above used baseline `02bfe71a6b`, the branch base at that
time. After the rebase, a repeat run with baseline `8004ad069e` (the new
merge base, with the helper guard fixed to run from the repository root)
gave the same picture on a busier machine: small 15.4 → 9.1 ms, lost
steer wakes 1,733 → 139 ms, busy hour 103 → 65 ms, all with equal
results.
- The "lost steer wakes" fixture shows the slow plan with default
planner settings. PostgreSQL hashed all runs by company and tested 273
comments against 38,002 runs.
- The last column runs each statement once with `enable_nestloop = off`.
This makes the company-only hash join that the old statement allows
visible, and it shows that the new statement has no such plan. The
result is checked as equal in this mode too. Staging's 23 s mean matches
this growth: recent comments multiplied by the company's run count.
- Shared buffers fell in every fixture except "small" with default
settings (for example, busy hour 176,416 → 89,803).
- Neutral path: in the "long steered runs" stress fixture, every steered
comment lost its wake. These comments correctly reach the steer check.
That check must read the payload of each item event after the comment in
the run that was live, so it is the remaining cost. In normal operation
a comment sent to a live run has a deferred wake that carries it, so
stage one removes it.

## Risks

- Low risk. The predicate is the same conjunction of conditions. Only
the evaluation order and the probe shape change, and the benchmark
asserts complete-row equality on six fixtures and in both planner modes.
- Rows with the same `created_at` have no defined order, as before. The
benchmark uses distinct times so positional comparison is meaningful.
- `offset 0` stops PostgreSQL from turning these probes into joins. If a
task someday has a very large number of runs, each probe still uses the
company and issue indexes.
- Upgrade safety: this change reads saved run, wake and event state
only. It changes no persisted shape, run identity, or saved session, so
state saved by the previous release carries forward unchanged. A cutover
test is not applicable.
- Not verified on the hosted database. The local proof uses synthetic
data, and I did not capture a hosted plan. Confirm with a fresh query
statistics window after deployment.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use (shell,
file editing, local PostgreSQL) 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
- [x] All Paperclip CI gates are green
- [x] 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 2026-11-09 18:12:02 -06:00
1 parent 60a60a0cd1
commit ab80fea62f
4 files changed
+384 -27
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/__tests__/ownership-reconciliation-report.test.ts
+58
View File
@@ -723,6 +723,64 @@ describe("report-only ownership reconciliation", () => {
.toEqual([lost.commentId]);
});
// The check reads a task's runs through two separate bindings (legacy
// context or native column) and reads event payloads only for comments that
// the indexed checks keep. Each binding must count alone, and only in the
// comment's company.
it("binds each task run by its legacy context or its native column in the user-comment wake check", async () => {
const otherCompanyId = randomUUID(), otherAgentId = randomUUID();
await db.insert(companies).values({ id: otherCompanyId, name: "Other company", issuePrefix: "OTH" });
await db.insert(agents).values({ id: otherAgentId, companyId: otherCompanyId, name: "Other", role: "engineer", adapterType: "codex_local" });
const task = async (title: string, runs: Array<{ binding: "context" | "native" | "both"; status?: string;
startedAt: Date; finishedAt?: Date | null; companyId?: string }>, commentAt: Date) => {
const issueId = randomUUID();
await db.insert(issues).values({ id: issueId, companyId, title, status: "todo", assigneeAgentId: agentId });
const runIds: string[] = [];
for (const run of runs) {
const runCompanyId = run.companyId ?? companyId;
const [row] = await db.insert(heartbeatRuns).values({ companyId: runCompanyId, agentId: runCompanyId === companyId ? agentId : otherAgentId,
status: run.status ?? "succeeded", invocationSource: "manual",
contextSnapshot: run.binding === "native" ? {} : { issueId },
nativeIssueId: run.binding === "context" ? null : issueId,
startedAt: run.startedAt, createdAt: run.startedAt, finishedAt: run.finishedAt === undefined ? minutesAgo(3) : run.finishedAt }).returning();
runIds.push(row!.id);
}
const [comment] = await db.insert(issueComments).values({ companyId, issueId, authorUserId: "user-1", body: title,
createdAt: commentAt }).returning();
return { issueId, runIds, commentId: comment!.id };
};
const steerEvent = (runId: string, commentId: string, eventType: string, createdAt: Date) => ({ companyId, runId, agentId,
seq: 1, eventType, stream: "system", level: "info", sourceEventId: randomUUID(), createdAt,
payload: { prpEvent: { itemId: `provider:steer:${commentId}`, payload: { kind: "steering_acknowledgement" } } } });
const lostNative = await task("lost, native prior run", [{ binding: "native", startedAt: minutesAgo(10), finishedAt: minutesAgo(9) }], minutesAgo(8));
const lostContext = await task("lost, legacy prior run", [{ binding: "context", startedAt: minutesAgo(10), finishedAt: minutesAgo(9) }], minutesAgo(7));
await task("no run at all", [], minutesAgo(8));
await task("only another company's run", [{ binding: "both", startedAt: minutesAgo(10), finishedAt: minutesAgo(9), companyId: otherCompanyId }], minutesAgo(8));
await task("native later run", [{ binding: "context", startedAt: minutesAgo(10), finishedAt: minutesAgo(9) },
{ binding: "native", startedAt: minutesAgo(5) }], minutesAgo(8));
await task("legacy later run", [{ binding: "native", startedAt: minutesAgo(10), finishedAt: minutesAgo(9) },
{ binding: "context", startedAt: minutesAgo(5) }], minutesAgo(8));
await task("native live run", [{ binding: "native", status: "running", startedAt: minutesAgo(10), finishedAt: null }], minutesAgo(8));
const nativeSteered = await task("native steered run", [{ binding: "native", startedAt: minutesAgo(10) }], minutesAgo(8));
const contextSteered = await task("legacy steered run", [{ binding: "context", startedAt: minutesAgo(10) }], minutesAgo(8));
const earlyAck = await task("acknowledged before the comment", [{ binding: "native", startedAt: minutesAgo(10) }], minutesAgo(6));
const logOnly = await task("named only in a log event", [{ binding: "context", startedAt: minutesAgo(10) }], minutesAgo(5));
const otherRunAck = await task("acknowledged by a run that started after it", [{ binding: "native", startedAt: minutesAgo(10), finishedAt: minutesAgo(9) }], minutesAgo(4));
await db.insert(heartbeatRunEvents).values([
steerEvent(nativeSteered.runIds[0]!, nativeSteered.commentId, "item.completed", minutesAgo(7)),
steerEvent(contextSteered.runIds[0]!, contextSteered.commentId, "item.started", minutesAgo(7)),
steerEvent(earlyAck.runIds[0]!, earlyAck.commentId, "item.completed", minutesAgo(7)),
steerEvent(logOnly.runIds[0]!, logOnly.commentId, "log", minutesAgo(4)),
// The run that names this comment ended before it: not a steer of this comment.
steerEvent(otherRunAck.runIds[0]!, otherRunAck.commentId, "item.completed", minutesAgo(3)),
]);
const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false,
processStartedAt: minutesAgo(30) });
// Oldest first, as the check orders them.
expect(report.violations.filter(row => row.reason === "user_comment_without_wake").map(row => row.recordId))
.toEqual([lostNative.commentId, lostContext.commentId, earlyAck.commentId, logOnly.commentId, otherRunAck.commentId]);
});
it("keeps a backstop deadline for a lifted pause still inside its grace window", async () => {
const updatedAt = minutesAgo(1);
await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation", status: "deferred_issue_execution",
server/src/services/recovery/ownership-invariant-report.ts
+44 -27
View File
@@ -1,4 +1,4 @@
import { and, asc, eq, inArray, isNotNull, isNull, lt, notExists, or, sql } from "drizzle-orm";
import { and, asc, eq, inArray, isNotNull, isNull, lt, notExists, or, sql, type SQL } from "drizzle-orm";
import {
agentWakeupRequests,
heartbeatRuns,
@@ -467,36 +467,53 @@ async function checkStaleDroppedComments(db: Db, now: Date, limit: number): Prom
* finds a user comment on an assigned, non-conversation task with no live run
* or lock, newer than every run of the task, that no wake request carries.
* The reconciliation worker saves it as a deferred receipt and promotes it. */
async function checkUserCommentsWithoutWake(db: Db, now: Date, limit: number, processStartedAt: Date): Promise<{ violations: InvariantViolation[]; nextDueAt: Date | null }> {
export async function checkUserCommentsWithoutWake(db: Db, now: Date, limit: number, processStartedAt: Date): Promise<{ violations: InvariantViolation[]; nextDueAt: Date | null }> {
const since = new Date(now.getTime() - INTENT_LOOKBACK_MS).toISOString();
// A run belongs to the task by its legacy context or its native column. Each
// binding is a separate indexed probe per comment: one OR across both lets
// PostgreSQL hash every run of the company and read each run's context.
// `offset 0` keeps each probe, and the steer check's event lookup, a
// per-row index lookup; without it PostgreSQL can also build a hashed
// subplan or a hash join over all runs or all events of the company.
const taskRuns = (alias: string, condition: SQL) => {
const run = sql.raw(alias);
return sql`(exists (select 1 from heartbeat_runs ${run} where ${run}.company_id = c.company_id
and ${run}.context_snapshot->>'issueId' = c.issue_id::text and ${condition} offset 0)
or exists (select 1 from heartbeat_runs ${run} where ${run}.company_id = c.company_id
and ${run}.native_issue_id = c.issue_id and ${condition} offset 0))`;
};
// Three stages, cheapest first. Most user comments have a wake (stage one)
// or a later run (stage two). Only the few comments left reach the steer
// check, which reads run event payloads.
const rows = await db.execute<{ id: string; company_id: string; issue_id: string; created_at: string }>(sql`
select c.id, c.company_id, c.issue_id, c.created_at::text as created_at from issue_comments c
join issues i on i.id = c.issue_id and i.company_id = c.company_id
where c.author_user_id is not null and c.deleted_at is null
and c.author_agent_id is null and c.created_by_run_id is null
and c.created_at >= ${since}::timestamptz
and i.assignee_agent_id is not null and i.conversation_agent_id is null
and (i.status in ('todo', 'in_progress')
or (i.status = 'done' and ${taskClosedByAgentSql(sql`c.company_id`, sql`c.issue_id`)}))
and i.execution_run_id is null
and not exists (select 1 from heartbeat_runs steered join heartbeat_run_events steer_event
on steer_event.company_id = steered.company_id and steer_event.run_id = steered.id
where steered.company_id = c.company_id
and (steered.context_snapshot->>'issueId' = c.issue_id::text or steered.native_issue_id = c.issue_id)
and coalesce(steered.started_at, steered.created_at) <= c.created_at
and (steered.finished_at is null or steered.finished_at >= c.created_at)
with unwoken as materialized (
select c.id, c.company_id, c.issue_id, c.created_at from issue_comments c
join issues i on i.id = c.issue_id and i.company_id = c.company_id
where c.author_user_id is not null and c.deleted_at is null
and c.author_agent_id is null and c.created_by_run_id is null
and c.created_at >= ${since}::timestamptz
and i.assignee_agent_id is not null and i.conversation_agent_id is null
and (i.status in ('todo', 'in_progress')
or (i.status = 'done' and ${taskClosedByAgentSql(sql`c.company_id`, sql`c.issue_id`)}))
and i.execution_run_id is null
and not exists (select 1 from agent_wakeup_requests w where w.company_id = c.company_id
and w.payload->>'issueId' = c.issue_id::text
and (w.payload->>'commentId' = c.id::text
or coalesce(w.payload#>'{_paperclipWakeContext,wakeCommentIds}', '[]'::jsonb) ? c.id::text))
), candidate as materialized (
select c.id, c.company_id, c.issue_id, c.created_at from unwoken c
where not ${taskRuns("r", sql`(r.status in ('queued', 'running', 'scheduled_retry')
or coalesce(r.started_at, r.created_at) >= c.created_at)`)}
and ${taskRuns("prior", sql`true`)}
)
select c.id, c.company_id, c.issue_id, c.created_at::text as created_at from candidate c
where not ${taskRuns("steered", sql`coalesce(steered.started_at, steered.created_at) <= c.created_at
and (steered.finished_at is null or steered.finished_at >= c.created_at)
and exists (select 1 from heartbeat_run_events steer_event
where steer_event.company_id = steered.company_id and steer_event.run_id = steered.id
and steer_event.event_type in ('item.started', 'item.completed')
and steer_event.created_at >= c.created_at
and strpos(steer_event.payload::text, c.id::text) > 0)
and exists (select 1 from heartbeat_runs prior where prior.company_id = c.company_id
and (prior.context_snapshot->>'issueId' = c.issue_id::text or prior.native_issue_id = c.issue_id))
and not exists (select 1 from heartbeat_runs r where r.company_id = c.company_id
and (r.context_snapshot->>'issueId' = c.issue_id::text or r.native_issue_id = c.issue_id)
and (r.status in ('queued', 'running', 'scheduled_retry') or coalesce(r.started_at, r.created_at) >= c.created_at))
and not exists (select 1 from agent_wakeup_requests w where w.company_id = c.company_id
and w.payload->>'issueId' = c.issue_id::text
and (w.payload->>'commentId' = c.id::text
or coalesce(w.payload#>'{_paperclipWakeContext,wakeCommentIds}', '[]'::jsonb) ? c.id::text))
and strpos(steer_event.payload::text, c.id::text) > 0 offset 0)`)}
order by c.created_at limit ${Math.min(limit, MAX_RECORDS_PER_KIND)}`);
// A comment inside the grace may still get its normal wake: check it again
// when the grace ends (a saved deadline), and report only older ones. This