mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-06 23:20:53 +01:00
perf(recovery): project current wait facts during action reconciliation (#18793)
## Thinking Path > - Paperclip manages work and execution for AI agents. > - Recovery actions keep unresolved work visible and preserve its owner. > - Action reconciliation checks whether a durable wait or another live path has appeared. > - Its current-wait check collected a complete retry fingerprint through seven queries. > - This caller used only the durable-wait flag and reason. > - This pull request reads those facts in one fresh statement and shares the existing reason policy. > - The local service uses less CPU. Fresh mutation guards and full retry fingerprints stay in place. ## Linked Issues or Issue Description **What existing behavior does this improve?** Active recovery action reconciliation. **Subsystem affected** server/ — orchestration and recovery. **Current behavior** Each admitted source action collects blockers, children, interactions, approvals, work products, runs and wakes. It sorts and hashes these records. This caller uses only the durable-wait flag and reason. **Proposed behavior** Read three current existence facts in one company-bound statement. Use the same reason policy as the complete fingerprint collector. Keep child checks, outside-action checks and mutation services. **Reason and benefit** The actual local PostgreSQL service uses less CPU across 8, 32 and 100-action synthetic cohorts. This removes unused work on mostly empty results. The live workload mix and fleet benefit remain unknown. **Breaking changes** No API or schema changes. Query facts now share one statement snapshot. Scalar issue fields still come from the source cohort read. **Additional context** No duplicate projection change was found in the public PR search. The roadmap has no overlapping core feature. This is a narrow performance change. ## What Changed - Add a current durable-wait reader and share reason selection with the full fingerprint collector. - Use the reader only in active action reconciliation. Keep the full collector for retry and child policy. - Add PostgreSQL regressions for reason priority, scope, board waits, native holds, healthy children and concurrent decisions. - Add a manual reproducer and local paired results. Each invocation uses a unique evidence directory and exclusive output files. ## Verification - Focused PostgreSQL regressions: 11 passed. - Separate actual-service proof: 7 passed. Four CPU tests were excluded. Results, records, semantic relation edges and notifications match the pinned baseline. Late query facts and a concurrent action decision are covered. - Retained primary local paired CPU medians: 8 actions 7.091 → 2.886 ms; 32 actions 21.922 → 8.374 ms; 100 actions 68.384 → 32.718 ms. All nine CPU and wall pairs improve at these sizes. The data file retains all 72 primary samples and ranges. - Warm application counts: 81 → 33 queries at 8 actions; 321 → 129 at 32; 1001 → 401 at 100. Rows rise from N to 2N because each EXISTS statement returns one row. - See `doc/benchmarks/2026-11-03-active-recovery-projection.md` and its JSON data for commands, provenance and limits. - `pnpm -r typecheck` and `pnpm build` passed on head `f2f8c4a755ac67679d78c6005f420a21a101c1ad`. - Shipped manual helper smoke with `--proof-only`: 7 passed, 4 timing tests skipped. All count-proof sample arrays are empty. Unique output names and cleanup of the three temporary source files were verified. - Exact-head CI completed with 52 successful checks, three intentional skips and a successful Snyk status. Greptile completed successfully at 5/5 with no inline findings. The `review` workflow passed commitperclip dependency and PR quality gates. No separate Paperclip Review check or verdict was published. The local `pnpm test:run` attempt passed all six chat shards (1,834 tests), then reported 62 Muse failures and one gateway failure in the general-server batch. The initial cases each reached the unchanged 15-second timeout. The Muse case was followed by fast failures. After the full batch breadth and complete CI coverage were established, the owned known-failed run was interrupted once with SIGINT. Remaining general-server, serialized and workspace scopes were not completed by that local invocation. Its interrupted tool exit code is not a pass. Each initial timeout case then passed in one fresh isolated control. A single fresh full Muse file passed all 256 tests. Source review found no changed projection path in either initial case. The original timeout causes remain unresolved; later passes and green CI do not erase them. The first reserved CPU run lost its raw files after a harness path error caused an unintended repeat to overwrite them. Its summary and log remain historical evidence. The separately retained repeat is the primary inspectable evidence above. It had no renewed quiet reservation. The runs are not pooled. The corrected untimed proof and public helper prevent output replacement. No third CPU run was made. ## Risks - One statement observes its query facts together. This does not reproduce every race history of seven concurrent statements. - Existing action update predicates, issue mutation locks, board ownership, native reconciliation holds and company guards remain in place. - The evidence is a local source service benchmark on macOS and real PostgreSQL. It is not deployed CPU or request latency acceptance. The census was startup-adjacent. Conditional scale suggests only a modest fleet return. - The original local batch timeout causes remain unexplained. No hosted, Sentry or OTLP improvement is claimed. No merge or deployment is requested. ## Model Used OpenAI `gpt-6.1-sol`, with reasoning, tool use and code execution. The host did not expose the context window size. ## 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 (focused and isolated controls; the broad local failure/interruption is described above) - [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 Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
f7ad1dc7ca
commit
f79264d253
7 files changed
+1099
-19
No files matched your search
@@ -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"
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -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.
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
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<string,unknown>={};
|
||||
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<keyof typeof services>){
|
||||
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<void>(r=>{started=r;});const gate=new Promise<void>(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<typeof setTimeout>|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);
|
||||
|
||||
});
|
||||
@@ -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=<pre-change commit>. 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`);
|
||||
}
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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<void>(resolve => { started = resolve; });
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
const execute = db.execute.bind(db), dialect = new PgDialect();
|
||||
const spy = vi.spyOn(db, "execute").mockImplementation(((query: Parameters<Db["execute"]>[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 }); }
|
||||
});
|
||||
});
|
||||
@@ -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<Pick<DispositionRepairSourceState, "hasDurableWaitingPath" | "durablePathReason">> {
|
||||
const issue = input.issue;
|
||||
const [facts] = await db.execute<DurableWaitingFacts>(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: {
|
||||
|
||||
@@ -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),
|
||||
]);
|
||||
|
||||
Reference in new issue
Block a user