diff --git a/doc/benchmarks/2026-11-03-active-recovery-projection.json b/doc/benchmarks/2026-11-03-active-recovery-projection.json new file mode 100644 index 00000000000..53e0e73328d --- /dev/null +++ b/doc/benchmarks/2026-11-03-active-recovery-projection.json @@ -0,0 +1,619 @@ +{ + "baseline": "a497bae83a128feae8206f27e4930e6407a67630", + "runtime": { + "node": "24.22.2", + "platform": "darwin", + "arch": "arm64", + "mode": "source Vitest + disposable PostgreSQL" + }, + "evidenceRun": "unintended-repeat; separately retained raw evidence; not pooled with original summary-only run", + "observer": "No debug or row-transform observer on timed client. Separate warmed clients for corrected counts.", + "samples": [ + { + "size": 1, + "round": 3, + "arm": "treatment", + "cpuMs": 0.763, + "wallMs": 1.6585830000003625 + }, + { + "size": 1, + "round": 3, + "arm": "baseline", + "cpuMs": 1.341, + "wallMs": 1.7993749999995998 + }, + { + "size": 1, + "round": 4, + "arm": "baseline", + "cpuMs": 1.378, + "wallMs": 1.4206670000003214 + }, + { + "size": 1, + "round": 4, + "arm": "treatment", + "cpuMs": 1.42, + "wallMs": 1.4197500000000218 + }, + { + "size": 1, + "round": 5, + "arm": "treatment", + "cpuMs": 0.746, + "wallMs": 1.396708999999646 + }, + { + "size": 1, + "round": 5, + "arm": "baseline", + "cpuMs": 1.468, + "wallMs": 1.7617499999996653 + }, + { + "size": 1, + "round": 6, + "arm": "baseline", + "cpuMs": 1.209, + "wallMs": 1.3915839999999662 + }, + { + "size": 1, + "round": 6, + "arm": "treatment", + "cpuMs": 0.641, + "wallMs": 0.9872919999997976 + }, + { + "size": 1, + "round": 7, + "arm": "treatment", + "cpuMs": 0.68, + "wallMs": 1.2235829999999623 + }, + { + "size": 1, + "round": 7, + "arm": "baseline", + "cpuMs": 1.23, + "wallMs": 1.524833000000399 + }, + { + "size": 1, + "round": 8, + "arm": "baseline", + "cpuMs": 1.495, + "wallMs": 1.3648330000005444 + }, + { + "size": 1, + "round": 8, + "arm": "treatment", + "cpuMs": 0.79, + "wallMs": 1.048584000000119 + }, + { + "size": 1, + "round": 9, + "arm": "treatment", + "cpuMs": 1.149, + "wallMs": 0.9705829999993512 + }, + { + "size": 1, + "round": 9, + "arm": "baseline", + "cpuMs": 3.502, + "wallMs": 1.666417000000365 + }, + { + "size": 1, + "round": 10, + "arm": "baseline", + "cpuMs": 2.278, + "wallMs": 1.311374999999316 + }, + { + "size": 1, + "round": 10, + "arm": "treatment", + "cpuMs": 1.179, + "wallMs": 1.0157079999999041 + }, + { + "size": 1, + "round": 11, + "arm": "treatment", + "cpuMs": 0.67, + "wallMs": 0.9623750000000655 + }, + { + "size": 1, + "round": 11, + "arm": "baseline", + "cpuMs": 1.117, + "wallMs": 1.2188329999999041 + }, + { + "size": 8, + "round": 3, + "arm": "treatment", + "cpuMs": 3.035, + "wallMs": 4.044708000000355 + }, + { + "size": 8, + "round": 3, + "arm": "baseline", + "cpuMs": 7.091, + "wallMs": 7.024334000000636 + }, + { + "size": 8, + "round": 4, + "arm": "baseline", + "cpuMs": 14.74, + "wallMs": 6.817332999999962 + }, + { + "size": 8, + "round": 4, + "arm": "treatment", + "cpuMs": 5.515, + "wallMs": 4.373458000000028 + }, + { + "size": 8, + "round": 5, + "arm": "treatment", + "cpuMs": 6.001, + "wallMs": 3.829291000000012 + }, + { + "size": 8, + "round": 5, + "arm": "baseline", + "cpuMs": 6.11, + "wallMs": 6.322791999999936 + }, + { + "size": 8, + "round": 6, + "arm": "baseline", + "cpuMs": 10.566, + "wallMs": 6.322791999999936 + }, + { + "size": 8, + "round": 6, + "arm": "treatment", + "cpuMs": 2.181, + "wallMs": 3.694792000000234 + }, + { + "size": 8, + "round": 7, + "arm": "treatment", + "cpuMs": 2.367, + "wallMs": 3.7172920000002705 + }, + { + "size": 8, + "round": 7, + "arm": "baseline", + "cpuMs": 20.971, + "wallMs": 9.899207999999817 + }, + { + "size": 8, + "round": 8, + "arm": "baseline", + "cpuMs": 8.928, + "wallMs": 6.159249999999702 + }, + { + "size": 8, + "round": 8, + "arm": "treatment", + "cpuMs": 4.288, + "wallMs": 4.029959000000417 + }, + { + "size": 8, + "round": 9, + "arm": "treatment", + "cpuMs": 2.886, + "wallMs": 4.439124999999876 + }, + { + "size": 8, + "round": 9, + "arm": "baseline", + "cpuMs": 6.527, + "wallMs": 7.623250000000553 + }, + { + "size": 8, + "round": 10, + "arm": "baseline", + "cpuMs": 6.644, + "wallMs": 6.614042000000154 + }, + { + "size": 8, + "round": 10, + "arm": "treatment", + "cpuMs": 2.882, + "wallMs": 3.775292000000263 + }, + { + "size": 8, + "round": 11, + "arm": "treatment", + "cpuMs": 2.301, + "wallMs": 3.774250000000393 + }, + { + "size": 8, + "round": 11, + "arm": "baseline", + "cpuMs": 6.854, + "wallMs": 6.11029199999939 + }, + { + "size": 32, + "round": 3, + "arm": "treatment", + "cpuMs": 8.231, + "wallMs": 11.996124999999665 + }, + { + "size": 32, + "round": 3, + "arm": "baseline", + "cpuMs": 20.818, + "wallMs": 21.149207999998907 + }, + { + "size": 32, + "round": 4, + "arm": "baseline", + "cpuMs": 27.948, + "wallMs": 23.184290999999575 + }, + { + "size": 32, + "round": 4, + "arm": "treatment", + "cpuMs": 8.843, + "wallMs": 12.473207999999431 + }, + { + "size": 32, + "round": 5, + "arm": "treatment", + "cpuMs": 7.363, + "wallMs": 12.15533399999913 + }, + { + "size": 32, + "round": 5, + "arm": "baseline", + "cpuMs": 18.398, + "wallMs": 20.87029100000109 + }, + { + "size": 32, + "round": 6, + "arm": "baseline", + "cpuMs": 36.55, + "wallMs": 23.66070899999977 + }, + { + "size": 32, + "round": 6, + "arm": "treatment", + "cpuMs": 9.292, + "wallMs": 12.562832999999955 + }, + { + "size": 32, + "round": 7, + "arm": "treatment", + "cpuMs": 8.918, + "wallMs": 12.686416999998983 + }, + { + "size": 32, + "round": 7, + "arm": "baseline", + "cpuMs": 21.922, + "wallMs": 20.46008299999994 + }, + { + "size": 32, + "round": 8, + "arm": "baseline", + "cpuMs": 30.474, + "wallMs": 22.94824999999946 + }, + { + "size": 32, + "round": 8, + "arm": "treatment", + "cpuMs": 8.374, + "wallMs": 12.847208999999566 + }, + { + "size": 32, + "round": 9, + "arm": "treatment", + "cpuMs": 7.612, + "wallMs": 12.91295799999898 + }, + { + "size": 32, + "round": 9, + "arm": "baseline", + "cpuMs": 17.255, + "wallMs": 21.539458000001105 + }, + { + "size": 32, + "round": 10, + "arm": "baseline", + "cpuMs": 36.617, + "wallMs": 24.36845799999901 + }, + { + "size": 32, + "round": 10, + "arm": "treatment", + "cpuMs": 8.035, + "wallMs": 13.467709000000468 + }, + { + "size": 32, + "round": 11, + "arm": "treatment", + "cpuMs": 9.651, + "wallMs": 13.0765420000007 + }, + { + "size": 32, + "round": 11, + "arm": "baseline", + "cpuMs": 19.53, + "wallMs": 21.1825000000008 + }, + { + "size": 100, + "round": 3, + "arm": "treatment", + "cpuMs": 30.24, + "wallMs": 39.02287499999875 + }, + { + "size": 100, + "round": 3, + "arm": "baseline", + "cpuMs": 60.393, + "wallMs": 64.40520800000013 + }, + { + "size": 100, + "round": 4, + "arm": "baseline", + "cpuMs": 68.959, + "wallMs": 65.32254199999988 + }, + { + "size": 100, + "round": 4, + "arm": "treatment", + "cpuMs": 32.718, + "wallMs": 40.862082999999984 + }, + { + "size": 100, + "round": 5, + "arm": "treatment", + "cpuMs": 22.919, + "wallMs": 37.28120800000033 + }, + { + "size": 100, + "round": 5, + "arm": "baseline", + "cpuMs": 67.8, + "wallMs": 69.60349999999926 + }, + { + "size": 100, + "round": 6, + "arm": "baseline", + "cpuMs": 74.871, + "wallMs": 69.87325000000055 + }, + { + "size": 100, + "round": 6, + "arm": "treatment", + "cpuMs": 22.672, + "wallMs": 38.44508300000052 + }, + { + "size": 100, + "round": 7, + "arm": "treatment", + "cpuMs": 34, + "wallMs": 40.74645799999962 + }, + { + "size": 100, + "round": 7, + "arm": "baseline", + "cpuMs": 64.48, + "wallMs": 68.01870800000142 + }, + { + "size": 100, + "round": 8, + "arm": "baseline", + "cpuMs": 68.384, + "wallMs": 68.15237499999967 + }, + { + "size": 100, + "round": 8, + "arm": "treatment", + "cpuMs": 34.617, + "wallMs": 40.32725000000028 + }, + { + "size": 100, + "round": 9, + "arm": "treatment", + "cpuMs": 37.502, + "wallMs": 41.068459000000075 + }, + { + "size": 100, + "round": 9, + "arm": "baseline", + "cpuMs": 70.518, + "wallMs": 69.73304199999984 + }, + { + "size": 100, + "round": 10, + "arm": "baseline", + "cpuMs": 62.568, + "wallMs": 67.05624999999964 + }, + { + "size": 100, + "round": 10, + "arm": "treatment", + "cpuMs": 33.52, + "wallMs": 40.88299999999981 + }, + { + "size": 100, + "round": 11, + "arm": "treatment", + "cpuMs": 22.194, + "wallMs": 38.63199999999961 + }, + { + "size": 100, + "round": 11, + "arm": "baseline", + "cpuMs": 71.12, + "wallMs": 67.40216699999837 + } + ], + "counts": [ + { + "size": 1, + "baseline": { + "queries": 11, + "rows": 1 + }, + "treatment": { + "queries": 5, + "rows": 2 + } + }, + { + "size": 8, + "baseline": { + "queries": 81, + "rows": 8 + }, + "treatment": { + "queries": 33, + "rows": 16 + } + }, + { + "size": 32, + "baseline": { + "queries": 321, + "rows": 32 + }, + "treatment": { + "queries": 129, + "rows": 64 + } + }, + { + "size": 100, + "baseline": { + "queries": 1001, + "rows": 100 + }, + "treatment": { + "queries": 401, + "rows": 200 + } + } + ], + "rawHashes": [ + { + "path": "paired-1.json", + "sha256": "37c0303a1e1f239c031d43d6f777ad2aa7941f26cd4d5f0a754f4176d9bfa866", + "bytes": 28269, + "mtimeUtc": "2026-11-03T14:57:54.061424+00:00" + }, + { + "path": "paired-100.json", + "sha256": "c940cd58c7d9304a7a6dcd2760f0568f13e69f308f23a25b67ace391457418dd", + "bytes": 1284197, + "mtimeUtc": "2026-11-03T14:57:57.648820+00:00" + }, + { + "path": "paired-mixed.json", + "sha256": "8e60615b908d7758bcc42f27b4bd8d6ff50941848e43e24ac68e2157d978106b", + "bytes": 435894, + "mtimeUtc": "2026-11-03T14:57:58.185272+00:00" + }, + { + "path": "paired-summary.json", + "sha256": "0b5a3d3abef8dd631721a528429785749dd6dee258b8f7cd2df486acbfce9d1a", + "bytes": 8078, + "mtimeUtc": "2026-11-03T14:57:10.841853+00:00" + }, + { + "path": "paired-8.json", + "sha256": "798b765c70b83eeccbc9a1b00065de4655999c614492a059d69bf7ccdde0f302", + "bytes": 116984, + "mtimeUtc": "2026-11-03T14:57:54.809182+00:00" + }, + { + "path": "paired-32.json", + "sha256": "bcd6170aae651a63d19ee2c7a2a6b2e34bfe3d512d2377cc073c0f5c2985eaed", + "bytes": 421480, + "mtimeUtc": "2026-11-03T14:57:55.794276+00:00" + }, + { + "path": "late-true.json", + "sha256": "71fcd67113cd6b93120164924f28395888c09c6535decd5e4f35f5b2e06b97f1", + "bytes": 1180, + "mtimeUtc": "2026-11-03T14:57:59.127832+00:00" + }, + { + "path": "late-false.json", + "sha256": "295122048af56a9cb82a572ebfd02a6e5b7012f6959ef6b144b11871dbd65e13", + "bytes": 1245, + "mtimeUtc": "2026-11-03T14:57:58.663439+00:00" + }, + { + "path": "proof-corrected.log", + "sha256": "cb543676a7ece2127bf74586833625d5afb73c5fa339cbd94640caf13febb0bf", + "bytes": 475, + "mtimeUtc": "2026-11-03T14:58:00.792619+00:00" + } + ] +} diff --git a/doc/benchmarks/2026-11-03-active-recovery-projection.md b/doc/benchmarks/2026-11-03-active-recovery-projection.md new file mode 100644 index 00000000000..0ea184ce29f --- /dev/null +++ b/doc/benchmarks/2026-11-03-active-recovery-projection.md @@ -0,0 +1,52 @@ +# Active recovery current-wait projection + +An active recovery check needs a current durable-wait reason. It previously built the full seven-query retry fingerprint state for every admitted source action. This change reads the three needed existence facts in one statement. The full fingerprint collector keeps all of its inputs. Both readers use the same reason policy. Child checks, outside-action run/wake checks, native safety holds and mutation services stay in place. + +## Local evidence + +Baseline: `a497bae83a128feae8206f27e4930e6407a67630`. Node24.22.2, macOS arm64, source Vitest and disposable PostgreSQL. The actual `recoveryService.reconcileActiveRecoveryActions` ran against synthetic cohorts with mostly empty wait facts, approximately75%board and25%legacy agent recovery owners. A second company's records stayed unchanged. The live company/owner/cohort mix is unknown. + +Nine alternating pairs per cohort follow three warmup pairs. The timed database client has no SQL debug or row-transform observer. Count proof uses separate warmed clients outside timing. Node process CPU includes JIT/GC/driver work and excludes PostgreSQL process CPU. The [data file](2026-11-03-active-recovery-projection.json) contains all72retained primary timing samples, independent count proof and raw-file hashes. + +| Actions | CPU median before → after (ms) | CPU paired wins | Wall median before → after (ms) | Application queries before → after | +|---:|---:|---:|---:|---:| +|1|1.378 →0.763|8/9|1.421 →1.049|11 →5| +|8|7.091 →2.886|9/9|6.614 →3.829|81 →33| +|32|21.922 →8.374|9/9|21.539 →12.686|321 →129| +|100|68.384 →32.718|9/9|68.019 →40.327|1001 →401| + +All nine wall pairs improve at8/32/100. CPU ranges at8 are[6.110,20.971]before and[2.181,6.001]after; at32,[17.255,36.617]and[7.363,9.651]; at100,[60.393,74.871]and[22.194,37.502]. Size1is noisy. Driver rows rise from N to2N because each EXISTS statement returns one boolean row. This is reduced query construction/dispatch and unused fingerprint work, not reduced returned rows. + +The source-consistent census pattern suggests about162outer action checks/30s, but that caller partition is unproven and was observed only3m19safter startup readiness. Scaling the local per-action difference to that assumed mix would yield only a few milliseconds CPU/s. It is not measured fleet savings, a hard ceiling, or a deployed improvement. No request latency, production-build, Sentry or OTLP result is claimed. + +## Correctness + +The real-service proof preserves exact results, action/issue/wake/activity/comment records and queue notifications between baseline and treatment on the same fixture. Relation edges preserve exact semantic fields; generated edge IDs and timestamps are excluded from that comparison. Mixed controls cover user/blocker/monitor/execution/interaction/approval reason priority, board ownership, same-action versus outside run paths, healthy children and native reconciliation holds. Full retry fingerprints match the pinned collector on fresh mixed issue inputs. + +A deterministic gate pauses after the source cohort read and before the new statement. An independent client then commits a pending interaction. The check sees that fact. A second case also resolves the action independently; the existing conditional mutation preserves that decision. These tests prove fresh query facts and the current action update guard. They do not make a single statement's snapshot identical to seven concurrent statement snapshots. Scalar issue fields still come from the source cohort read. + +Permanent PostgreSQL regressions live in `server/src/__tests__/recovery-durable-wait-projection.test.ts`. + +## Reproduce in an isolated checkout + +Use installed dependencies and Node24.11or newer. Keep CPU measurement exclusive. The manual helper creates a unique evidence directory, materializes the pinned baseline and fixture, and removes only its own temporary source files afterward. It refuses existing temporary paths. It does not run in the default test suite. + +```sh +node scripts/benchmark-active-recovery-projection.mjs --baseline-ref=a497bae83a128feae8206f27e4930e6407a67630 +``` + +For only count and correctness proof, with all four timing tests excluded: + +```sh +node scripts/benchmark-active-recovery-projection.mjs --baseline-ref=a497bae83a128feae8206f27e4930e6407a67630 --proof-only +``` + +An optional `--evidence-dir=/absolute/parent` places a unique run directory under that parent. Preserve the printed directory and command log. No external provider is called. + +## Evidence incident and limits + +The first observer-free run had an exclusive team timing reservation and passed all seven controls. Its aggregate summary and log remain, but a later harness error overwrote its raw samples. Those original samples are lost and are not reconstructed or pooled into the table above. + +An intended untimed proof edit failed because it used a root-relative file path from the server directory. The shell then ran the unchanged timing fixture. The resulting unintended repeat is the separately retained primary raw evidence above. It had no renewed quiet reservation; unrelated system activity cannot be excluded. The direction and ranges support the same local conclusion as the earlier summary, but this provenance limit remains. No third CPU run was made. The corrected proof used a verified explicit test filter:7passed,4timing tests skipped, all count-proof sample arrays empty. Initial count clients included PostgreSQL type discovery; those counts were superseded by the separate warmed count proof shown above. + +The reproducible helper now creates a unique run directory. Edit, validation and execution were separated for the corrected proof. This change has not been deployed. Serving CPU and real workload acceptance remain separate follow-up work. diff --git a/doc/benchmarks/fixtures/active-recovery-projection.fixture.ts b/doc/benchmarks/fixtures/active-recovery-projection.fixture.ts new file mode 100644 index 00000000000..443de362d47 --- /dev/null +++ b/doc/benchmarks/fixtures/active-recovery-projection.fixture.ts @@ -0,0 +1,160 @@ +import { randomUUID } from "node:crypto"; +import { writeFileSync } from "node:fs"; +import { and, eq, sql } from "drizzle-orm"; +import { PgDialect } from "drizzle-orm/pg-core"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { createDb, companies, agents, issues, issueRecoveryActions, heartbeatRuns, agentWakeupRequests, + approvals, issueApprovals, issueRelations, issueThreadInteractions, activityLog, issueComments, subscribeDatabaseWork, type Db } from "@paperclipai/db"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { recoveryService } from "../services/recovery/service.js"; +import { recoveryService as baselineService } from "../services/recovery/.bench-baseline-service.js"; +import { collectDispositionRepairSourceState } from "../services/recovery/disposition-repair.js"; +import { collectDispositionRepairSourceState as baselineState } from "../services/recovery/.bench-baseline-disposition-repair.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; + +const evidence = process.env.RECOVERY_EVIDENCE_DIR; +if (!evidence) throw new Error("Run scripts/benchmark-active-recovery-projection.mjs to select a unique evidence directory."); +describe("actual active recovery attribution", () => { + let temporary: Awaited>; + let db: Db; let companyId: string, agentId: string; + + const notifications: unknown[]=[]; const stops:Array<()=>void>=[]; + beforeAll(async()=>{ + temporary=await startEmbeddedPostgresTestDatabase("active-recovery-projection-"); + db=createDb(temporary.connectionString,{maxConnections:6,idleTimeoutSeconds:0,maxLifetimeSeconds:0}); + db.$client.options.onnotice=()=>{}; + for(const topic of Object.values(DELIVERY_QUEUES)) stops.push(subscribeDatabaseWork(db,topic,(event,hints)=>notifications.push({topic,event,hints:[...hints??[]]}))); + },30000); + afterEach(async()=>{vi.restoreAllMocks();vi.useRealTimers();await db.execute(sql`truncate companies cascade`);}); + afterAll(async()=>{for(const stop of stops)stop();await db?.$client.end({timeout:1});await temporary?.cleanup();}); + async function seed(size:number){ + companyId=randomUUID();agentId=randomUUID(); + await db.insert(companies).values({id:companyId,name:"Projection",issuePrefix:`P${companyId.slice(0,6)}`}); + await db.insert(agents).values({id:agentId,companyId,name:"Owner",role:"engineer",status:"idle"}); + const roots=await db.insert(issues).values(Array.from({length:size},(_,i)=>({companyId,title:`source-${i}`,status:"blocked",assigneeAgentId:agentId}))).returning(); + const actions=await db.insert(issueRecoveryActions).values(roots.map((r,i)=>({companyId,sourceIssueId:r.id,kind:"deliberate_wait_without_target",ownerType:i%4===0?"agent":"board",ownerAgentId:i%4===0?agentId:null,cause:"deliberate_wait_without_target",fingerprint:`fixture-${i}`,nextAction:"Board decision",wakePolicy:{type:i%4===0?"bounded_recovery_owner":"board_escalation"}}))).returning(); + return {roots,actions}; + } + async function snapshot(database:Db=db){return { + actions:await database.select().from(issueRecoveryActions).orderBy(issueRecoveryActions.id), + issues:await database.select().from(issues).orderBy(issues.id), + wakes:await database.select().from(agentWakeupRequests).orderBy(agentWakeupRequests.id), + activity:await database.select().from(activityLog).orderBy(activityLog.id), + comments:await database.select().from(issueComments).orderBy(issueComments.id), + relations:(await database.select().from(issueRelations).orderBy(issueRelations.issueId,issueRelations.relatedIssueId)).map(({id,createdAt,updatedAt,...edge})=>edge), + };} + async function emptyCohort(size:number,includeTiming:boolean){ + const {roots,actions}=await seed(size); + const foreign=randomUUID();await db.insert(companies).values({id:foreign,name:"Foreign",issuePrefix:`F${foreign.slice(0,6)}`}); + const [foreignIssue]=await db.insert(issues).values({companyId:foreign,title:"foreign",status:"blocked"}).returning(); + await db.insert(issueRecoveryActions).values({companyId:foreign,sourceIssueId:foreignIssue.id,kind:"deliberate_wait_without_target",ownerType:"board",cause:"deliberate_wait_without_target",fingerprint:"foreign",nextAction:"Decide"}); + const before=await snapshot();notifications.length=0; + const enqueueWakeup=async()=>{throw new Error("Unexpected enqueue");}; + const services={baseline:baselineService(db,{enqueueWakeup}),treatment:recoveryService(db,{enqueueWakeup})}; + const counts:Record={}; + for(const [arm,factory]of Object.entries({baseline:baselineService,treatment:recoveryService})){ + const countDb=createDb(temporary.connectionString,{maxConnections:6,idleTimeoutSeconds:0,maxLifetimeSeconds:0}); + await Promise.all(Array.from({length:6},()=>countDb.execute(sql`select pg_sleep(0.01)`))); + const statements:string[]=[];let rows=0; + countDb.$client.options.debug=(_c,q)=>statements.push(q); + countDb.$client.options.transform.row.from=(row:any)=>{rows++;return row;}; + try{const result=await factory(countDb,{enqueueWakeup}).reconcileActiveRecoveryActions({companyId});counts[arm]={queries:statements.length,rows,sql:statements,result};} + finally{await countDb.$client.end({timeout:1});} + } + // CPU client has no debug or row-transform observer installed. + expect(db.$client.options.debug).toBe(false); + expect(db.$client.options.transform.row.from).toBeUndefined(); + const samples=[];let exact:any=(counts.baseline as any).result; + expect((counts.treatment as any).result).toEqual(exact); + expect(exact).toEqual({requeued:0,escalated:0,resolved:0,skipped:actions.filter(a=>a.ownerType==="agent").length,issueIds:[]}); + for(let round=0;round<(includeTiming?12:0);round++){ + for(const arm of (round%2?["treatment","baseline"]:["baseline","treatment"])as Array){ + const cpu=process.cpuUsage(),start=performance.now(); + const result=await services[arm].reconcileActiveRecoveryActions({companyId}); + const wallMs=performance.now()-start,used=process.cpuUsage(cpu); + if(exact)expect(result).toEqual(exact);else exact=result; + expect(result).toEqual({requeued:0,escalated:0,resolved:0,skipped:actions.filter(a=>a.ownerType==="agent").length,issueIds:[]}); + if(round>=3)samples.push({round,arm,cpuMs:(used.user+used.system)/1000,wallMs}); + } + } + expect(await snapshot()).toEqual(before);expect(notifications).toEqual([]); + writeFileSync(`${evidence}/${includeTiming ? "paired" : "proof"}-${size}.json`,JSON.stringify({size,node:process.version,samples,counts,exact,before,after:await snapshot(),notifications},null,2),{flag:"wx"}); + } + it.each([1,8,32,100])("measures mostly empty cohort %i",size=>emptyCohort(size,true),60000); + it.each([1,8,32,100])("untimed count and empty-state proof %i",size=>emptyCohort(size,false),60000); + it("records mixed current wait and ownership controls",async()=>{ + const {roots,actions}=await seed(12); + const pending={status:"pending",currentStageId:null,currentStageIndex:null,currentStageType:null,currentParticipant:null,returnAssignee:null,completedStageIds:[],lastDecisionId:null,lastDecisionOutcome:null}; + // Four agent-owned rows select user, blocker, interaction and approval reasons. + for(const i of [1,2,3,4,5]) await db.update(issueRecoveryActions).set({ownerType:"agent",ownerAgentId:agentId,wakePolicy:{type:"bounded_recovery_owner"}}).where(eq(issueRecoveryActions.id,actions[i].id)); + await db.update(issues).set({assigneeUserId:"board-user",assigneeAgentId:null,monitorNextCheckAt:new Date(Date.now()+3600000),executionState:pending}).where(eq(issues.id,roots[0].id)); + const [blocker]=await db.insert(issues).values({companyId,title:"blocker",status:"todo"}).returning(); + await db.insert(issueRelations).values({companyId,issueId:blocker.id,relatedIssueId:roots[1].id,type:"blocks"}); + await db.update(issues).set({monitorNextCheckAt:new Date(Date.now()+3600000),executionState:pending}).where(eq(issues.id,roots[1].id)); + await db.update(issues).set({monitorNextCheckAt:new Date(Date.now()+3600000),executionState:pending}).where(eq(issues.id,roots[2].id)); + await db.update(issues).set({executionState:pending}).where(eq(issues.id,roots[3].id)); + await db.insert(issueThreadInteractions).values([0,1,2,3,4,6].map(i=>({companyId,issueId:roots[i].id,kind:"request_confirmation",status:"pending",payload:{version:1,prompt:"Decide"}}))); + const ap=await db.insert(approvals).values([4,5].map(i=>({companyId,type:"hire_agent",payload:{index:i},status:"revision_requested"}))).returning(); + await db.insert(issueApprovals).values(ap.map((a,i)=>({companyId,issueId:roots[i+4].id,approvalId:a.id}))); + // Board interaction alone remains active. Same-action paths cannot clear recovery. + await db.insert(heartbeatRuns).values({companyId,agentId,status:"running",contextSnapshot:{issueId:roots[7].id,recoveryActionId:actions[7].id}}); + await db.insert(agentWakeupRequests).values({companyId,agentId,source:"system",status:"queued",payload:{issueId:roots[7].id,recoveryActionId:actions[7].id}}); + // An independent path does clear a board action. + await db.insert(heartbeatRuns).values({companyId,agentId,status:"running",contextSnapshot:{taskId:roots[9].id}}); + // Native reconciliation/export safety hold remains even with an independent path. + await db.update(issueRecoveryActions).set({cause:"native_workspace_finalization_owner_unverified"}).where(eq(issueRecoveryActions.id,actions[10].id)); + await db.insert(heartbeatRuns).values({companyId,agentId,status:"running",contextSnapshot:{issueId:roots[10].id}}); + const [child]=await db.insert(issues).values({companyId,title:"healthy-child",status:"todo",parentId:roots[11].id,assigneeAgentId:agentId}).returning(); + await db.insert(issueThreadInteractions).values({companyId,issueId:child.id,kind:"request_confirmation",status:"pending",payload:{version:1,prompt:"Child decision"}}); + for(const issue of await db.select().from(issues).where(eq(issues.companyId,companyId))) expect(await collectDispositionRepairSourceState(db,{issue})).toEqual(await baselineState(db,{issue})); + const before=await snapshot(); + const outcomes:any[]=[]; + vi.useFakeTimers({now:new Date("2026-11-03T15:00:00Z"),toFake:["Date"]}); + const rollback=new Error("owned fixture rollback"); + for(const factory of [baselineService,recoveryService]){ + try{await db.transaction(async tx=>{ + notifications.length=0; + const result=await factory(tx as unknown as Db,{enqueueWakeup:async()=>{throw new Error("Unexpected enqueue");}}).reconcileActiveRecoveryActions({companyId}); + const after=await snapshot(tx as unknown as Db); + outcomes.push({result,after,notifications:[...notifications]}); + throw rollback; + });}catch(error){if(error!==rollback)throw error;} + } + vi.useRealTimers(); + expect(outcomes[1]).toEqual(outcomes[0]); + expect(await snapshot()).toEqual(before); + const {result,after,notifications:observedNotifications}=outcomes[1]; + const state=(i:number)=>after.actions.find(a=>a.id===actions[i].id)!; + for(const [i,reason]of [[0,"user_owner"],[1,"blocker"],[2,"monitor"],[3,"execution_stage"],[4,"interaction"],[5,"approval"]]as const) expect(state(i)).toMatchObject({status:"resolved",resolutionNote:`durable_path_restored:${reason}`}); + for(const i of [6,7,10])expect(state(i).status).toBe("active"); + expect(state(9)).toMatchObject({status:"resolved",resolutionNote:"new_source_execution_path"}); + expect(state(11)).toMatchObject({status:"resolved",resolutionNote:"durable_path_restored:healthy_child"}); + expect(after.wakes).toEqual(before.wakes); + writeFileSync(`${evidence}/paired-mixed.json`,JSON.stringify({result,before,after,notifications:observedNotifications,outcomes},null,2),{flag:"wx"}); + },60000); + + it.each([false,true])("reads late query facts and preserves concurrent resolution: %s",async operatorResolves=>{ + const {roots,actions}=await seed(1); + let started!:()=>void,release!:()=>void; + const ready=new Promise(r=>{started=r;});const gate=new Promise(r=>{release=r;}); + const execute=db.execute.bind(db),dialect=new PgDialect(); + const observer=vi.spyOn(db,"execute").mockImplementation(((query:any)=>{ + if(dialect.sqlToQuery(query).sql.includes('as "hasBlocker"')){started();return gate.then(()=>execute(query));} + return execute(query); + })as typeof db.execute); + const running=recoveryService(db,{enqueueWakeup:async()=>{throw new Error("Unexpected enqueue");}}).reconcileActiveRecoveryActions({companyId}); + const operator=createDb(temporary.connectionString,{maxConnections:1}); + try{ + let timer:ReturnType|undefined; + try{await Promise.race([ready,new Promise((_,reject)=>{timer=setTimeout(()=>reject(new Error("fresh read gate not reached")),3000);})]);}finally{clearTimeout(timer);} + await operator.insert(issueThreadInteractions).values({companyId,issueId:roots[0].id,kind:"request_confirmation",status:"pending",payload:{version:1,prompt:"Late decision"}}); + if(operatorResolves)await operator.update(issueRecoveryActions).set({status:"resolved",resolutionNote:"concurrent_board_decision",resolvedAt:new Date()}).where(eq(issueRecoveryActions.id,actions[0].id)); + release();const result=await running; + const [state]=await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id,actions[0].id)); + expect(state).toMatchObject({status:"resolved",resolutionNote:operatorResolves?"concurrent_board_decision":"durable_path_restored:interaction"}); + expect(result.resolved).toBe(operatorResolves?0:1); + writeFileSync(`${evidence}/late-${operatorResolves}.json`,JSON.stringify({result,state,lateFactObserved:true,concurrentDecisionPreserved:operatorResolves},null,2),{flag:"wx"}); + }finally{release();await running;observer.mockRestore();await operator.$client.end({timeout:1});} + },15000); + +}); diff --git a/scripts/benchmark-active-recovery-projection.mjs b/scripts/benchmark-active-recovery-projection.mjs new file mode 100644 index 00000000000..505b2471ba8 --- /dev/null +++ b/scripts/benchmark-active-recovery-projection.mjs @@ -0,0 +1,47 @@ +#!/usr/bin/env node +import { execFileSync, spawnSync } from "node:child_process"; +import { existsSync, mkdirSync, mkdtempSync, readFileSync, unlinkSync, writeFileSync } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; + +const root = path.resolve(path.dirname(fileURLToPath(import.meta.url)), ".."); +const argument = name => process.argv.slice(2).find(value => value.startsWith(`${name}=`))?.slice(name.length + 1); +const baselineRef = argument("--baseline-ref"); +if (!baselineRef) throw new Error("Pass --baseline-ref=. Use an isolated checkout."); +const baseline = execFileSync("git", ["rev-parse", "--verify", baselineRef], { cwd: root, encoding: "utf8" }).trim(); +const evidenceParent = path.resolve(argument("--evidence-dir") ?? os.tmpdir()); +mkdirSync(evidenceParent, { recursive: true }); +const evidence = mkdtempSync(path.join(evidenceParent, "active-recovery-projection-")); +const recovery = path.join(root, "server/src/services/recovery"); +const fixture = path.join(root, "server/src/__tests__/active-recovery-projection.benchmark.test.ts"); +const targets = [path.join(recovery, ".bench-baseline-service.ts"), path.join(recovery, ".bench-baseline-disposition-repair.ts"), fixture]; +for (const target of targets) if (existsSync(target)) throw new Error(`Refusing to replace existing file: ${target}`); +const original = name => execFileSync("git", ["show", `${baseline}:server/src/services/recovery/${name}.ts`], { cwd: root, encoding: "utf8" }); +const sources = [ + original("service").replace('from "./disposition-repair.js"', 'from "./.bench-baseline-disposition-repair.js"'), + original("disposition-repair"), + readFileSync(path.join(root, "doc/benchmarks/fixtures/active-recovery-projection.fixture.ts"), "utf8"), +]; +const created = []; +try { + for (const [index, target] of targets.entries()) { + writeFileSync(target, sources[index], { flag: "wx" }); + created.push(target); + } + writeFileSync(path.join(evidence, "provenance.json"), JSON.stringify({ baseline, + treatment: execFileSync("git", ["rev-parse", "HEAD"], { cwd: root, encoding: "utf8" }).trim(), + dirty: execFileSync("git", ["status", "--short"], { cwd: root, encoding: "utf8" }).trim(), + node: process.version, platform: process.platform, architecture: process.arch, evidence, + }, null, 2)); + const filter = process.argv.includes("--proof-only") ? ["-t", "untimed count|records mixed current|reads late query"] : []; + const result = spawnSync(process.execPath, ["../node_modules/vitest/vitest.mjs", "run", + "src/__tests__/active-recovery-projection.benchmark.test.ts", ...filter], { + cwd: path.join(root, "server"), env: { ...process.env, RECOVERY_EVIDENCE_DIR: evidence }, stdio: "inherit", + }); + if (result.error) throw result.error; + process.exitCode = result.status ?? 1; +} finally { + for (const target of created.reverse()) unlinkSync(target); + process.stdout.write(`Evidence: ${evidence}\n`); +} diff --git a/server/src/__tests__/recovery-durable-wait-projection.test.ts b/server/src/__tests__/recovery-durable-wait-projection.test.ts new file mode 100644 index 00000000000..16958a89283 --- /dev/null +++ b/server/src/__tests__/recovery-durable-wait-projection.test.ts @@ -0,0 +1,162 @@ +import { randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { PgDialect } from "drizzle-orm/pg-core"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { + agents, agentWakeupRequests, approvals, companies, createDb, heartbeatRuns, + issueApprovals, issueRecoveryActions, issueRelations, issueThreadInteractions, issues, type Db, +} from "@paperclipai/db"; +import { collectDispositionRepairSourceState, readDispositionRepairDurableWaitingState } from "../services/recovery/disposition-repair.js"; +import { recoveryService } from "../services/recovery/service.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_RECOVERY_ACTION_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +const pendingExecution = { status: "pending", currentStageId: null, currentStageIndex: null, + currentStageType: null, currentParticipant: null, returnAssignee: null, completedStageIds: [], + lastDecisionId: null, lastDecisionOutcome: null }; + +(support.supported ? describe : describe.skip)("recovery durable wait projection", () => { + let temporary: Awaited> | undefined; + let db: Db; + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("recovery-durable-wait-"); + db = createDb(external ?? temporary!.connectionString); + db.$client.options.onnotice = () => {}; + }, 30_000); + afterEach(async () => { vi.restoreAllMocks(); await db.execute(sql`truncate companies cascade`); }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + + async function fixture() { + const companyId = randomUUID(), agentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Recovery", issuePrefix: `R${companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Owner", role: "engineer", status: "idle" }); + const [issue] = await db.insert(issues).values({ companyId, title: "Source", status: "blocked", assigneeAgentId: agentId }).returning(); + const [action] = await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issue.id, + kind: "deliberate_wait_without_target", ownerType: "agent", ownerAgentId: agentId, + cause: "deliberate_wait_without_target", fingerprint: "fixture", nextAction: "Decide", + wakePolicy: { type: "bounded_recovery_owner" } }).returning(); + return { companyId, agentId, issue, action }; + } + const enqueueWakeup = vi.fn(async () => { throw new Error("Unexpected recovery wake"); }); + + it("ignores foreign, hidden and settled evidence, then sees a newly pending wait", async () => { + const f = await fixture(), foreign = await fixture(); + const blockers = await db.insert(issues).values([ + { companyId: f.companyId, title: "Hidden", status: "todo", hiddenAt: new Date() }, + { companyId: f.companyId, title: "Done", status: "done" }, + ]).returning(); + await db.insert(issueRelations).values([ + ...blockers.map(blocker => ({ companyId: f.companyId, issueId: blocker.id, relatedIssueId: f.issue.id, type: "blocks" as const })), + { companyId: f.companyId, issueId: foreign.issue.id, relatedIssueId: f.issue.id, type: "blocks" }, + ]); + await db.insert(issueThreadInteractions).values([ + { companyId: foreign.companyId, issueId: f.issue.id, kind: "request_confirmation", status: "pending", payload: { version: 1, prompt: "Foreign" } }, + { companyId: f.companyId, issueId: f.issue.id, kind: "request_confirmation", status: "answered", payload: { version: 1, prompt: "Answered" } }, + ]); + const [foreignApproval] = await db.insert(approvals).values({ companyId: foreign.companyId, type: "hire_agent", status: "pending", payload: {} }).returning(); + const [approved] = await db.insert(approvals).values({ companyId: f.companyId, type: "hire_agent", status: "approved", payload: {} }).returning(); + await db.insert(issueApprovals).values([ + { companyId: f.companyId, issueId: f.issue.id, approvalId: foreignApproval.id }, + { companyId: f.companyId, issueId: f.issue.id, approvalId: approved.id }, + ]); + const read = () => readDispositionRepairDurableWaitingState(db, { issue: f.issue }); + expect(await read()).toEqual({ hasDurableWaitingPath: false, durablePathReason: null }); + const full = await collectDispositionRepairSourceState(db, { issue: f.issue }); + expect(full.hasDurableWaitingPath).toBe(false); + await db.update(approvals).set({ status: "revision_requested" }).where(eq(approvals.id, approved.id)); + expect(await read()).toEqual({ hasDurableWaitingPath: true, durablePathReason: "approval" }); + expect((await collectDispositionRepairSourceState(db, { issue: f.issue })).fingerprint).not.toBe(full.fingerprint); + }); + + it.each(["user_owner", "blocker", "monitor", "execution_stage", "interaction", "approval"] as const)( + "preserves %s precedence and resolves the original action without a wake", async reason => { + const f = await fixture(); + const order = ["user_owner", "blocker", "monitor", "execution_stage", "interaction", "approval"]; + const priority = order.indexOf(reason); + if (priority <= 0) await db.update(issues).set({ assigneeAgentId: null, assigneeUserId: "board" }).where(eq(issues.id, f.issue.id)); + if (priority <= 1) { + const [blocker] = await db.insert(issues).values({ companyId: f.companyId, title: "Blocker", status: "todo" }).returning(); + await db.insert(issueRelations).values({ companyId: f.companyId, issueId: blocker.id, relatedIssueId: f.issue.id, type: "blocks" }); + } + if (priority <= 2) await db.update(issues).set({ monitorNextCheckAt: new Date(Date.now() + 3_600_000) }).where(eq(issues.id, f.issue.id)); + if (priority <= 3) await db.update(issues).set({ executionState: pendingExecution }).where(eq(issues.id, f.issue.id)); + if (priority <= 4) await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issue.id, + kind: "request_confirmation", status: "pending", payload: { version: 1, prompt: "Decide" } }); + const [approval] = await db.insert(approvals).values({ companyId: f.companyId, type: "hire_agent", status: "pending", payload: {} }).returning(); + await db.insert(issueApprovals).values({ companyId: f.companyId, issueId: f.issue.id, approvalId: approval.id }); + const [current] = await db.select().from(issues).where(eq(issues.id, f.issue.id)); + const projected = await readDispositionRepairDurableWaitingState(db, { issue: current }); + expect(projected).toEqual({ hasDurableWaitingPath: true, durablePathReason: reason }); + expect(await collectDispositionRepairSourceState(db, { issue: current })).toMatchObject(projected); + const result = await recoveryService(db, { enqueueWakeup }).reconcileActiveRecoveryActions({ companyId: f.companyId }); + expect(result.resolved).toBe(1); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.action.id)); + expect(action).toMatchObject({ status: "resolved", ownerType: "agent", ownerAgentId: f.agentId, + resolutionNote: `durable_path_restored:${reason}` }); + }, + ); + + it("retains board-owned waits and same-action paths but accepts an independent path", async () => { + const f = await fixture(); + await db.update(issueRecoveryActions).set({ ownerType: "board", ownerAgentId: null }).where(eq(issueRecoveryActions.id, f.action.id)); + await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issue.id, + kind: "request_confirmation", status: "pending", payload: { version: 1, prompt: "Board" } }); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running", + contextSnapshot: { issueId: f.issue.id, recoveryActionId: f.action.id } }); + await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, source: "system", status: "queued", + payload: { issueId: f.issue.id, recoveryActionId: f.action.id } }); + const service = recoveryService(db, { enqueueWakeup }); + expect((await service.reconcileActiveRecoveryActions({ companyId: f.companyId })).resolved).toBe(0); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running", contextSnapshot: { taskId: f.issue.id } }); + expect((await service.reconcileActiveRecoveryActions({ companyId: f.companyId })).resolved).toBe(1); + }); + + it("keeps native reconciliation holds and resolves a board action through a healthy child", async () => { + const held = await fixture(), parent = await fixture(); + await db.update(issueRecoveryActions).set({ ownerType: "board", ownerAgentId: null, + cause: "native_workspace_finalization_owner_unverified" }).where(eq(issueRecoveryActions.id, held.action.id)); + await db.insert(heartbeatRuns).values({ companyId: held.companyId, agentId: held.agentId, status: "running", contextSnapshot: { issueId: held.issue.id } }); + const service = recoveryService(db, { enqueueWakeup }); + expect((await service.reconcileActiveRecoveryActions({ companyId: held.companyId })).resolved).toBe(0); + await db.update(issueRecoveryActions).set({ ownerType: "board", ownerAgentId: null }).where(eq(issueRecoveryActions.id, parent.action.id)); + const [child] = await db.insert(issues).values({ companyId: parent.companyId, title: "Child", status: "todo", + parentId: parent.issue.id, assigneeAgentId: parent.agentId }).returning(); + await db.insert(issueThreadInteractions).values({ companyId: parent.companyId, issueId: child.id, + kind: "request_confirmation", status: "pending", payload: { version: 1, prompt: "Child decision" } }); + expect((await service.reconcileActiveRecoveryActions({ companyId: parent.companyId })).resolved).toBe(1); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, parent.action.id)); + expect(action.resolutionNote).toBe("durable_path_restored:healthy_child"); + expect(await db.select().from(issueRelations).where(eq(issueRelations.relatedIssueId, parent.issue.id))) + .toMatchObject([{ issueId: child.id, type: "blocks" }]); + const [unchanged] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, held.action.id)); + expect(unchanged.status).toBe("active"); + }); + + it.each([false, true])("reads late facts and preserves a concurrent action decision (%s)", async operatorResolves => { + const f = await fixture(); + let started!: () => void, release!: () => void; + const ready = new Promise(resolve => { started = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + const execute = db.execute.bind(db), dialect = new PgDialect(); + const spy = vi.spyOn(db, "execute").mockImplementation(((query: Parameters[0]) => { + if (dialect.sqlToQuery(query as never).sql.includes('as "hasBlocker"')) { + started(); return gate.then(() => execute(query)); + } + return execute(query); + }) as Db["execute"]); + const running = recoveryService(db, { enqueueWakeup }).reconcileActiveRecoveryActions({ companyId: f.companyId }); + const operator = createDb(external ?? temporary!.connectionString, { maxConnections: 1 }); + try { + await vi.waitFor(async () => { await ready; }, { timeout: 3_000 }); + await operator.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issue.id, + kind: "request_confirmation", status: "pending", payload: { version: 1, prompt: "Late decision" } }); + if (operatorResolves) await operator.update(issueRecoveryActions).set({ status: "resolved", resolutionNote: "board_decision" }).where(eq(issueRecoveryActions.id, f.action.id)); + release(); const result = await running; + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.action.id)); + expect(action.resolutionNote).toBe(operatorResolves ? "board_decision" : "durable_path_restored:interaction"); + expect(result.resolved).toBe(operatorResolves ? 0 : 1); + } finally { release(); await running; spy.mockRestore(); await operator.$client.end({ timeout: 1 }); } + }); +}); diff --git a/server/src/services/recovery/disposition-repair.ts b/server/src/services/recovery/disposition-repair.ts index b3ac840187a..d2e737c5d2e 100644 --- a/server/src/services/recovery/disposition-repair.ts +++ b/server/src/services/recovery/disposition-repair.ts @@ -86,6 +86,58 @@ export function dispositionRepairDelayMs(attempt: number, fingerprint: string) { ); } +type DurableWaitingFacts = { + hasBlocker: boolean; + hasPendingInteraction: boolean; + hasPendingApproval: boolean; +}; + +function dispositionRepairDurablePathReason(issue: DispositionRepairIssue, facts: DurableWaitingFacts) { + return issue.assigneeUserId + ? "user_owner" + : facts.hasBlocker + ? "blocker" + : issue.monitorNextCheckAt && issue.monitorNextCheckAt.getTime() > Date.now() + ? "monitor" + : parseIssueExecutionState(issue.executionState)?.status === "pending" + ? "execution_stage" + : facts.hasPendingInteraction + ? "interaction" + : facts.hasPendingApproval + ? "approval" + : null; +} + +/** Current wait facts for recovery resolution, without collecting retry fingerprints. */ +export async function readDispositionRepairDurableWaitingState( + db: Db, + input: { issue: DispositionRepairIssue }, +): Promise> { + const issue = input.issue; + const [facts] = await db.execute(sql`select + exists (select 1 from ${issueRelations} + inner join ${issues} on ${issues.companyId} = ${issueRelations.companyId} + and ${issues.id} = ${issueRelations.issueId} + where ${issueRelations.companyId} = ${issue.companyId} + and ${issueRelations.relatedIssueId} = ${issue.id} + and ${issueRelations.type} = 'blocks' + and ${issues.status} not in ('done', 'cancelled') + and ${issues.hiddenAt} is null) as "hasBlocker", + exists (select 1 from ${issueThreadInteractions} + where ${issueThreadInteractions.companyId} = ${issue.companyId} + and ${issueThreadInteractions.issueId} = ${issue.id} + and ${issueThreadInteractions.status} = 'pending') as "hasPendingInteraction", + exists (select 1 from ${issueApprovals} + inner join ${approvals} on ${issueApprovals.approvalId} = ${approvals.id} + and ${issueApprovals.companyId} = ${approvals.companyId} + where ${issueApprovals.companyId} = ${issue.companyId} + and ${approvals.companyId} = ${issue.companyId} + and ${issueApprovals.issueId} = ${issue.id} + and ${approvals.status} in ('pending', 'revision_requested')) as "hasPendingApproval"`); + const durablePathReason = dispositionRepairDurablePathReason(issue, facts!); + return { hasDurableWaitingPath: durablePathReason !== null, durablePathReason }; +} + export async function collectDispositionRepairSourceState( db: Db, input: { @@ -199,24 +251,11 @@ export async function collectDispositionRepairSourceState( ), ]); - const pendingExecutionState = parseIssueExecutionState(issue.executionState); - const pendingInteraction = interactions.some((row) => row.status === "pending"); - const pendingApproval = linkedApprovals.some((row) => - row.status === "pending" || row.status === "revision_requested", - ); - const durablePathReason = issue.assigneeUserId - ? "user_owner" - : blockers.length > 0 - ? "blocker" - : issue.monitorNextCheckAt && issue.monitorNextCheckAt.getTime() > Date.now() - ? "monitor" - : pendingExecutionState?.status === "pending" - ? "execution_stage" - : pendingInteraction - ? "interaction" - : pendingApproval - ? "approval" - : null; + const durablePathReason = dispositionRepairDurablePathReason(issue, { + hasBlocker: blockers.length > 0, + hasPendingInteraction: interactions.some((row) => row.status === "pending"), + hasPendingApproval: linkedApprovals.some((row) => row.status === "pending" || row.status === "revision_requested"), + }); const durableState = { source: { diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 0b7bfc6db7a..e020141baa4 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -168,6 +168,7 @@ import { withRecoveryContext } from "./status-only-context.js"; import { isAutomaticRecoverySuppressedByPauseHold } from "./pause-hold-guard.js"; import { collectDispositionRepairSourceState, + readDispositionRepairDurableWaitingState, dispositionRepairIssueFields, dispositionRepairDelayMs, DISPOSITION_REPAIR_MAX_ATTEMPTS, @@ -3575,7 +3576,7 @@ export function recoveryService( const [sourceState, healthyChildren, hasNewSourcePath] = await Promise.all([ - collectDispositionRepairSourceState(db, { issue }), + readDispositionRepairDurableWaitingState(db, { issue }), healthyOpenChildIssues(issue), sourceHasNewPathOutsideRecoveryAction(action), ]);