mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-28 12:04:41 +01:00
fix(server): schedule join request retention from durable deadlines (#17508)
## Thinking Path > - Paperclip manages agent companies and their access requests. > - Pending join requests appear in the attention feed and can age into automatic archival. > - The maintenance tick reads those requests even when no archive is due. > - Existing request rows already store the activity timestamp needed for retention. > - Committed changes can wake a worker that registers the next actual deadline. > - This PR gives join-request retention its own bounded worker and removes its recurring feed scan. > - Idle and resolved history stop producing retention traffic without changing access approval. ## Linked Issues or Issue Description **What happened?** Every retention maintenance pass reads pending join requests through the attention feed. Empty, resolved, kept, and future-due states still participate in recurring discovery. **Expected behavior** Committed request and retention changes wake the owner. It waits until the next actual archive deadline and stops querying when no eligible request remains. Existing approval and visibility behavior is preserved. **Steps to reproduce** Start an idle company with no pending join requests, or with requests younger than the 90-day archive threshold. Observe the maintenance feed repeatedly select join_requests. No state change is needed to cause those reads. This PR targets master after #17505 merged. Approval retention is #17497. Other attention source kinds remain separate migration duties. No duplicate open join-retention PR was found. ## What Changed - Add a join-request retention owner that preserves pending status, active-company visibility, matching invite ownership, and keep/archive controls. Invite expiry or revocation retains the existing feed policy. - Reuse one `createDecisionRetentionSweep` for approval and join paging, atomic retention and finite scan cycles that revisit failures despite continuing arrivals. Eligibility and source-row locks stay with each owner. - Register real saved deadlines, rounded upward in SQL to millisecond precision, with the existing coordinator. Preserve 100-row pages and thirty-second notification coalescing without idle polling. - Materialize unseen join state inside the existing authorized keep/archive/revive transaction, preserving archive choices and source locking. - Notify before request inserts and updates in the existing acceptance and Slack transactions. Use a small SQL-only helper for replay and approval/rejection updates. Company status, budget pause/resume, deletion, and retention controls notify this owner too. - Preserve the approved-only API-key consumption path without a new wrapper; it cannot produce pending retention work. - Exclude join requests from maintenance feed collection while retaining normal UI reads. - Add an idempotent concurrent partial pending index and prune old schema snapshots through the existing generator. - Document producer coverage, boundaries, failure evidence, and remaining work. ## Verification Test revision: `df1815fbba`. Normal integration of landed workspace precision `d23dd19750` has an identical complete source tree. This remains only the join-retention duty and its shared approval-retention mechanics, targeting master. - **55 worker cases pass**, including two new PostgreSQL regressions that failed before repair: continuing overdue arrivals no longer postpone a recovered failure indefinitely, while an entire failed page still allows later healthy rows to progress. - The nine complete suites passed 341 cases but failed an attention cleanup hook after its 16,400-run case. After vacuum/analyze of only that disposable fixture, the full 31-case attention suite passes unchanged. Together with the original eight passing suites (311 cases), all 342 distinct cases pass across complete-suite executions. Recovery is established; the original cleanup slowdown is not claimed explained. Server typecheck and production build pass. - Earlier CI at `df1d4d0` failed the unchanged decision-expiry restart deadline assertion (a thirty-second retry appeared before the saved expiry). All twenty unchanged expiry cases pass locally, but the CI timing cause remains unproven. No retry, timeout relaxation or assertion change is used to dismiss it; fresh final-head CI remains required. - **Five regressions failed before repair:** three unseen join controls returned 404, and both owners advertised a microsecond deadline one millisecond early. All pass. Two old real-producer expectations failed after the correct rounding change; they now derive exact expected milliseconds independently from database integer microseconds instead of truncating a JavaScript Date. - The initial timing-only concurrency test still passed a missing-lock mutation. It now observes the actual PostgreSQL wait before releasing the writer. Production passes; removing the lock fails with zero waiters instead of one. Source is restored before commit. The archive-first committed-order case confirms that later activity stays archived until explicit revival; no broader race guarantee is claimed. - Generated concurrent/idempotent `0393_open_grandmaster.sql` adds only the pending join index. All 391 parent journal entries and historical SQL files remain unchanged. Five generated snapshots remain (0389–0393). Migration safety/numbering, schema drift and actual apply/replay on six isolated PostgreSQL fixtures pass. - Existing lifecycle cases retain real HTTP invite acceptance/approval/rejection, Slack identity linking, rollback/outer commit, source visibility, failure rotation, drain/resume and shutdown coverage. Earlier stack setup and CI failures remain in the owning audit and stack ledger; they are historical evidence, not this head's acceptance. Commands: server Vitest for `join-request-retention-work`, `approval-retention-work`, `approvals-service`, `attention-service`, `decision-retention-service`, `budgets-service`, `tool-gateway-service`, `server-startup-feedback-export`, `scoped-deadline-work`; server typecheck/build; DB migration checks, snapshot drift and actual apply/replay. Fresh exact-head CI, security and Greptile/Ponytail 5/5 remain required. No full local repository-suite, deployed suspension or cost claim is made. ## Risks - A future supported producer must signal its transaction. Independent SQL writers and another active server process are outside the process-local notification contract. - A writer that wins the source lock first is rechecked before archival. If archival commits first, later activity preserves the archived state until explicit revival. Existing invite replay expires before the retention cutoff and Slack replay deduplicates. Future producers needing renewed visibility must request revival. Cross-table visibility retains the existing contract. - The worker registers only under the existing maintenance enablement switch. The internal deadline does not add Cloud alarm coverage or relax conservative sleep blockers. - New eligibility can wait within the thirty-second aggregate notification window; known deadlines preempt it. - Archival preserves pending approval status. Other attention sources and aggregate maintenance remain separate work. ## Model Used OpenAI Codex, GPT-6 family, with reasoning and code execution. The exact serving model identifier and context window were not exposed. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
d23dd19750
commit
8406047b4e
23 files changed
+963
-72
No files matched your search
@@ -0,0 +1,183 @@
|
||||
# Join request retention while idle
|
||||
|
||||
Pending join requests already have durable activity timestamps. Retention now
|
||||
uses their next 90-day archive deadline instead of including them in the main
|
||||
maintenance tick's attention-feed scan. The regular attention UI still includes
|
||||
join requests. This changes automatic retention discovery, not access approval.
|
||||
|
||||
## Owner and eligibility
|
||||
|
||||
`join-request-retention.ts` owns candidate eligibility, its locked recheck and
|
||||
deadline. `decision-retention-sweep.ts` shares bounded paging, per-row retention
|
||||
transactions and failure handling with the approval-retention owner. The shared
|
||||
sweep consumes normalized candidates; each owner retains its table/query policy.
|
||||
A candidate is a pending_approval request belonging to an active company, with
|
||||
an invite whose company matches the request. Kept and already archived items do
|
||||
not schedule work. Invite expiration or revocation does not remove the existing
|
||||
attention item and therefore does not disable its retention. No provider policy
|
||||
is interpreted by the worker or shared scheduler.
|
||||
|
||||
The next deadline is the earliest eligible updatedAt plus 90 days, rounded up
|
||||
in SQL to the scheduler's millisecond resolution. It must not wake a fraction of
|
||||
a millisecond early and then wait thirty seconds for recovery. The partial
|
||||
pending index orders updated_at and id. Resolved request history is excluded by
|
||||
the index predicate. Each pass attempts at most 100 due rows. The private cursor
|
||||
preserves database timestamp precision so a failed page cannot starve later work.
|
||||
Errors keep the shared 30-second retry; successful full pages continue until
|
||||
work is exhausted. No eligible rows means no scheduled SQL or network scan.
|
||||
|
||||
Each candidate is re-read under its request row lock before retention changes.
|
||||
A writer that wins that lock first is rechecked before archival. If archival
|
||||
commits first, a later activity write leaves the item archived until explicit
|
||||
revival; updating activity does not clear existing retention choices. The tests
|
||||
cover the writer-first race and the archive-first committed outcome. Supported
|
||||
invite replay expires before the 90-day cutoff and Slack access replay deduplicates;
|
||||
a future producer that needs to reappear after archival must request revival.
|
||||
The shared retention owner still controls keep, archive/revive, version checks,
|
||||
activity logging, and notification creation. Archiving does not approve or reject a join.
|
||||
Join requests have no origin agent, so their archival creates no agent delivery.
|
||||
Company and invite joins preserve existing visibility policy; this does not add
|
||||
cross-table serializability beyond the existing retention contract.
|
||||
|
||||
## Producer audit
|
||||
|
||||
| Producer | Notification boundary |
|
||||
| --- | --- |
|
||||
| Invite acceptance creates a pending request | Existing acceptance transaction, before the request insert |
|
||||
| Supported gateway invite replay changes activity | Small writeJoinRequestChange transaction around the existing SQL update |
|
||||
| Automatic human approval after acceptance | Same helper around the status update |
|
||||
| Explicit approval and rejection | Same helper around the status update |
|
||||
| Slack self-service access request | Existing platform transaction, before the request insert |
|
||||
| Company pause, resume, archive, or deletion | Company-level retention notification inside the existing transaction |
|
||||
| Budget-driven company pause or resume | Same company notification in the accounting transaction |
|
||||
| Keep/unkeep, manual archive/revive, accepted archive proposal | Existing typed retention notification now routes join_request to its worker |
|
||||
|
||||
Provider calls, membership grants, and unrelated route behavior stay outside the
|
||||
small SQL mutation helper. Nested callers retain the existing outer transaction
|
||||
settlement and uncertain-outcome protocol. The access router itself requires a
|
||||
root database connection; it is not wrapped in a caller transaction.
|
||||
|
||||
API-key consumption only permits an already-approved request. It cannot create
|
||||
pending retention work and does not acquire a new wrapper or signal. Invite
|
||||
acceptance that reuses a human request does not update that request. Revocation,
|
||||
expiry, and acceptedAt changes do not change the feed predicate. Invite/company
|
||||
identity is immutable in supported routes; company deletion is the invite delete
|
||||
owner. A future producer that changes these facts must join the same contract.
|
||||
Process-local signals do not cover independent SQL writers or another instance.
|
||||
|
||||
## Lifecycle and limits
|
||||
|
||||
Startup reconstructs pending deadlines. The worker is registered only when the
|
||||
existing heartbeat maintenance switch is enabled. Writes during drain retain a
|
||||
wake without passive SQL. Failed work retains its retry across drain. Shutdown
|
||||
joins an admitted archive and stops before subsequent candidates; restart picks
|
||||
up the durable remainder. There is no new queue table or permanent DB listener.
|
||||
The in-process deadline is not a Cloud alarm. Existing unsupported-future-work
|
||||
sleep blockers remain in place.
|
||||
|
||||
## Earlier verification
|
||||
|
||||
Seventeen new real-PostgreSQL cases pass. They cover actual invite acceptance,
|
||||
rejection, human approval, gateway replay, and Slack access requests; empty and
|
||||
resolved-history quiet; saved deadlines; startup reconstruction; keep and manual
|
||||
archival controls; company pause/resume; unchanged expired/revoked invite policy;
|
||||
normal versus maintenance feeds; nested commit/rollback; first writes and failed
|
||||
work during drain; concurrent source updates; 101-row microsecond paging with a
|
||||
full failed page; and shutdown/restart.
|
||||
|
||||
Another 139 consumer cases pass: 71 invitation/startup cases, 21 approval worker
|
||||
cases, and 47 shared retention/attention/budget cases. This is 156 selected tests,
|
||||
not a full local suite claim. The first consumer command used a nonexistent
|
||||
retention test filename and ran only the approval suite; the corrected selection
|
||||
ran the actual retention suite against its dedicated external database.
|
||||
|
||||
Initial test failures exposed fixture assumptions: the access router cannot run
|
||||
on a transaction handle; unresolved recovery cannot be expected to become idle;
|
||||
Slack links require actual endpoint/principal ownership rows; and shutdown tests
|
||||
must establish admission and deterministic candidate order. Those fixtures were
|
||||
corrected without weakening assertions or production checks. The first replay
|
||||
fixture lacked required gateway credentials. The final fixture supplies test-only
|
||||
credentials. A reused transaction spy retained admission calls during the quiet
|
||||
assertion; restoring it before the quiet measurement fixed the measurement.
|
||||
|
||||
Migration 0387 applies, the full migration set replays, and the concurrent index
|
||||
statement replays idempotently. Migration safety and numbering checks pass; the
|
||||
five newest schema snapshots are retained. The actual next-deadline query with
|
||||
40,000 resolved requests and one pending request uses the pending index and
|
||||
executes in 0.017 ms locally. The probe rolls back its data. This is query-plan
|
||||
evidence, not production latency or measured savings. Workspace typechecking and the production build pass. Fresh PR CI/review remain
|
||||
required before readiness.
|
||||
|
||||
|
||||
## Current review repair and verification
|
||||
|
||||
The duplicate approval/join archive loop is now one `createDecisionRetentionSweep`
|
||||
implementation. Each owner supplies its candidate ordering boundary, page, locked current-state
|
||||
lookup and next deadline. This keeps eligibility, company/invite joins and
|
||||
source-row locks with their owner while sharing cursor/error/retention behavior.
|
||||
The existing 100-row budget, PostgreSQL text cursor and failure rotation remain.
|
||||
Both workers coalesce notification-driven global reads under a thirty-second
|
||||
aggregate budget; saved deadlines preempt it and no idle scan is scheduled.
|
||||
|
||||
Manual keep/archive/revive can materialize an unseen join request in the existing
|
||||
retention write transaction, as approvals already can. The company-scoped source
|
||||
is locked, its real activity is read, and existing versioned state is synchronized.
|
||||
Authorization, archive choices and other source kinds retain their contracts.
|
||||
No route bookkeeping, queue table, permanent listener or Cloud exemption is added.
|
||||
|
||||
Five real regressions failed before repair: three unseen-request controls returned
|
||||
404 and both owners rounded a saved microsecond deadline down. All pass. The first
|
||||
two-owner run passed 50 cases but failed two old producer expectations that used
|
||||
a truncated JavaScript Date. They now compute the expected scheduler millisecond
|
||||
independently from integer database microseconds; exact assertions remain.
|
||||
|
||||
The first missing-lock mutation exposed a weak test: its timing-only writer gate
|
||||
released before proving the sweep had reached the lock, so it still passed without
|
||||
`FOR UPDATE`. The strengthened test observes the actual PostgreSQL lock wait before
|
||||
releasing the writer. It passes with the production lock and fails with zero waiters
|
||||
when that lock is removed. The production source was restored in `finally` before
|
||||
commit. A separate archive-first committed-order case preserves the existing
|
||||
archived-stays-archived behavior and verifies explicit revival.
|
||||
|
||||
All 340 cases in nine complete suites pass at `405c7dcdef`, including all 31
|
||||
approval and 22 join worker cases. Server typecheck and production build pass.
|
||||
Normal integration of parent `0b0e844b77` changes only four independently merged
|
||||
Runner E2E accounting files and the ledger; runtime and acceptance sources remain
|
||||
byte-identical. Existing results retain their tested revision.
|
||||
|
||||
Generated concurrent/idempotent migration `0393_open_grandmaster.sql` adds only the
|
||||
pending-request index. All 391 parent journal entries and historical SQL files are
|
||||
unchanged; five generated snapshots remain (0389–0393). Migration safety/numbering,
|
||||
schema drift and actual apply/replay on six isolated PostgreSQL fixtures pass.
|
||||
Earlier setup, timing, mutation and CI failures remain recorded here and in the
|
||||
stack ledger. Fresh final-head CI, security and both 5/5 reviews remain required.
|
||||
|
||||
|
||||
## Retry progress under continuing arrivals
|
||||
|
||||
Each scan cycle saves the final due `(updated_at, id)` tuple, at full database
|
||||
precision. Its pages stop at that boundary. Later due rows cannot keep extending
|
||||
the cycle and postpone failed rows forever. The cursor still advances past an
|
||||
individual failure, so one bad request—or an entire failed page—does not block
|
||||
later healthy rows. Once that finite cycle ends, the next cycle revisits failures.
|
||||
There is no new durable queue or per-row retry registry. Restart reconstructs
|
||||
eligibility from existing rows.
|
||||
|
||||
Two real PostgreSQL regressions failed before this repair: continued newer overdue
|
||||
arrivals left a recovered approval and join request behind the cursor. Both now
|
||||
archive in the next cycle. All 55 worker cases pass, including the existing
|
||||
100-failure-page progress and microsecond-cursor cases. The nine-suite run passed
|
||||
341 cases but failed the attention suite's cleanup hook after its 16,400-run case:
|
||||
`delete from heartbeat_runs` exceeded thirty seconds. After vacuum/analyze of only
|
||||
the disposable attention fixture, all 31 attention cases pass without code or
|
||||
timeout changes. The other eight suites passed 311 cases in the original run;
|
||||
342 distinct cases pass across those complete-suite executions. This demonstrates
|
||||
recovery, not a proven explanation of the cleanup slowdown. Server typecheck and
|
||||
production build pass at `df1815fbba`; the subsequent master integration has an
|
||||
identical complete source tree.
|
||||
|
||||
Earlier CI at `df1d4d0` also failed the unchanged decision-expiry restart test:
|
||||
its earliest wake was a thirty-second retry rather than the saved expiry. The
|
||||
complete unchanged expiry suite passes all twenty cases locally. Its CI timing
|
||||
cause remains unproven; it is not labeled infrastructure or silently discarded.
|
||||
Fresh final-head CI and reviews must pass before merging.
|
||||
@@ -17,12 +17,8 @@ Do not mark an item done from a timer removal or a green check alone.
|
||||
| 03 | Run and task recovery | Orphan runs and active leases; shutdown workspace settlement; scheduled retries; queued runs; stranded assignments; resolved dependencies; task watchdogs; silent-run monitoring; stale locks | Pending |
|
||||
| 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Approval retention [#17497](https://github.com/paperclipai/paperclip/pull/17497) and archive delivery [#17475](https://github.com/paperclipai/paperclip/pull/17475) merged; other source retention pending |
|
||||
| 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Continuations [#17478](https://github.com/paperclipai/paperclip/pull/17478) merged; expiry [#17479](https://github.com/paperclipai/paperclip/pull/17479) merged; effects [#17482](https://github.com/paperclipai/paperclip/pull/17482) merged; targets [#17486](https://github.com/paperclipai/paperclip/pull/17486) merged |
|
||||
|
||||
|
||||
|
||||
| 06 | Tool connection health | Configuration/use activation and defined proactive health deadlines | Pending |
|
||||
| 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Slices 02–07 merged: [secret proposals](https://github.com/paperclipai/paperclip/pull/17446), [credentials](https://github.com/paperclipai/paperclip/pull/17450), [review expiry](https://github.com/paperclipai/paperclip/pull/17453), [approved recovery](https://github.com/paperclipai/paperclip/pull/17454), [interrupted claims](https://github.com/paperclipai/paperclip/pull/17458), [projection](https://github.com/paperclipai/paperclip/pull/17468) |
|
||||
|
||||
| 08 | GitHub connection events and continuity | Event discovery/replay and credential continuity separately | Pending |
|
||||
| 09 | Merged-PR confirmations | Candidate writes, webhook activation where supported, outstanding-candidate fallback | Pending |
|
||||
| 10 | External-object refresh | Configuration changes, refresh deadlines, source events | Pending |
|
||||
@@ -41,7 +37,8 @@ The first PR targets master. Each following PR targets its predecessor branch so
|
||||
its diff shows only its own reason. Parent changes must be propagated before
|
||||
final verification. Record the exact checked head and base for every PR. All PRs
|
||||
must finish with passing checks, fresh 5/5 Greptile review, and no unresolved
|
||||
findings. This request does not authorize merging the stack.
|
||||
findings. The later user authorization permits merges after current-head Greptile
|
||||
5/5, a passing Ponytail review when present, and passing CI, in dependency order.
|
||||
|
||||
Reuse the delivery coordinator, transaction notifications, and shared scheduler.
|
||||
Existing durable rows own work; no table per loop. Keep domain/provider behavior
|
||||
@@ -1083,3 +1080,55 @@ accounting files. Its entire runtime and acceptance sources match the tested/
|
||||
reviewed `a34dc50de3` tree (136 cases, server types/build); migration metadata is
|
||||
unchanged. No local checks were repeated on unchanged source. Fresh final-head
|
||||
CI/reviews are required against master; all remaining checklist duties persist.
|
||||
|
||||
## Join request retention implementation
|
||||
|
||||
Terminal workspace pagination is [#17505](https://github.com/paperclipai/paperclip/pull/17505),
|
||||
head `28a197cdfd`, stacked on #17497. It has fresh 5/5 review without unresolved
|
||||
findings. The full CI matrix is running. Workspace typecheck/build and all 87
|
||||
selected service/worker tests passed locally.
|
||||
|
||||
Join retention is implemented on `codex/sleep-15-join-retention`, based on #17505.
|
||||
The worker preserves pending request/invite/company eligibility, manual controls,
|
||||
transaction ownership, microsecond paging, and bounded recovery. All supported
|
||||
writers are audited; approved-only API-key consumption is deliberately exempt.
|
||||
See [join retention](../idle-join-request-retention.md). Seventeen new real-DB
|
||||
lifecycle cases and 139 consumers pass (156 total). Initial fixture and command
|
||||
failures are recorded in that document. Index generation, apply, replay, safety,
|
||||
numbering, and an actual-query plan with 40,000 resolved requests pass. Workspace
|
||||
typecheck/build and remote publication/CI/review remain in progress. No slice is
|
||||
claimed ready based only on local tests, and no merge is authorized.
|
||||
|
||||
After join retention, migrate the remaining attention source owners separately:
|
||||
decisions, interactions, recovery actions, reviews, and other source kinds. Keep
|
||||
the full original server inventory and later aggregate-tick retirement in scope.
|
||||
|
||||
|
||||
Slice 15 review repair: approval and join retention share the bounded archive
|
||||
sweep; owners retain eligibility and source-row locks. SQL deadlines round up
|
||||
from database precision. Unseen join controls materialize state atomically in
|
||||
the existing retention boundary. Five real regressions failed then passed. The
|
||||
initial timing-only concurrency test failed to detect a missing lock; it now
|
||||
observes the actual database wait, passes normally and rejects lock removal.
|
||||
Archive-first committed activity remains archived until explicit revival.
|
||||
|
||||
All 340 cases across nine complete suites (31 approval/22 join worker cases),
|
||||
server types/build, migration checks and schema drift pass at `405c7dcdef`.
|
||||
All six isolated database fixtures pass actual migration apply/replay. Index 0393
|
||||
is generated after 391 unchanged parent entries; five snapshots remain. Parent
|
||||
`0b0e844b77` adds only four Runner E2E accounting files and ledger changes; runtime
|
||||
and acceptance sources are unchanged. Fresh publication checks/reviews required.
|
||||
|
||||
|
||||
Slice 15 final master integration: #17505 merged as `d23dd19750` after both
|
||||
final-head 5/5 reviews and all 58 accepted checks. Join retention now targets
|
||||
master and integrates that landed parent without changing the tested source tree.
|
||||
The shared retention scan saves a finite upper tuple so continuing arrivals cannot
|
||||
starve failed rows; both real arrival regressions failed then passed. All 55 worker
|
||||
cases pass. Eight consumer suites passed 311 cases; the ninth attention suite had
|
||||
a cleanup-hook timeout and then passed all 31 after disposable-fixture vacuuming.
|
||||
The unchanged expiry suite passes twenty local cases after a separate CI deadline
|
||||
assertion failure; neither passing rerun is presented as a proven root cause.
|
||||
Server types/build pass. See the join-retention document for the full audit.
|
||||
Fresh exact-head reviews and CI remain required. All original duties remain open
|
||||
until their own accepted merge, including native-harness backup expiration.
|
||||
@@ -511,6 +511,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([
|
||||
"0390_salty_killraven.sql",
|
||||
"0391_third_roulette.sql",
|
||||
"0392_small_frightful_four.sql",
|
||||
"0393_open_grandmaster.sql",
|
||||
|
||||
|
||||
"0384_funny_adam_warlock.sql",
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS "join_requests_pending_retention_idx" ON "join_requests" USING btree ("updated_at","id") WHERE "join_requests"."status" = 'pending_approval';
|
||||
+112
-2
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"id": "4f21a636-735f-4dce-9ab8-0583f3bc7b79",
|
||||
"prevId": "de27fdf8-430a-443a-9502-725f92c48892",
|
||||
"id": "e1ddc6f7-6335-4b3d-aeec-a5df27793874",
|
||||
"prevId": "6974dd25-b8e9-4143-b4b7-ee5a5c3ff54a",
|
||||
"version": "7",
|
||||
"dialect": "postgresql",
|
||||
"tables": {
|
||||
@@ -5721,6 +5721,28 @@
|
||||
}
|
||||
},
|
||||
"indexes": {
|
||||
"approvals_pending_retention_idx": {
|
||||
"name": "approvals_pending_retention_idx",
|
||||
"columns": [
|
||||
{
|
||||
"expression": "updated_at",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
},
|
||||
{
|
||||
"expression": "id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
}
|
||||
],
|
||||
"isUnique": false,
|
||||
"where": "\"approvals\".\"status\" = 'pending'",
|
||||
"concurrently": false,
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"approvals_company_status_type_idx": {
|
||||
"name": "approvals_company_status_type_idx",
|
||||
"columns": [
|
||||
@@ -24058,6 +24080,50 @@
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"decisions_running_recovery_idx": {
|
||||
"name": "decisions_running_recovery_idx",
|
||||
"columns": [
|
||||
{
|
||||
"expression": "updated_at",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
},
|
||||
{
|
||||
"expression": "id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
}
|
||||
],
|
||||
"isUnique": false,
|
||||
"where": "\"decisions\".\"status\" = 'decided' AND \"decisions\".\"execution_status\" = 'running'",
|
||||
"concurrently": false,
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"decisions_open_expiry_idx": {
|
||||
"name": "decisions_open_expiry_idx",
|
||||
"columns": [
|
||||
{
|
||||
"expression": "expires_at",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
},
|
||||
{
|
||||
"expression": "id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
}
|
||||
],
|
||||
"isUnique": false,
|
||||
"where": "\"decisions\".\"status\" = 'open'",
|
||||
"concurrently": false,
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"decisions_pending_continuation_idx": {
|
||||
"name": "decisions_pending_continuation_idx",
|
||||
"columns": [
|
||||
@@ -32063,6 +32129,28 @@
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"grok_dispatch_prepared_idx": {
|
||||
"name": "grok_dispatch_prepared_idx",
|
||||
"columns": [
|
||||
{
|
||||
"expression": "created_at",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
},
|
||||
{
|
||||
"expression": "id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
}
|
||||
],
|
||||
"isUnique": false,
|
||||
"where": "\"grok_bot_dispatches\".\"status\" = 'prepared'",
|
||||
"concurrently": false,
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"grok_dispatch_company_status_idx": {
|
||||
"name": "grok_dispatch_company_status_idx",
|
||||
"columns": [
|
||||
@@ -41857,6 +41945,28 @@
|
||||
}
|
||||
},
|
||||
"indexes": {
|
||||
"join_requests_pending_retention_idx": {
|
||||
"name": "join_requests_pending_retention_idx",
|
||||
"columns": [
|
||||
{
|
||||
"expression": "updated_at",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
},
|
||||
{
|
||||
"expression": "id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
}
|
||||
],
|
||||
"isUnique": false,
|
||||
"where": "\"join_requests\".\"status\" = 'pending_approval'",
|
||||
"concurrently": false,
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"join_requests_invite_unique_idx": {
|
||||
"name": "join_requests_invite_unique_idx",
|
||||
"columns": [
|
||||
+7
@@ -2738,6 +2738,13 @@
|
||||
"when": 1793143372644,
|
||||
"tag": "0392_small_frightful_four",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 393,
|
||||
"version": "7",
|
||||
"when": 1793145214502,
|
||||
"tag": "0393_open_grandmaster",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -31,6 +31,8 @@ export const joinRequests = pgTable(
|
||||
updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(),
|
||||
},
|
||||
(table) => ({
|
||||
pendingRetentionIdx: index("join_requests_pending_retention_idx").on(table.updatedAt, table.id)
|
||||
.where(sql`${table.status} = 'pending_approval'`),
|
||||
inviteUniqueIdx: uniqueIndex("join_requests_invite_unique_idx").on(table.inviteId),
|
||||
companyStatusTypeCreatedIdx: index("join_requests_company_status_type_created_idx").on(
|
||||
table.companyId,
|
||||
|
||||
@@ -145,6 +145,17 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
|
||||
expect((await db.select().from(approvals).where(eq(approvals.id, other.id)))[0].status).toBe("pending");
|
||||
});
|
||||
|
||||
it("rounds a submillisecond approval deadline up to the next scheduler millisecond", async () => {
|
||||
const row = await create(), floor = Date.now() + 120_000;
|
||||
const activity = new Date(floor - delay).toISOString();
|
||||
await db.execute(sql`update approvals set updated_at = ${activity}::timestamptz + interval '456 microseconds' where id = ${row.id}`);
|
||||
const service = approvalRetentionService(db);
|
||||
expect(await service.next()).toBe(floor + 1);
|
||||
expect(await service.sweep(new Date(floor))).toEqual({ archived: 0 });
|
||||
expect(await state(row.id)).toBeNull();
|
||||
expect(await service.sweep(new Date(floor + 1))).toEqual({ archived: 1 });
|
||||
});
|
||||
|
||||
it("recovers old unseen approvals on startup and leaves completed history quiet", async () => {
|
||||
const due = await create(old()), completed = await create(old());
|
||||
await approvalRecords(db).resolveApproval(completed.id, "approved", "board-user", null);
|
||||
@@ -334,6 +345,26 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
|
||||
expect(await service.sweep()).toEqual({ archived: 1 }); expect(await service.next()).toBeNull();
|
||||
});
|
||||
|
||||
it("retries a failed approval while newer overdue pages keep arriving", async () => {
|
||||
const at = old();
|
||||
const [failed] = await db.insert(approvals).values({ companyId, type: "request_board_approval",
|
||||
payload: { retentionTestFailure: true }, createdAt: at, updatedAt: at }).returning();
|
||||
const append = async (offset: number) => db.insert(approvals).values(Array.from({ length: 100 }, (_, i) => ({
|
||||
companyId, type: "request_board_approval", payload: {}, createdAt: at,
|
||||
updatedAt: new Date(at.getTime() + offset + i),
|
||||
})));
|
||||
await append(1); await failArchives();
|
||||
const service = approvalRetentionService(db);
|
||||
await expect(service.sweep()).rejects.toThrow("Failed query");
|
||||
await append(1000); await service.sweep();
|
||||
await append(2000);
|
||||
await db.execute(sql`drop trigger approval_retention_fail on decision_retention`);
|
||||
await service.sweep();
|
||||
expect((await state(failed.id))?.archivedAt).toBeInstanceOf(Date);
|
||||
while (await service.next() !== null) await service.sweep();
|
||||
expect(await service.next()).toBeNull();
|
||||
});
|
||||
|
||||
it("reconstructs startup obligations while paused and performs no passive SQL", async () => {
|
||||
const row = await create(old()); draining = true;
|
||||
await start();
|
||||
|
||||
@@ -0,0 +1,377 @@
|
||||
import { createSlackIdentityLinks } from "../services/chat-channels/slack/identity-links.js";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import express from "express";
|
||||
import request from "supertest";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { activityLog, agents, authUsers, chatEndpoints, chatExternalPrincipals, chatIdentityLinks, companies, companyMemberships, companySecretProviderConfigs, companySecrets, companySecretVersions, createDb, decisionArchiveNotificationOutbox,
|
||||
decisionRetention, invites, joinRequests, principalPermissionGrants, toolApplications, toolConnections, type Db } from "@paperclipai/db";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import { joinRequestRetentionService, registerJoinRequestRetentionWork } from "../services/join-request-retention.js";
|
||||
import { decisionRetentionService, DEFAULT_DECISION_ARCHIVE_DAYS } from "../services/decision-retention.js";
|
||||
import { attentionService } from "../services/attention.js";
|
||||
import { companyService } from "../services/companies.js";
|
||||
import { writeJoinRequestChange } from "../services/retention-work-notifications.js";
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
|
||||
const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_JOIN_RETENTION_TEST_DATABASE_URL");
|
||||
const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
|
||||
(support.supported ? describe : describe.skip)("join request retention worker", () => {
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: ReturnType<typeof createDb>, companyId: string, userId: string;
|
||||
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
|
||||
let draining: boolean;
|
||||
let accessRoutes: typeof import("../routes/access.js").accessRoutes;
|
||||
const errors: unknown[] = [];
|
||||
beforeAll(async () => {
|
||||
({ accessRoutes } = await import("../routes/access.js"));
|
||||
if (!externalUrl) temporary = await startEmbeddedPostgresTestDatabase("join-retention-");
|
||||
db = createDb(externalUrl ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
|
||||
}, 30_000);
|
||||
beforeEach(async () => {
|
||||
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "0123456789abcdef0123456789abcdef");
|
||||
companyId = randomUUID(); userId = randomUUID(); draining = false; errors.length = 0;
|
||||
await db.insert(companies).values({ id: companyId, name: "Join retention", issuePrefix: `J${companyId.slice(0, 6)}` });
|
||||
await db.insert(authUsers).values({ id: userId, name: "Owner", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() });
|
||||
await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "owner", status: "active" });
|
||||
await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true });
|
||||
});
|
||||
afterEach(async () => {
|
||||
await coordinator?.stop(); coordinator = undefined;
|
||||
vi.useRealTimers(); vi.restoreAllMocks(); vi.unstubAllEnvs();
|
||||
await db.execute(sql`drop trigger if exists join_retention_fail on decision_retention`);
|
||||
await db.execute(sql`drop function if exists join_retention_fail()`);
|
||||
await db.delete(decisionArchiveNotificationOutbox); await db.delete(decisionRetention); await db.delete(activityLog);
|
||||
await db.delete(chatIdentityLinks); await db.delete(chatExternalPrincipals); await db.delete(chatEndpoints);
|
||||
await db.delete(toolConnections); await db.delete(toolApplications); await db.delete(agents);
|
||||
await db.delete(joinRequests); await db.delete(invites); await db.delete(principalPermissionGrants);
|
||||
await db.delete(companySecretVersions); await db.delete(companySecrets); await db.delete(companySecretProviderConfigs);
|
||||
await db.delete(companyMemberships); await db.delete(authUsers); await db.delete(companies);
|
||||
expect(errors).toEqual([]);
|
||||
});
|
||||
afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
|
||||
const old = () => new Date(Date.now() - delay - 60_000);
|
||||
const state = (id: string) => decisionRetentionService(db).getState(companyId, "join_request", id);
|
||||
async function exactDeadline(id: string) {
|
||||
const [row] = await db.select({ micros: sql<string>`(extract(epoch from ${joinRequests.updatedAt}) * 1000000)::bigint::text` })
|
||||
.from(joinRequests).where(eq(joinRequests.id, id));
|
||||
return Number((BigInt(row!.micros) + 999n) / 1000n) + delay;
|
||||
}
|
||||
const authActor = () => ({ type: "board" as const, source: "local_implicit" as const, userId, companyIds: [companyId], isInstanceAdmin: true,
|
||||
memberships: [{ companyId, membershipRole: "owner", status: "active" }] });
|
||||
const mutation = (id: string) => ({ companyId, sourceKind: "join_request" as const, sourceId: id, authActor: authActor(),
|
||||
actor: { actorType: "user" as const, actorId: userId, agentId: null, userId, runId: null, agentApiKeyId: null, responsibleUserId: userId } });
|
||||
async function invite(database: Db = db) {
|
||||
const token = randomUUID();
|
||||
const [row] = await database.insert(invites).values({ companyId, tokenHash: createHash("sha256").update(token).digest("hex"),
|
||||
allowedJoinTypes: "both", expiresAt: new Date(Date.now() + 86_400_000), invitedByUserId: userId }).returning();
|
||||
return { ...row, token };
|
||||
}
|
||||
async function create(at = new Date(), database: Db = db) {
|
||||
const invitation = await invite(database);
|
||||
const [row] = await writeJoinRequestChange(database, tx => tx.insert(joinRequests).values({ companyId, inviteId: invitation.id,
|
||||
requestType: "agent", requestIp: "127.0.0.1", agentName: "Pending agent", createdAt: at, updatedAt: at }).returning());
|
||||
return row;
|
||||
}
|
||||
function app(database: Db = db) {
|
||||
const server = express(); server.use(express.json());
|
||||
server.use((req, _res, next) => { req.actor = authActor(); next(); });
|
||||
server.use("/api", accessRoutes(database, { deploymentMode: "authenticated", deploymentExposure: "private",
|
||||
bindHost: "127.0.0.1", allowedHostnames: [] }));
|
||||
server.use((error: { status?: number; message?: string }, _req: express.Request, res: express.Response, _next: express.NextFunction) => {
|
||||
res.status(error.status ?? 500).json({ error: error.message });
|
||||
});
|
||||
return server;
|
||||
}
|
||||
async function accept(token: string, database: Db = db) {
|
||||
const response = await request(app(database)).post(`/api/invites/${token}/accept`)
|
||||
.send({ requestType: "agent", agentName: "New agent", adapterType: "claude_local" });
|
||||
expect(response.status, JSON.stringify(response.body)).toBe(202);
|
||||
const [row] = await database.select().from(joinRequests).where(eq(joinRequests.companyId, companyId));
|
||||
return row;
|
||||
}
|
||||
async function start(service = joinRequestRetentionService(db)) {
|
||||
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true, onError: error => errors.push(error) });
|
||||
await registerJoinRequestRetentionWork(coordinator, service, () => !draining).ready;
|
||||
return service;
|
||||
}
|
||||
async function settle() {
|
||||
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 5000 });
|
||||
}
|
||||
async function quiet() {
|
||||
await settle(); expect(coordinator!.nextWakeAt()).toBeNull();
|
||||
const reads = vi.spyOn(db, "select"), executes = vi.spyOn(db, "execute"), writes = vi.spyOn(db, "transaction");
|
||||
await vi.advanceTimersByTimeAsync(300_000);
|
||||
expect(reads).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
|
||||
reads.mockRestore(); executes.mockRestore(); writes.mockRestore();
|
||||
}
|
||||
async function archived(id: string) {
|
||||
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect((await state(id))?.archivedAt).toBeInstanceOf(Date); }, { timeout: 5000 });
|
||||
}
|
||||
|
||||
it("wakes from real invite acceptance, waits for the saved deadline, and disarms on rejection", async () => {
|
||||
await start(); await quiet();
|
||||
const invitation = await invite(); const row = await accept(invitation.token);
|
||||
await settle(); expect(coordinator!.nextWakeAt()).toBe(await exactDeadline(row.id));
|
||||
const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(300_000);
|
||||
expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
|
||||
const response = await request(app()).post(`/api/companies/${companyId}/join-requests/${row.id}/reject`).send({});
|
||||
expect(response.status, JSON.stringify(response.body)).toBe(200); await quiet();
|
||||
expect(await state(row.id)).toBeNull();
|
||||
});
|
||||
|
||||
it("archives unseen pending requests at the actual deadline without changing approval status", async () => {
|
||||
await start(); await quiet(); const deadline = Date.now() + 600_000;
|
||||
const row = await create(new Date(deadline - delay)); await settle();
|
||||
expect(coordinator!.nextWakeAt()).toBe(deadline); expect(await state(row.id)).toBeNull();
|
||||
const window = deadline - Date.now() - 30_000; expect(window).toBeGreaterThan(0);
|
||||
const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(window);
|
||||
expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
|
||||
await vi.advanceTimersByTimeAsync(60_000); await archived(row.id); await quiet();
|
||||
expect((await db.select().from(joinRequests))[0].status).toBe("pending_approval");
|
||||
expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("rounds a submillisecond join deadline up to the next scheduler millisecond", async () => {
|
||||
const row = await create(), floor = Date.now() + 120_000;
|
||||
const activity = new Date(floor - delay).toISOString();
|
||||
await db.execute(sql`update join_requests set updated_at = ${activity}::timestamptz + interval '456 microseconds' where id = ${row.id}`);
|
||||
const service = joinRequestRetentionService(db);
|
||||
expect(await service.next()).toBe(floor + 1);
|
||||
expect(await service.sweep(new Date(floor))).toEqual({ archived: 0 });
|
||||
expect(await state(row.id)).toBeNull();
|
||||
expect(await service.sweep(new Date(floor + 1))).toEqual({ archived: 1 });
|
||||
});
|
||||
|
||||
it.each(["keep", "archive", "revive"] as const)("applies %s to an unseen young request without reading the feed", async action => {
|
||||
const row = await create(); expect(await state(row.id)).toBeNull();
|
||||
const retention = decisionRetentionService(db);
|
||||
const result = action === "keep" ? await retention.setKeep({ ...mutation(row.id), keep: true })
|
||||
: action === "archive" ? await retention.archive(mutation(row.id)) : await retention.revive(mutation(row.id));
|
||||
expect(result).toMatchObject({ sourceKind: "join_request", sourceId: row.id, sourceActivityAt: row.updatedAt, keep: action === "keep" });
|
||||
expect(result.archivedAt).toEqual(action === "archive" ? expect.any(Date) : null);
|
||||
expect(await state(row.id)).toMatchObject({ id: result.id });
|
||||
});
|
||||
|
||||
it("reconstructs pending requests at startup and ignores resolved history", async () => {
|
||||
const due = await create(old()), resolved = await create(old());
|
||||
await db.update(joinRequests).set({ status: "approved" }).where(eq(joinRequests.id, resolved.id));
|
||||
await start(); await archived(due.id); await quiet(); expect(await state(resolved.id)).toBeNull();
|
||||
});
|
||||
|
||||
it("keeps revoked and expired invite requests visible under the existing retention policy", async () => {
|
||||
const row = await create(old()); await db.update(invites).set({ expiresAt: old(), revokedAt: old() });
|
||||
const feed = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true });
|
||||
expect(feed.items.some(item => item.subject.id === row.id)).toBe(true);
|
||||
await start(); await archived(row.id); await quiet();
|
||||
});
|
||||
|
||||
it("waits for company activation and disarms when the company pauses", async () => {
|
||||
const row = await create(old()); await companyService(db).update(companyId, { status: "paused" });
|
||||
await start(); await quiet(); expect(await state(row.id)).toBeNull();
|
||||
await companyService(db).update(companyId, { status: "active" }); await archived(row.id); await quiet();
|
||||
await create(); await settle(); expect(coordinator!.nextWakeAt()).not.toBeNull();
|
||||
await companyService(db).update(companyId, { status: "paused" }); await quiet();
|
||||
});
|
||||
|
||||
it("honors keep and schedules an old request again when unkept", async () => {
|
||||
const row = await create(old()); await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true });
|
||||
await decisionRetentionService(db).setKeep({ ...mutation(row.id), keep: true });
|
||||
await start(); await quiet();
|
||||
await decisionRetentionService(db).setKeep({ ...mutation(row.id), keep: false }); await archived(row.id); await quiet();
|
||||
});
|
||||
|
||||
it("honors manual archival and reactivates on revive", async () => {
|
||||
const row = await create(old()); await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true });
|
||||
await decisionRetentionService(db).archive(mutation(row.id)); await start(); await quiet();
|
||||
await decisionRetentionService(db).revive(mutation(row.id)); await archived(row.id); await quiet();
|
||||
});
|
||||
|
||||
it("preserves normal attention reads but omits join requests from maintenance", async () => {
|
||||
const row = await create(old());
|
||||
const excluded = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true, excludeSources: ["join_request"] });
|
||||
expect(excluded.items.some(item => item.sourceKind === "join_request")).toBe(false); expect(await state(row.id)).toBeNull();
|
||||
const feed = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true });
|
||||
expect(feed.items.some(item => item.subject.id === row.id)).toBe(true);
|
||||
});
|
||||
|
||||
it("defers nested request writes until outer commit and releases rolled-back writes", async () => {
|
||||
await start(); await quiet();
|
||||
await expect(db.transaction(async tx => {
|
||||
await create(new Date(), tx as unknown as Db);
|
||||
await vi.advanceTimersByTimeAsync(120_000); expect(coordinator!.nextWakeAt()).toBeNull(); throw new Error("rollback");
|
||||
})).rejects.toThrow("rollback");
|
||||
await quiet(); expect(await db.select().from(joinRequests)).toHaveLength(0);
|
||||
let row: typeof joinRequests.$inferSelect;
|
||||
await db.transaction(async tx => {
|
||||
row = await create(new Date(), tx as unknown as Db);
|
||||
await vi.advanceTimersByTimeAsync(120_000); expect(coordinator!.nextWakeAt()).toBeNull();
|
||||
});
|
||||
await settle(); expect(coordinator!.nextWakeAt()).toBe(row!.updatedAt.getTime() + delay);
|
||||
});
|
||||
|
||||
it("retains a first write during drain without passive SQL", async () => {
|
||||
await start(); await quiet(); draining = true; const row = await create(old());
|
||||
const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
|
||||
draining = false; await archived(row.id); await quiet();
|
||||
});
|
||||
|
||||
it("retains a failed archival through drain until the database recovers", async () => {
|
||||
const row = await create(old());
|
||||
await db.execute(sql`create function join_retention_fail() returns trigger language plpgsql as $$ begin
|
||||
raise exception 'retention unavailable'; end $$`);
|
||||
await db.execute(sql`create trigger join_retention_fail before insert or update on decision_retention for each row execute function join_retention_fail()`);
|
||||
const service = joinRequestRetentionService(db); const sweep = vi.spyOn(service, "sweep");
|
||||
await start(service); await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0)); draining = true;
|
||||
await Promise.allSettled(sweep.mock.results.map(result => result.value));
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
|
||||
await db.execute(sql`drop trigger join_retention_fail on decision_retention`); draining = false;
|
||||
await archived(row.id); await quiet();
|
||||
for (const error of errors) expect(error).toMatchObject({ cause: { message: "retention unavailable" } }); errors.length = 0;
|
||||
});
|
||||
|
||||
it("wakes from the Slack access-request producer and deduplicates repeated requests", async () => {
|
||||
await start(); await quiet(); const token = randomUUID(), endpointId = randomUUID();
|
||||
const [agent] = await db.insert(agents).values({ companyId, name: "Chat agent", role: "engineer", status: "active", adapterType: "codex_local" }).returning();
|
||||
const [application] = await db.insert(toolApplications).values({ companyId, name: "Slack", type: "chat" }).returning();
|
||||
const [connection] = await db.insert(toolConnections).values({ companyId, applicationId: application.id, name: "Slack", uid: randomUUID(), transport: "chat_sdk", connectionPurpose: "channel" }).returning();
|
||||
await db.insert(chatEndpoints).values({ id: endpointId, companyId, connectionId: connection.id, provider: "slack", publicId: randomUUID(), assignedAgentId: agent.id, status: "active" });
|
||||
const [principal] = await db.insert(chatExternalPrincipals).values({ companyId, provider: "slack", providerAccountId: "team", externalId: randomUUID() }).returning();
|
||||
await db.insert(chatIdentityLinks).values({ companyId, endpointId, principalId: principal.id,
|
||||
confirmationTokenHash: createHash("sha256").update(token).digest("hex"), expiresAt: new Date(Date.now() + 600_000) });
|
||||
const links = createSlackIdentityLinks(db, {
|
||||
previewIdentityLink: async () => ({ selfService: true, canConfirm: false, companyId, endpointId }),
|
||||
stageProviderEffect: vi.fn(async () => null),
|
||||
});
|
||||
expect(await links.requestIdentityAccess(token, userId, "127.0.0.1")).toEqual({ status: "pending_approval" });
|
||||
await settle(); const [row] = await db.select().from(joinRequests);
|
||||
expect(coordinator!.nextWakeAt()).toBe(await exactDeadline(row.id));
|
||||
await links.requestIdentityAccess(token, userId, "127.0.0.1"); await settle();
|
||||
expect(await db.select().from(joinRequests)).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("disarms after the real approval route resolves a human request", async () => {
|
||||
const requester = randomUUID();
|
||||
await db.insert(authUsers).values({ id: requester, name: "Joiner", email: `${requester}@example.test`, createdAt: new Date(), updatedAt: new Date() });
|
||||
const row = await create();
|
||||
await db.update(joinRequests).set({ requestType: "human", requestingUserId: requester }).where(eq(joinRequests.id, row.id));
|
||||
await start(); await settle(); expect(coordinator!.nextWakeAt()).not.toBeNull();
|
||||
const response = await request(app()).post(`/api/companies/${companyId}/join-requests/${row.id}/approve`).send({});
|
||||
expect(response.status, JSON.stringify(response.body)).toBe(200); await quiet();
|
||||
expect((await db.select().from(joinRequests))[0].status).toBe("approved");
|
||||
});
|
||||
|
||||
it("rechecks activity after waiting for a concurrent request writer", async () => {
|
||||
const row = await create(old()); let release!: () => void, locked!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
const admitted = new Promise<void>(resolve => { locked = resolve; });
|
||||
const write = db.transaction(async tx => {
|
||||
await tx.execute(sql`select id from join_requests where id = ${row.id} for update`); locked(); await gate;
|
||||
await writeJoinRequestChange(tx as unknown as Db, owner => owner.update(joinRequests).set({ updatedAt: new Date() }).where(eq(joinRequests.id, row.id)));
|
||||
});
|
||||
await admitted; const sweep = joinRequestRetentionService(db).sweep();
|
||||
try {
|
||||
await vi.waitFor(async () => {
|
||||
const [row] = await db.execute(sql`select count(*)::int as waiting from pg_stat_activity
|
||||
where datname = current_database() and pid <> pg_backend_pid() and wait_event_type = 'Lock'
|
||||
and query like 'select%' and query like '%from "join_requests"%'`);
|
||||
expect(row.waiting).toBe(1);
|
||||
});
|
||||
} finally { release(); await write; await sweep; }
|
||||
expect(await state(row.id)).toBeNull();
|
||||
});
|
||||
|
||||
it("keeps an archived request archived when activity commits after the sweep until explicitly revived", async () => {
|
||||
const row = await create(old()), service = joinRequestRetentionService(db);
|
||||
expect(await service.sweep()).toEqual({ archived: 1 });
|
||||
const archivedAt = (await state(row.id))!.archivedAt;
|
||||
await writeJoinRequestChange(db, tx => tx.update(joinRequests).set({ updatedAt: new Date() }).where(eq(joinRequests.id, row.id)));
|
||||
expect(await state(row.id)).toMatchObject({ archivedAt });
|
||||
expect(await service.next()).toBeNull();
|
||||
expect((await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true })).items
|
||||
.some(item => item.subject.id === row.id)).toBe(false);
|
||||
await decisionRetentionService(db).revive(mutation(row.id));
|
||||
expect((await state(row.id))!.archivedAt).toBeNull();
|
||||
expect(await service.next()).toBe(await exactDeadline(row.id));
|
||||
expect(await service.sweep()).toEqual({ archived: 0 });
|
||||
});
|
||||
|
||||
it("pages past failed requests with database timestamp precision and retries them later", async () => {
|
||||
const ids: string[] = [];
|
||||
for (let i = 0; i < 101; i++) ids.push((await create(old())).id);
|
||||
ids.sort(); const last = ids.at(-1)!;
|
||||
await db.execute(sql`update join_requests set updated_at = date_trunc('milliseconds', now() - interval '91 days') + interval '456 microseconds'`);
|
||||
await db.execute(sql`create function join_retention_fail() returns trigger language plpgsql as $$ begin
|
||||
if new.archived_at is not null and new.source_id <> TG_ARGV[0] then raise exception 'retention unavailable'; end if;
|
||||
return new; end $$`);
|
||||
await db.execute(sql`create trigger join_retention_fail before update on decision_retention for each row execute function join_retention_fail(${sql.raw("'" + last + "'")})`);
|
||||
const service = joinRequestRetentionService(db);
|
||||
await expect(service.sweep()).rejects.toThrow("Join request retention failed");
|
||||
expect(await service.sweep()).toEqual({ archived: 1 }); expect((await state(last))?.archivedAt).toBeInstanceOf(Date);
|
||||
await db.execute(sql`drop trigger join_retention_fail on decision_retention`);
|
||||
expect(await service.sweep()).toEqual({ archived: 100 }); expect(await service.sweep()).toEqual({ archived: 0 });
|
||||
expect(await service.next()).toBeNull();
|
||||
});
|
||||
|
||||
it("retries a failed request while newer overdue pages keep arriving", async () => {
|
||||
const at = old(); const failed = await create(at);
|
||||
const append = async (offset: number) => {
|
||||
for (let i = 0; i < 100; i++) await create(new Date(at.getTime() + offset + i));
|
||||
};
|
||||
await append(1);
|
||||
await db.execute(sql`create function join_retention_fail() returns trigger language plpgsql as $$ begin
|
||||
if new.archived_at is not null and new.source_id = TG_ARGV[0] then raise exception 'retention unavailable'; end if;
|
||||
return new; end $$`);
|
||||
await db.execute(sql`create trigger join_retention_fail before update on decision_retention for each row execute function join_retention_fail(${sql.raw("'" + failed.id + "'")})`);
|
||||
const service = joinRequestRetentionService(db);
|
||||
await expect(service.sweep()).rejects.toThrow("Failed query");
|
||||
await append(1000); await service.sweep();
|
||||
await append(2000);
|
||||
await db.execute(sql`drop trigger join_retention_fail on decision_retention`);
|
||||
await service.sweep();
|
||||
expect((await state(failed.id))?.archivedAt).toBeInstanceOf(Date);
|
||||
while (await service.next() !== null) await service.sweep();
|
||||
expect(await service.next()).toBeNull();
|
||||
});
|
||||
|
||||
it("joins an admitted archive on shutdown and leaves later candidates for restart", async () => {
|
||||
const first = await create(new Date(old().getTime() - 1000)), second = await create(old());
|
||||
let release!: () => void, locked!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
const acquired = new Promise<void>(resolve => { locked = resolve; });
|
||||
const transaction = db.transaction.bind(db);
|
||||
const transactionSpy = vi.spyOn(db, "transaction").mockImplementationOnce((write, options) => transaction(async tx => {
|
||||
locked(); await gate; return write(tx);
|
||||
}, options));
|
||||
const starting = start(); await acquired;
|
||||
let stopped = false; const stopping = coordinator!.stop().then(() => { stopped = true; });
|
||||
try { await vi.advanceTimersByTimeAsync(500); expect(stopped).toBe(false); } finally { release(); }
|
||||
await starting; await stopping; coordinator = undefined; transactionSpy.mockRestore();
|
||||
expect((await state(first.id))?.archivedAt).toBeInstanceOf(Date); expect(await state(second.id)).toBeNull();
|
||||
await start(); await archived(second.id); await quiet();
|
||||
});
|
||||
|
||||
|
||||
it("moves the saved deadline after a supported gateway invite replay", async () => {
|
||||
const invitation = await invite(); await start(); await quiet();
|
||||
const body = { requestType: "agent", agentName: "Gateway", adapterType: "openclaw_gateway",
|
||||
agentDefaultsPayload: { url: "ws://127.0.0.1:18789", disableDeviceAuth: true, headers: { "x-openclaw-token": "fixture-gateway-token-1234567890" } } };
|
||||
const first = await request(app()).post(`/api/invites/${invitation.token}/accept`).send(body);
|
||||
expect(first.status, JSON.stringify(first.body)).toBe(202); await settle();
|
||||
const initial = coordinator!.nextWakeAt();
|
||||
await vi.advanceTimersByTimeAsync(60_000);
|
||||
const replay = await request(app()).post(`/api/invites/${invitation.token}/accept`).send({ ...body, agentName: "Renamed gateway" });
|
||||
expect(replay.status, JSON.stringify(replay.body)).toBe(202); await settle();
|
||||
const rows = await db.select().from(joinRequests); expect(rows).toHaveLength(1);
|
||||
expect(coordinator!.nextWakeAt()).toBe(rows[0].updatedAt.getTime() + delay);
|
||||
expect(coordinator!.nextWakeAt()).toBeGreaterThan(initial!);
|
||||
});
|
||||
|
||||
});
|
||||
@@ -1,3 +1,4 @@
|
||||
import { joinRequestRetentionService } from "../services/join-request-retention.js";
|
||||
import { approvalRetentionService } from "../services/approval-retention.js";
|
||||
import { createSecretProposalsService } from "../services/secret-proposals.js";
|
||||
import { logger } from "../middleware/logger.js";
|
||||
@@ -331,6 +332,11 @@ vi.mock("../realtime/live-events-ws.js", () => ({
|
||||
setupLiveEventsWebSocketServer: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("../services/join-request-retention.js", async importOriginal => ({
|
||||
...await importOriginal<typeof import("../services/join-request-retention.js")>(),
|
||||
joinRequestRetentionService: vi.fn(() => ({ sweep: vi.fn(async () => ({ archived: 0 })), next: vi.fn(async () => null) })),
|
||||
}));
|
||||
|
||||
vi.mock("../services/approval-retention.js", async importOriginal => ({
|
||||
...await importOriginal<typeof import("../services/approval-retention.js")>(),
|
||||
approvalRetentionService: vi.fn(() => ({ sweep: vi.fn(async () => ({ archived: 0 })), next: vi.fn(async () => null) })),
|
||||
@@ -548,6 +554,18 @@ describe("startServer feedback export wiring", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it.each([false, true])("registers join request retention only when maintenance is enabled (%s)", async enabled => {
|
||||
loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
|
||||
await startServer(); await settleStartupHousekeeping();
|
||||
const registration = deliveryWorkRegisterMock.mock.calls.find(([queue]) => queue === DELIVERY_QUEUES.joinRequestRetention);
|
||||
expect(Boolean(registration)).toBe(enabled);
|
||||
if (enabled) {
|
||||
const service = vi.mocked(joinRequestRetentionService).mock.results.at(-1)!.value;
|
||||
await registration![1].run(new AbortController().signal);
|
||||
expect(service.sweep).toHaveBeenCalled(); expect(service.next).toHaveBeenCalled();
|
||||
}
|
||||
});
|
||||
|
||||
it.each([false, true])("registers archive delivery separately from retention sweeps (scheduler %s)", async enabled => {
|
||||
loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled, heartbeatSchedulerIntervalMs: 30000 }));
|
||||
let tick: (() => void) | undefined;
|
||||
|
||||
+4
-1
@@ -1,3 +1,4 @@
|
||||
import { joinRequestRetentionService, registerJoinRequestRetentionWork } from "./services/join-request-retention.js";
|
||||
import { approvalRetentionService, registerApprovalRetentionWork } from "./services/approval-retention.js";
|
||||
import { registerDecisionTargetWork } from "./services/decision-target-work.js";
|
||||
import { registerDecisionEffectRecoveryWork } from "./services/decision-effect-recovery-work.js";
|
||||
@@ -1812,7 +1813,7 @@ async function startServerWithDatabaseTeardown(
|
||||
// per company per tick.
|
||||
const page = await attentionService(db as any).list(company.id, {
|
||||
includeDismissed: true,
|
||||
excludeSources: ["approval"],
|
||||
excludeSources: ["approval", "join_request"],
|
||||
all: true,
|
||||
allowUnscopedAll: true,
|
||||
});
|
||||
@@ -1862,6 +1863,8 @@ async function startServerWithDatabaseTeardown(
|
||||
() => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
|
||||
await step("approval_retention", () => registerApprovalRetentionWork(deliveryWork, approvalRetentionService(db as any),
|
||||
() => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
|
||||
await step("join_request_retention", () => registerJoinRequestRetentionWork(deliveryWork, joinRequestRetentionService(db as any),
|
||||
() => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
|
||||
await step("retention_sweep", runRetentionSweep);
|
||||
};
|
||||
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyRetentionWork, writeJoinRequestChange } from "../services/retention-work-notifications.js";
|
||||
import { createAgentLifecycle } from "../services/agent-lifecycle.js";
|
||||
import {
|
||||
createHash,
|
||||
@@ -3136,7 +3137,7 @@ export function accessRoutes(
|
||||
const approvedAt = new Date();
|
||||
const approvedByUserId =
|
||||
input.invite.invitedByUserId ?? (isLocalImplicit(input.req) ? "local-board" : null);
|
||||
const approved = await db
|
||||
const approved = await writeJoinRequestChange(db, tx => tx
|
||||
.update(joinRequests)
|
||||
.set({
|
||||
status: "approved",
|
||||
@@ -3146,7 +3147,7 @@ export function accessRoutes(
|
||||
})
|
||||
.where(eq(joinRequests.id, input.joinRequest.id))
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null);
|
||||
.then((rows) => rows[0] ?? null));
|
||||
|
||||
await logActivity(db, {
|
||||
companyId: input.companyId,
|
||||
@@ -3882,6 +3883,7 @@ export function accessRoutes(
|
||||
)
|
||||
);
|
||||
|
||||
await notifyRetentionWork(tx, "join_request");
|
||||
const row = await tx
|
||||
.insert(joinRequests)
|
||||
.values({
|
||||
@@ -3912,7 +3914,7 @@ export function accessRoutes(
|
||||
.then((rows) => rows[0]);
|
||||
return row;
|
||||
})
|
||||
: await db
|
||||
: await writeJoinRequestChange(db, tx => tx
|
||||
.update(joinRequests)
|
||||
.set({
|
||||
requestIp: requestIp(req),
|
||||
@@ -3935,7 +3937,7 @@ export function accessRoutes(
|
||||
})
|
||||
.where(eq(joinRequests.id, replayJoinRequestId as string))
|
||||
.returning()
|
||||
.then((rows) => rows[0]);
|
||||
.then((rows) => rows[0]));
|
||||
|
||||
if (!created) {
|
||||
throw conflict("Join request not found");
|
||||
@@ -4279,7 +4281,7 @@ export function accessRoutes(
|
||||
);
|
||||
}
|
||||
|
||||
const approved = await db
|
||||
const approved = await writeJoinRequestChange(db, tx => tx
|
||||
.update(joinRequests)
|
||||
.set({
|
||||
status: "approved",
|
||||
@@ -4291,7 +4293,7 @@ export function accessRoutes(
|
||||
})
|
||||
.where(eq(joinRequests.id, requestId))
|
||||
.returning()
|
||||
.then((rows) => rows[0]);
|
||||
.then((rows) => rows[0]));
|
||||
|
||||
await logActivity(db, {
|
||||
companyId,
|
||||
@@ -4338,7 +4340,7 @@ export function accessRoutes(
|
||||
if (existing.status !== "pending_approval")
|
||||
throw conflict("Join request is not pending");
|
||||
|
||||
const rejected = await db
|
||||
const rejected = await writeJoinRequestChange(db, tx => tx
|
||||
.update(joinRequests)
|
||||
.set({
|
||||
status: "rejected",
|
||||
@@ -4349,7 +4351,7 @@ export function accessRoutes(
|
||||
})
|
||||
.where(eq(joinRequests.id, requestId))
|
||||
.returning()
|
||||
.then((rows) => rows[0]);
|
||||
.then((rows) => rows[0]));
|
||||
|
||||
await logActivity(db, {
|
||||
companyId,
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
import { approvalExecutionIssueCondition } from "./approval-visibility.js";
|
||||
import { and, asc, eq, isNull, lte, or, sql, type SQL } from "drizzle-orm";
|
||||
import { and, asc, desc, eq, isNull, lte, or, sql, type SQL } from "drizzle-orm";
|
||||
import { approvals, companies, decisionRetention, type Db } from "@paperclipai/db";
|
||||
import { decisionRetentionService, DEFAULT_DECISION_ARCHIVE_DAYS } from "./decision-retention.js";
|
||||
import { DEFAULT_DECISION_ARCHIVE_DAYS } from "./decision-retention.js";
|
||||
import { createDecisionRetentionSweep } from "./decision-retention-sweep.js";
|
||||
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
|
||||
import { registerDeadlineWork } from "./deadline-work.js";
|
||||
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
@@ -9,51 +10,34 @@ import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
const archiveDelay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
|
||||
|
||||
export function approvalRetentionService(db: Db) {
|
||||
let cursor: { at: string; id: string } | undefined;
|
||||
function candidates(database: Db, condition?: SQL) {
|
||||
return database.select({ id: approvals.id, companyId: approvals.companyId, at: approvals.updatedAt, cursorAt: sql<string>`${approvals.updatedAt}::text` }).from(approvals)
|
||||
function candidates(database: Db, condition?: SQL, order = asc) {
|
||||
return database.select({ id: approvals.id, companyId: approvals.companyId, at: approvals.updatedAt, cursorAt: sql<string>`${approvals.updatedAt}::text`, deadlineAt: sql<number>`ceil(extract(epoch from ${approvals.updatedAt}) * 1000 + ${archiveDelay})`.mapWith(Number) }).from(approvals)
|
||||
.innerJoin(companies, eq(companies.id, approvals.companyId))
|
||||
.leftJoin(decisionRetention, and(eq(decisionRetention.companyId, approvals.companyId),
|
||||
eq(decisionRetention.sourceKind, "approval"), sql`${decisionRetention.sourceId} = ${approvals.id}::text`))
|
||||
.where(and(sql`${approvals.status} = 'pending'`, eq(companies.status, "active"), approvalExecutionIssueCondition(),
|
||||
or(isNull(decisionRetention.id), eq(decisionRetention.keep, false)), isNull(decisionRetention.archivedAt), condition))
|
||||
.orderBy(asc(approvals.updatedAt), asc(approvals.id));
|
||||
.orderBy(order(approvals.updatedAt), order(approvals.id));
|
||||
}
|
||||
async function sweep(now = new Date(), signal?: AbortSignal) {
|
||||
const cutoff = new Date(now.getTime() - archiveDelay);
|
||||
const rows = await candidates(db, and(lte(approvals.updatedAt, cutoff), cursor
|
||||
? sql`(${approvals.updatedAt}, ${approvals.id}) > (${cursor.at}::timestamptz, ${cursor.id}::uuid)` : undefined)).limit(100);
|
||||
let archived = 0;
|
||||
const errors: unknown[] = [];
|
||||
for (const candidate of rows) {
|
||||
if (signal?.aborted) break;
|
||||
try {
|
||||
archived += await db.transaction(async tx => {
|
||||
const database = tx as unknown as Db;
|
||||
const [current] = await candidates(database, and(eq(approvals.id, candidate.id), lte(approvals.updatedAt, cutoff)))
|
||||
.for("update", { of: approvals });
|
||||
if (!current) return 0;
|
||||
const retention = decisionRetentionService(database);
|
||||
const source = { sourceKind: "approval" as const, subject: { id: current.id }, activityAt: current.at.toISOString() };
|
||||
const states = await retention.syncItems(current.companyId, [source]);
|
||||
const state = states.get(`approval:${current.id}`)!;
|
||||
return retention.autoArchive({ companyId: current.companyId, now, items: [{ ...source,
|
||||
keep: state.keep, archivedAt: state.archivedAt?.toISOString() ?? null, retentionVersion: state.version,
|
||||
}] });
|
||||
});
|
||||
} catch (error) { errors.push(error); }
|
||||
cursor = { at: candidate.cursorAt, id: candidate.id };
|
||||
}
|
||||
if (rows.length < 100) cursor = undefined;
|
||||
if (errors.length === 1) throw errors[0];
|
||||
if (errors.length) throw new AggregateError(errors, "Approval retention failed");
|
||||
return { archived };
|
||||
}
|
||||
async function next() {
|
||||
const [row] = await candidates(db).limit(1);
|
||||
return row ? row.at.getTime() + archiveDelay : null;
|
||||
}
|
||||
return { sweep, next };
|
||||
return createDecisionRetentionSweep(db, {
|
||||
sourceKind: "approval", failureMessage: "Approval retention failed",
|
||||
end: async cutoff => {
|
||||
const [row] = await candidates(db, lte(approvals.updatedAt, cutoff), desc).limit(1);
|
||||
return row ? { at: row.cursorAt, id: row.id } : undefined;
|
||||
},
|
||||
page: async (cutoff, cursor, end, limit) => candidates(db, and(lte(approvals.updatedAt, cutoff),
|
||||
sql`(${approvals.updatedAt}, ${approvals.id}) <= (${end.at}::timestamptz, ${end.id}::uuid)`, cursor
|
||||
? sql`(${approvals.updatedAt}, ${approvals.id}) > (${cursor.at}::timestamptz, ${cursor.id}::uuid)` : undefined)).limit(limit),
|
||||
lock: async (database, id, cutoff) => {
|
||||
const [row] = await candidates(database, and(eq(approvals.id, id), lte(approvals.updatedAt, cutoff)))
|
||||
.for("update", { of: approvals });
|
||||
return row;
|
||||
},
|
||||
next: async () => {
|
||||
const [row] = await candidates(db).limit(1);
|
||||
return row?.deadlineAt ?? null;
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
export function registerApprovalRetentionWork(
|
||||
|
||||
@@ -1421,6 +1421,7 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions
|
||||
}));
|
||||
}
|
||||
|
||||
if (!options.excludeSources?.includes("join_request")) {
|
||||
const pendingJoins = await db
|
||||
.select({
|
||||
id: joinRequests.id,
|
||||
@@ -1482,6 +1483,8 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions
|
||||
}));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
const recoveryRows = await db
|
||||
.select()
|
||||
.from(issueRecoveryActions)
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { notifyRetentionWork } from "./retention-work-notifications.js";
|
||||
import { notifyCompanyRetentionWork, notifyRetentionWork } from "./retention-work-notifications.js";
|
||||
import { addCents, compareCents, normalizeCents } from "@paperclipai/shared";
|
||||
import { and, desc, eq, inArray, sql, or, isNull } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
@@ -297,7 +297,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
|
||||
|
||||
const scope = await resolveScopeRecord(db, "company", policy.scopeId);
|
||||
if (scope.paused) return;
|
||||
await notifyRetentionWork(db, "approval");
|
||||
await notifyCompanyRetentionWork(db);
|
||||
await notifyDeliveryWork(db, DELIVERY_QUEUES.agentLifecycle, policy.companyId);
|
||||
await notifyDeliveryWork(db, DELIVERY_QUEUES.skillSourceSync);
|
||||
await db
|
||||
@@ -354,7 +354,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
|
||||
await notifyDeliveryWork(db, DELIVERY_QUEUES.skillSourceSync);
|
||||
const scope = await resolveScopeRecord(db, "company", policy.scopeId);
|
||||
if (scope.pauseReason !== "budget") return;
|
||||
await notifyRetentionWork(db, "approval");
|
||||
await notifyCompanyRetentionWork(db);
|
||||
await notifyDeliveryWork(db, DELIVERY_QUEUES.agentLifecycle, policy.companyId);
|
||||
await db
|
||||
.update(companies)
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyRetentionWork } from "../../retention-work-notifications.js";
|
||||
import { createHash, randomBytes } from "node:crypto";
|
||||
import { and, desc, eq, gt, or, sql } from "drizzle-orm";
|
||||
|
||||
@@ -97,6 +98,7 @@ export function createSlackIdentityLinks(db: Db, { previewIdentityLink, stagePro
|
||||
inviteType: "company_join", allowedJoinTypes: "human", acceptedAt: now, expiresAt: now,
|
||||
defaultsPayload: { human: { role: "operator" }, source: "slack_identity_link" },
|
||||
}).returning({ id: invites.id });
|
||||
await notifyRetentionWork(tx, "join_request");
|
||||
const [request] = await tx.insert(joinRequests).values({
|
||||
inviteId: invite.id, companyId: preview.companyId, requestType: "human", status: "pending_approval",
|
||||
requestIp, requestingUserId: userId, requestEmailSnapshot: user.email,
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { notifyRetentionWork } from "./retention-work-notifications.js";
|
||||
import { notifyCompanyRetentionWork } from "./retention-work-notifications.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
import { isRunnerApiResponseAsset } from "./asset-content-access.js";
|
||||
import { deleteCompany } from "./company-deletion.js";
|
||||
@@ -331,7 +331,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
|
||||
|
||||
if (companyPatch.status !== undefined && companyPatch.status !== existing.status) {
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
|
||||
await notifyRetentionWork(tx, "approval");
|
||||
await notifyCompanyRetentionWork(tx);
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id);
|
||||
}
|
||||
const updated = await tx
|
||||
@@ -483,7 +483,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
|
||||
|
||||
if (!wasAlreadyArchived) {
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
|
||||
await notifyRetentionWork(tx, "approval");
|
||||
await notifyCompanyRetentionWork(tx);
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id);
|
||||
await tx
|
||||
.update(companies)
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { notifyRetentionWork } from "./retention-work-notifications.js";
|
||||
import { notifyCompanyRetentionWork } from "./retention-work-notifications.js";
|
||||
import { notifyToolReviewChange } from "./tool-review-work-notifications.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
import { agentObservabilityService } from "./agent-observability.js";
|
||||
@@ -97,7 +97,7 @@ export async function deleteCompany(db: Db, id: string) {
|
||||
await tx.delete(agentRuntimeState).where(eq(agentRuntimeState.companyId, id));
|
||||
await tx.delete(issueComments).where(eq(issueComments.companyId, id));
|
||||
await tx.delete(approvalComments).where(eq(approvalComments.companyId, id));
|
||||
await notifyRetentionWork(tx, "approval");
|
||||
await notifyCompanyRetentionWork(tx);
|
||||
await tx.delete(approvals).where(eq(approvals.companyId, id));
|
||||
await tx.delete(companySecrets).where(eq(companySecrets.companyId, id));
|
||||
await tx.delete(joinRequests).where(eq(joinRequests.companyId, id));
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import type { AttentionSourceKind } from "@paperclipai/shared";
|
||||
import { decisionRetentionService, DEFAULT_DECISION_ARCHIVE_DAYS } from "./decision-retention.js";
|
||||
|
||||
type Candidate = { id: string; companyId: string; at: Date; cursorAt: string };
|
||||
type Cursor = { at: string; id: string };
|
||||
|
||||
export function createDecisionRetentionSweep(db: Db, options: {
|
||||
sourceKind: AttentionSourceKind;
|
||||
failureMessage: string;
|
||||
end(cutoff: Date): Promise<Cursor | undefined>;
|
||||
page(cutoff: Date, cursor: Cursor | undefined, end: Cursor, limit: number): Promise<Candidate[]>;
|
||||
lock(database: Db, id: string, cutoff: Date): Promise<Candidate | undefined>;
|
||||
next(): Promise<number | null>;
|
||||
}) {
|
||||
let cursor: Cursor | undefined;
|
||||
let end: Cursor | undefined;
|
||||
const batchSize = 100;
|
||||
async function sweep(now = new Date(), signal?: AbortSignal) {
|
||||
const cutoff = new Date(now.getTime() - DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000);
|
||||
end ??= await options.end(cutoff);
|
||||
if (!end) return { archived: 0 };
|
||||
const rows = await options.page(cutoff, cursor, end, batchSize);
|
||||
let archived = 0;
|
||||
const errors: unknown[] = [];
|
||||
for (const candidate of rows) {
|
||||
if (signal?.aborted) break;
|
||||
try {
|
||||
archived += await db.transaction(async tx => {
|
||||
const database = tx as unknown as Db;
|
||||
const current = await options.lock(database, candidate.id, cutoff);
|
||||
if (!current) return 0;
|
||||
const retention = decisionRetentionService(database);
|
||||
const source = { sourceKind: options.sourceKind, subject: { id: current.id }, activityAt: current.at.toISOString() };
|
||||
const states = await retention.syncItems(current.companyId, [source]);
|
||||
const state = states.get(`${options.sourceKind}:${current.id}`)!;
|
||||
return retention.autoArchive({ companyId: current.companyId, now, items: [{ ...source,
|
||||
keep: state.keep, archivedAt: state.archivedAt?.toISOString() ?? null, retentionVersion: state.version,
|
||||
}] });
|
||||
});
|
||||
} catch (error) { errors.push(error); }
|
||||
cursor = { at: candidate.cursorAt, id: candidate.id };
|
||||
}
|
||||
if (rows.length < batchSize) { cursor = undefined; end = undefined; }
|
||||
if (errors.length === 1) throw errors[0];
|
||||
if (errors.length) throw new AggregateError(errors, options.failureMessage);
|
||||
return { archived };
|
||||
}
|
||||
return { sweep, next: options.next };
|
||||
}
|
||||
@@ -15,6 +15,7 @@ import {
|
||||
issueRecoveryActions,
|
||||
issueThreadInteractions,
|
||||
issues,
|
||||
joinRequests,
|
||||
} from "@paperclipai/db";
|
||||
import type {
|
||||
AttentionArchiveManifestEntry,
|
||||
@@ -183,13 +184,14 @@ export function decisionRetentionService(
|
||||
)).then((rows) => rows[0] ?? null);
|
||||
}
|
||||
|
||||
async function materializeApprovalState(database: Db, input: { companyId: string; sourceKind: AttentionSourceKind; sourceId: string }) {
|
||||
if (input.sourceKind !== "approval") return;
|
||||
const [source] = await database.select({ updatedAt: approvals.updatedAt }).from(approvals)
|
||||
.where(and(eq(approvals.companyId, input.companyId), eq(approvals.id, input.sourceId))).for("update");
|
||||
async function materializeSourceState(database: Db, input: { companyId: string; sourceKind: AttentionSourceKind; sourceId: string }) {
|
||||
const table = input.sourceKind === "approval" ? approvals : input.sourceKind === "join_request" ? joinRequests : null;
|
||||
if (!table) return;
|
||||
const [source] = await database.select({ updatedAt: table.updatedAt }).from(table)
|
||||
.where(and(eq(table.companyId, input.companyId), eq(table.id, input.sourceId))).for("update");
|
||||
if (!source) throw notFound("Attention source not found");
|
||||
await decisionRetentionService(database).syncItems(input.companyId, [{
|
||||
sourceKind: "approval", subject: { id: input.sourceId }, activityAt: source.updatedAt.toISOString(),
|
||||
sourceKind: input.sourceKind, subject: { id: input.sourceId }, activityAt: source.updatedAt.toISOString(),
|
||||
}]);
|
||||
}
|
||||
|
||||
@@ -205,7 +207,7 @@ export function decisionRetentionService(
|
||||
throw notFound("Attention source not found");
|
||||
}
|
||||
const updated = await writeRetention(input.sourceKind, async tx => {
|
||||
await materializeApprovalState(tx, input);
|
||||
await materializeSourceState(tx, input);
|
||||
return tx.update(decisionRetention).set({
|
||||
keep: input.keep,
|
||||
version: sql`${decisionRetention.version} + 1`,
|
||||
@@ -246,7 +248,7 @@ export function decisionRetentionService(
|
||||
}
|
||||
const now = new Date();
|
||||
const updated = await writeRetention(input.sourceKind, async tx => {
|
||||
await materializeApprovalState(tx, input);
|
||||
await materializeSourceState(tx, input);
|
||||
return tx.update(decisionRetention).set(input.archived ? {
|
||||
archivedAt: now,
|
||||
archivedReason: "manual",
|
||||
|
||||
@@ -10,6 +10,7 @@ export const DELIVERY_QUEUES = {
|
||||
executionDisposition: "execution-disposition",
|
||||
environmentCleanup: "environment-cleanup",
|
||||
approvalRetention: "approval-retention",
|
||||
joinRequestRetention: "join-request-retention",
|
||||
decisionTargets: "decision-targets",
|
||||
decisionEffectRecovery: "decision-effect-recovery",
|
||||
decisionExpiry: "decision-expiry",
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
import { and, asc, desc, eq, isNull, lte, or, sql, type SQL } from "drizzle-orm";
|
||||
import { companies, decisionRetention, invites, joinRequests, type Db } from "@paperclipai/db";
|
||||
import { DEFAULT_DECISION_ARCHIVE_DAYS } from "./decision-retention.js";
|
||||
import { createDecisionRetentionSweep } from "./decision-retention-sweep.js";
|
||||
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
|
||||
import { registerDeadlineWork } from "./deadline-work.js";
|
||||
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
|
||||
const archiveDelay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
|
||||
|
||||
export function joinRequestRetentionService(db: Db) {
|
||||
function candidates(database: Db, condition?: SQL, order = asc) {
|
||||
return database.select({ id: joinRequests.id, companyId: joinRequests.companyId,
|
||||
at: joinRequests.updatedAt, cursorAt: sql<string>`${joinRequests.updatedAt}::text`, deadlineAt: sql<number>`ceil(extract(epoch from ${joinRequests.updatedAt}) * 1000 + ${archiveDelay})`.mapWith(Number) }).from(joinRequests)
|
||||
.innerJoin(companies, eq(companies.id, joinRequests.companyId))
|
||||
.innerJoin(invites, and(eq(invites.id, joinRequests.inviteId), eq(invites.companyId, joinRequests.companyId)))
|
||||
.leftJoin(decisionRetention, and(eq(decisionRetention.companyId, joinRequests.companyId),
|
||||
eq(decisionRetention.sourceKind, "join_request"), sql`${decisionRetention.sourceId} = ${joinRequests.id}::text`))
|
||||
.where(and(sql`${joinRequests.status} = 'pending_approval'`, eq(companies.status, "active"),
|
||||
or(isNull(decisionRetention.id), eq(decisionRetention.keep, false)), isNull(decisionRetention.archivedAt), condition))
|
||||
.orderBy(order(joinRequests.updatedAt), order(joinRequests.id));
|
||||
}
|
||||
return createDecisionRetentionSweep(db, {
|
||||
sourceKind: "join_request", failureMessage: "Join request retention failed",
|
||||
end: async cutoff => {
|
||||
const [row] = await candidates(db, lte(joinRequests.updatedAt, cutoff), desc).limit(1);
|
||||
return row ? { at: row.cursorAt, id: row.id } : undefined;
|
||||
},
|
||||
page: async (cutoff, cursor, end, limit) => candidates(db, and(lte(joinRequests.updatedAt, cutoff),
|
||||
sql`(${joinRequests.updatedAt}, ${joinRequests.id}) <= (${end.at}::timestamptz, ${end.id}::uuid)`, cursor
|
||||
? sql`(${joinRequests.updatedAt}, ${joinRequests.id}) > (${cursor.at}::timestamptz, ${cursor.id}::uuid)` : undefined)).limit(limit),
|
||||
lock: async (database, id, cutoff) => {
|
||||
const [row] = await candidates(database, and(eq(joinRequests.id, id), lte(joinRequests.updatedAt, cutoff)))
|
||||
.for("update", { of: joinRequests });
|
||||
return row;
|
||||
},
|
||||
next: async () => {
|
||||
const [row] = await candidates(db).limit(1);
|
||||
return row?.deadlineAt ?? null;
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
export function registerJoinRequestRetentionWork(
|
||||
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
|
||||
service: ReturnType<typeof joinRequestRetentionService>,
|
||||
canRun: () => boolean,
|
||||
) {
|
||||
return registerDeadlineWork(coordinator, DELIVERY_QUEUES.joinRequestRetention, {
|
||||
run: signal => service.sweep(new Date(), signal), next: service.next, canRun, retryMs: 30_000, notificationCoalesceMs: 30_000,
|
||||
});
|
||||
}
|
||||
@@ -4,6 +4,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio
|
||||
|
||||
export async function notifyRetentionWork(transaction: object, kind: AttentionSourceKind) {
|
||||
if (kind === "approval") await notifyDeliveryWork(transaction, DELIVERY_QUEUES.approvalRetention);
|
||||
if (kind === "join_request") await notifyDeliveryWork(transaction, DELIVERY_QUEUES.joinRequestRetention);
|
||||
}
|
||||
|
||||
export function writeApprovalChange<T>(db: Db, write: (transaction: Db) => Promise<T>): Promise<T> {
|
||||
@@ -12,3 +13,15 @@ export function writeApprovalChange<T>(db: Db, write: (transaction: Db) => Promi
|
||||
return write(tx as unknown as Db);
|
||||
});
|
||||
}
|
||||
|
||||
export async function notifyCompanyRetentionWork(transaction: object) {
|
||||
await notifyRetentionWork(transaction, "approval");
|
||||
await notifyRetentionWork(transaction, "join_request");
|
||||
}
|
||||
|
||||
export function writeJoinRequestChange<T>(db: Db, write: (transaction: Db) => Promise<T>): Promise<T> {
|
||||
return db.transaction(async tx => {
|
||||
await notifyRetentionWork(tx, "join_request");
|
||||
return write(tx as unknown as Db);
|
||||
});
|
||||
}
|
||||
Reference in new issue
Block a user