fix(server): schedule gateway credential expiry from committed changes (#17450)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - The tool gateway stores credentials for sessions and named gateways.
> - Those credentials already contain expiry dates.
> - The server repeatedly scans their tables even when no expiry work
exists.
> - Creation signals and saved deadlines can drive cleanup without idle
scans.
> - This PR registers that work with the existing coordinator.
> - Authentication continues to reject expired credentials independently
of physical cleanup.

## Linked Issues or Issue Description

Refs #17446, #17221, #17274. This is the third sleep-migration slice,
stacked on #17446.

**What problem does this solve?**
Gateway credential cleanup queries PostgreSQL on every maintenance tick,
including empty tables and retained non-expiring tokens.

**What is the proposed solution?**
Signal expiring token/session creation from its transaction. Restore the
earliest saved expiry at startup. Retry only while cleanup or
reconciliation remains outstanding.

**Why is this useful?**
Configured idle gateways no longer create recurring expiry traffic.
Existing credential data remains the durable source of truth.

Related work: #17221 and #17274 change run-token lifetime and cleanup
eligibility. They remain separate. This PR preserves current master
authorization and deletion rules. When those PRs are combined, update
the deadline query and signals to match their new operator-token expiry
predicate; run-token idle markers must not keep this worker scanning.
Retain the separate run-liveness reaper and run both regression suites.
Text mergeability is insufficient evidence for that integration.

## What Changed

- Make the credential test clock follow real scheduled work without
overshooting future-expiry assertions; production behavior is unchanged
by this fixture follow-up.

- Coalesce sequential credential notifications within the former cleanup
cadence, preserve loaded expiry deadlines and immediate authentication
rejection, and bound scheduled deletion to 500 rows per pass.

- Added transaction-aware intent before the two existing credential
inserts. Non-expiring named tokens do not signal expiry work.
- Added earliest-expiry selection in the gateway service and a narrow
registration wrapper for the shared deadline helper.
- Removed gateway credential cleanup from the aggregate maintenance
tick. Startup reconstructs saved deadlines. Heartbeat-disabled instances
retain their startup-only cleanup behavior.
- Preserved expiry/revocation checks, run authorization, audit behavior,
and the existing 500-token cleanup batch limit.
- Added real PostgreSQL tests and a lifecycle/producer document.
Tool-action review recovery remains a separate slice.

- Drain all expired session/token pages in explicit and
scheduler-disabled startup cleanup, while retaining one-page scheduled
passes and authentication enforcement.

## Verification

- At `907098cd3656bac59c60220c73b085d0349d4b5f`, all 290 Runner
transport tests pass locally (including all seven prepared-environment
variants). CI run `40570815360` attempt 1 had one provider-drain
settlement failure in the `close-attachment` case; the identical source
passed the preceding CI lane. A single failed-job rerun is in progress.
The failure remains recorded; no shutdown assertion or production
behavior was changed to suppress it.

- Master landing reconciliation at
`907098cd3656bac59c60220c73b085d0349d4b5f` after #17446 merged: all
production and test files are byte-identical to
`24318d9b99ee7f044a864ff3981fc1929bf3e509`, where the 150-case
acceptance and server types passed. Only commit ancestry and the plan's
merged status changed. Fresh CI/review is pending on this head.

- Parent reconciliation at `24318d9b99ee7f044a864ff3981fc1929bf3e509`:
150 affected tests pass across gateway credential expiry, the complete
tool gateway suite, custom-image worker and startup. The accepted
PostgreSQL runs skip no cases. Server types pass. Existing direct tests
cover sequential-write coalescing and saved-deadline precedence; the
older Ponytail finding needs a fresh review. Remote CI is pending.

Current head: `10c85a5eaa288867aa2997a27aee5f675c2a713d`. This remains a
single-duty PR stacked on #17446.

- Integrated pinned master `8548173fc9efdc0fa07644fb92dcc0b6851a95e5`
with normal merges. Every branch passed generated-schema checks and its
target-branch migration-order gate. Existing SQL is unchanged; the 25
stack migrations follow master’s new migration.
- Duty-specific producer coverage, lifecycle tests, and earlier failure
records:
[2026-10-08-instance-sleep-and-scheduler-inventory.md](https://github.com/paperclipai/paperclip/blob/10c85a5eaa288867aa2997a27aee5f675c2a713d/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md),
[2026-10-24-recurring-server-sleep-stack.md](https://github.com/paperclipai/paperclip/blob/10c85a5eaa288867aa2997a27aee5f675c2a713d/doc/plans/2026-10-24-recurring-server-sleep-stack.md).
- Combined tip `51e6cbe67979b55b340ea9f0f434c5b5c682ccd6`: **355
startup/cancellation/recovery/dispatch/preservation tests**, **75
connection-management tests**, and production server typechecking pass.
These combined checks do not mean every intermediate PR independently
ran the full repository suite.
- Real master-to-stack database upgrade passes: **379 unchanged prior
migration receipts plus 25 new migrations**, with the gateway
responsibility fixture preserved.
- Owning-branch checks: 84 runner installation/settlement cases plus
runner typechecking and generated TypeScript/Rust profile agreement; 253
GitHub review workflow cases. Stop-attribution and held-worker readiness
regressions failed before their fixes.
- Failure accounting: the initial combined run passed 349 cases and
failed six idle assertions in a reused preservation database containing
old cancelled runs. Repeating the combined command with a fresh
preservation database passed. Earlier invalid project selection, merge
import/type errors, and the interrupted full local suite are not counted
as passing checks.
- **Latest-head CI and review remain required.** Prior CI findings
include GitHub idle observation, route teardown, attention cleanup,
touch-picker E2E, workspace health, and voice-cleanup timing; local
passes do not erase those failures. No full local repository-suite,
deployed sleep, fleet cost, or live Copilot qualification claim is made.

## Risks

- Shared-owner signals cover supported local producers. Independent SQL
writers or another replica need a separate ingress contract; startup
recovery is not live discovery.
- This is scheduling only. The open run-token lifetime PRs need semantic
integration as described above before combining their behavior.
- Expiry cleanup can wait during drain; authentication still enforces
expiry on use. Existing Cloud safety gates are preserved. Local
deadlines do not prove external wake coverage or deployed suspension.
- Revocation does not move expiry. Non-expiring tokens remain retained
by design; this PR does not invent a retention deadline.

## 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

Latest-head CI is now green after one evidence-backed retry. Greptile is
5/5 and all review threads are resolved.

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-10-27 14:58:11 -05:00
1 parent 3cd3af5d75
commit 5398275d2a
16 files changed
+609 -75

No files matched your search

+141
View File
@@ -0,0 +1,141 @@
# Gateway credential expiry and idle behavior
Gateway sessions and named gateway tokens already persist their expiry times.
`cleanupExpiredSessions` remains the owner of physical deletion. Authentication
continues to check expiry, revocation, company scope, run ownership, and active
run state on use. No table, authorization rule, or external alarm is added.
## Producer inventory
- `createSession` registers intent before inserting a run-bound session in a
transaction. Its existing audit and returned credential remain unchanged.
- `createNamedGatewayToken` registers intent before inserting an expiring token
in a transaction. A token without an expiry owes no expiry cleanup and does
not notify this worker.
- Last-use updates do not change expiry. Token/session revocation, including
heartbeat completion and connection removal, also leaves the saved expiry
intact. Revoked rows retain the same physical deletion deadline.
- Gateway/company deletion removes dependent rows and can leave one harmless
stale deadline. It cannot create overdue work. No supported service changes
a saved credential expiry after creation.
The database signal layer owns outer commit, rollback, savepoints, and uncertain
outcomes. The worker supports producers sharing that signal owner. Independent
SQL writers and another process need a separate live ingress contract. Startup
reconstructs existing rows but does not provide continuous replica discovery.
## Worker contract
`nextCredentialExpiryAt` reads one earliest non-null expiry from each table using
expiry-first indexes, then compares those at most two values. The migration adds
the session expiry index; the token expiry index already exists.
An empty credential store or retained non-expiring tokens returns no deadline.
`registerGatewayCredentialExpiryWork` supplies cleanup and deadline selection to
the shared `registerDeadlineWork` helper. Provider and authentication behavior
stay in the gateway service; startup only registers the worker.
The worker starts where the old startup cleanup ran. With heartbeat scheduling
enabled, committed changes and actual expiry times replace the recurring cleanup
call. With heartbeat scheduling disabled, the existing startup-only cleanup is
preserved. Tool action-review recovery is a separate obligation and is unchanged.
Token and session deletion are each bounded to 500 rows per pass. Session
selection orders by expiry and rechecks expiry when deleting the selected IDs.
A remaining overdue expiry schedules another pass after 30 seconds. Failed reconciliation also
retries at that bound. No unfinished reconciliation or future expiry means no
recurring SQL. A generation retains writes arriving during reconciliation.
Startup restores deadlines; drain retains first writes without passive SQL;
resume completes overdue cleanup. The coordinator owns shutdown and waits for
in-flight work.
Sequential committed writes are coalesced for up to 30 seconds after a
successful cleanup pass, matching the former heartbeat cleanup cadence. A
previously loaded expiry can preempt that window. A newly reported credential
may physically remain until that bounded follow-up; authentication still rejects
it immediately at expiry. A successful pass with no future expiry and no new
notifications creates no timer. This uses the shared optional notification
coalescing capability, promoted from the later plugin/decision recovery slices;
workers that omit it retain their existing behavior.
## Evidence and limits
The concurrent session index runs through the migration executor's existing
per-file top-level path. Initial application and repeat application were tested.
A PostgreSQL plan with 20,000 sessions and 20,000 tokens, including 19,000
non-expiring tokens, used both expiry indexes and read one row from each.
This establishes bounded lookup on that fixture, not production latency.
The new PostgreSQL suite exercises real session/token creation, configured idle
state, non-expiring history, expiry ordering, revoked-token cleanup, outer commit
and rollback, startup reconstruction, a backlog larger than one cleanup batch,
and drain/resume. It tests expired authentication before deferred physical
cleanup, through both session and named-gateway entry points. Existing gateway
acceptance and server-startup tests cover the changed integration boundaries.
Use only disposable migrated databases for the test URL overrides. Their names
must end in `_test`; the suites delete their fixtures. Ordinary CI uses embedded
PostgreSQL. Test credentials are synthetic and no external provider is contacted
by the expiry suite.
The migration proves worker behavior, not host suspension. Existing Cloud idle
proof remains conservative. An in-process deadline cannot wake stopped compute;
external wake coverage must be verified before changing sleep eligibility.
## Related unmerged token-lifetime work
At implementation time, PRs #17221 and #17274 change run-scoped token lifetime
and cleanup eligibility. This slice preserves the current master policy and
changes scheduling only. When combining those changes, the deadline query and
producer signals must use the same token eligibility as cleanup: run-token idle
markers must not schedule the operator-token expiry worker. Keep the run-liveness
reaper with its own owner and retain both regression suites. A conflict-free
text merge alone does not prove this integration.
## Notification amplification review repair (2026-10-27)
Ten sequential real session/token creation transactions reproduced ten extra
instance-wide cleanup passes. With coalescing they cause one follow-up pass.
A separate 501-session fixture reproduced the unbounded delete, then passed
with 500 removed initially and the remaining overdue row removed at the saved
30-second retry. A still-valid session survives both passes. The tests also
preserve an already loaded expiry only 500 milliseconds away and verify that a
new one-second session is rejected while physical cleanup waits for coalescing.
The initial full run exposed four outer-transaction tests whose fake-clock wait
expected the intent hold to release before the new coalescing follow-up. Their
settlement helper now advances the clock through that window while preserving
the original commit/rollback assertions. The complete gateway expiry, gateway
acceptance, custom-image deadline, and startup suites pass: 147 tests in four
files. Server typechecking also passes on the owning branch. This is local
behavioral evidence; combined-stack and fresh remote checks remain required.
The full local test:run remains unverified because of embedded PostgreSQL
startup/capacity limits; the two real gateway suites used separate migrated,
task-owned disposable PostgreSQL databases.
A later combined run exposed a test-clock overshoot: repeated blind 30-second
advances could expire the future session before its intermediate assertion.
The settlement helper now lets real SQL settle and advances only to an actually
armed follow-up within the coalescing window. All 12 credential expiry tests pass
on the owning gateway branch with the existing exact-deadline and transaction
assertions intact. This follow-up changes the test fixture only.
## One-shot cleanup preserves complete backlog removal
Review found that bounding worker passes also truncated startup cleanup when the
heartbeat scheduler was disabled. That path calls cleanup only once, so it could
leave every expired token or session after the first 500 rows indefinitely.
`cleanupExpiredSessions` now drains the existing backlog in 500-row SQL batches
by default, using one fixed expiry cutoff for the invocation. The deadline worker
opts into `singleBatch: true` and retains its existing coalescing and next-page
deadline. Future and nonexpiring credentials survive both paths. The return value
still counts deleted sessions, as before. No periodic fallback was added.
Three real-PostgreSQL regressions reproduced the truncation with 1,001 expired
sessions, tokens, or both, without registering a worker. They pass after the fix.
All 55 expiry/startup cases pass, including the existing worker's 501-row bounded
pass and commit/rollback tests. The original explicit-cleanup compatibility case
and server typecheck also pass. Evidence is in `/tmp/sleep-master-gateway-manual-`
logs: `red`, `green`, `compatibility`, and `types`. This fixes one-shot behavior;
it does not establish provider suspension or complete the stack's CI acceptance.
@@ -649,8 +649,10 @@ Longer polling intervals alone do not complete an item.
- [ ] **JOB-21 — Secret proposals and expired action tokens (partially migrated).**
Secret-proposal expiry now uses transaction signals and saved expiry deadlines;
see [the lifecycle contract](../idle-secret-proposal-expiry.md). Tool review
recovery and gateway token cleanup still use **30s** queries. Change: enforce expiry synchronously on use and
see [the lifecycle contract](../idle-secret-proposal-expiry.md). Gateway token
and session expiry now also uses saved deadlines and producer signals; see
[its lifecycle contract](../idle-gateway-credential-expiry.md). Tool review
recovery still uses **30s** queries. Change: enforce expiry synchronously on use and
register required scrub/cleanup deadlines. **Complete when:** idle cleanup
does not extend authorization or secret-retention periods. Sources:
[secret-proposals.ts](../../server/src/services/secret-proposals.ts),
@@ -18,7 +18,7 @@ Do not mark an item done from a timer removal or a green check alone.
| 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Pending |
| 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Pending |
| 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 | Secret-proposal expiry in progress; tool reviews and gateway tokens pending |
| 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Secret proposals: [#17446](https://github.com/paperclipai/paperclip/pull/17446), merged after CI and passing reviews; gateway credential expiry in progress; tool reviews pending |
| 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 |
+1
View File
@@ -501,6 +501,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([
"0332_little_night_nurse.sql",
"0345_luxuriant_carmella_unuscione.sql",
"0369_late_logan.sql",
"0381_mean_scream.sql",
]);
async function applyPendingMigrationsManually(
@@ -0,0 +1 @@
CREATE INDEX CONCURRENTLY IF NOT EXISTS "tool_gateway_sessions_expiry_idx" ON "tool_gateway_sessions" USING btree ("expires_at");
@@ -1,6 +1,6 @@
{
"id": "3e17203b-ec95-41e9-8363-685982450199",
"prevId": "945c31ed-f39f-4272-a91d-85de925f5361",
"id": "2e93deb7-f24c-42d1-8c8e-30b21cff90bc",
"prevId": "bfda87fa-662b-47bb-87c0-dabc814aebd7",
"version": "7",
"dialect": "postgresql",
"tables": {
@@ -47782,6 +47782,18 @@
"primaryKey": false,
"notNull": false
},
"git_authorization_grant_id": {
"name": "git_authorization_grant_id",
"type": "uuid",
"primaryKey": false,
"notNull": false
},
"git_authorized_by_user_id": {
"name": "git_authorized_by_user_id",
"type": "text",
"primaryKey": false,
"notNull": false
},
"pause_reason": {
"name": "pause_reason",
"type": "text",
@@ -47920,6 +47932,19 @@
],
"onDelete": "no action",
"onUpdate": "no action"
},
"projects_git_authorization_grant_id_connection_grants_id_fk": {
"name": "projects_git_authorization_grant_id_connection_grants_id_fk",
"tableFrom": "projects",
"tableTo": "connection_grants",
"columnsFrom": [
"git_authorization_grant_id"
],
"columnsTo": [
"id"
],
"onDelete": "set null",
"onUpdate": "no action"
}
},
"compositePrimaryKeys": {},
@@ -55197,6 +55222,21 @@
"method": "btree",
"with": {}
},
"tool_gateway_sessions_expiry_idx": {
"name": "tool_gateway_sessions_expiry_idx",
"columns": [
{
"expression": "expires_at",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"concurrently": false,
"method": "btree",
"with": {}
},
"tool_gateway_sessions_run_idx": {
"name": "tool_gateway_sessions_run_idx",
"columns": [
@@ -55942,6 +55982,12 @@
"notNull": true,
"default": "''"
},
"responsible_user_id": {
"name": "responsible_user_id",
"type": "text",
"primaryKey": false,
"notNull": false
},
"owner_note": {
"name": "owner_note",
"type": "text",
+7
View File
@@ -2654,6 +2654,13 @@
"when": 1793060791724,
"tag": "0380_broken_aqueduct",
"breakpoints": true
},
{
"idx": 381,
"version": "7",
"when": 1793060791725,
"tag": "0381_mean_scream",
"breakpoints": true
}
]
}
+1
View File
@@ -714,6 +714,7 @@ export const toolGatewaySessions = pgTable(
uniqueIndex("tool_gateway_sessions_token_hash_uq").on(table.tokenHash),
index("tool_gateway_sessions_company_agent_idx").on(table.companyId, table.agentId),
index("tool_gateway_sessions_company_expires_idx").on(table.companyId, table.expiresAt),
index("tool_gateway_sessions_expiry_idx").on(table.expiresAt),
index("tool_gateway_sessions_run_idx").on(table.companyId, table.runId),
index("tool_gateway_sessions_issue_idx").on(table.companyId, table.issueId),
index("tool_gateway_sessions_gateway_idx").on(table.companyId, table.gatewayId),
@@ -0,0 +1,269 @@
import { randomUUID } from "node:crypto";
import { sql } from "drizzle-orm";
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
import { activityLog, agents, companies, createDb, heartbeatRuns, toolAccessAuditEvents, toolGatewaySessions,
toolMcpGateways, toolMcpGatewayTokens, toolProfiles } from "@paperclipai/db";
import { createToolGatewayService } from "../services/tool-gateway.js";
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
import { registerGatewayCredentialExpiryWork } from "../services/gateway-credential-expiry-work.js";
import { idleWorkSnapshot } from "../services/task-admission.js";
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_GATEWAY_EXPIRY_TEST_DATABASE_URL");
const supported = external ? true : (await getEmbeddedPostgresTestSupport()).supported;
describe.skipIf(!supported)("gateway credential expiry", () => {
let db: ReturnType<typeof createDb>;
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
let gateway: ReturnType<typeof createToolGatewayService>;
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
let companyId: string, agentId: string, runId: string, namedId: string;
let draining: boolean;
const errors: unknown[] = [];
beforeAll(async () => {
if (!external) temporary = await startEmbeddedPostgresTestDatabase("gateway-expiry-");
db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
}, 90_000);
afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
beforeEach(async () => {
await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
companyId = randomUUID(); agentId = randomUUID(); runId = randomUUID(); draining = false; errors.length = 0;
await db.insert(companies).values({ id: companyId, name: "Credential expiry", issuePrefix: "EXP" });
await db.insert(agents).values({ id: agentId, companyId, name: "Agent", role: "engineer", adapterType: "process" });
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "assignment", status: "running" });
const [profile] = await db.insert(toolProfiles).values({ companyId, profileKey: "expiry", name: "Expiry", defaultAction: "deny" }).returning();
gateway = createToolGatewayService(db, { toolActionSigningSecret: "expiry-test-only" });
namedId = (await gateway.createNamedGateway({ companyId, body: { name: "Expiry", profileId: profile.id } })).id;
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true });
});
afterEach(async () => {
await coordinator?.stop(); coordinator = undefined;
vi.useRealTimers(); vi.restoreAllMocks();
await db.delete(activityLog);
await db.delete(toolAccessAuditEvents);
await db.delete(toolGatewaySessions);
await db.delete(toolMcpGatewayTokens);
await db.delete(toolMcpGateways);
await db.delete(toolProfiles);
await db.delete(heartbeatRuns);
await db.delete(agents);
await db.delete(companies);
expect(errors).toEqual([]);
});
async function start() {
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining,
canReconcile: () => true, onError: error => errors.push(error) });
await registerGatewayCredentialExpiryWork(coordinator, gateway, () => !draining).ready;
}
async function settle(ms = 60_000) {
await vi.advanceTimersByTimeAsync(ms);
if (!draining) await vi.waitFor(async () => {
// Let real SQL settle before advancing only to an actually armed follow-up.
// Repeated blind clock jumps can expire the future credential under test.
if (idleWorkSnapshot().active > 0) {
await db.execute(sql`select pg_sleep(0.01)`);
const next = coordinator!.nextWakeAt();
if (next !== null && next <= Date.now() + 30_000) {
await vi.advanceTimersByTimeAsync(Math.max(0, next - Date.now()));
}
}
expect(idleWorkSnapshot().active).toBe(0);
}, { timeout: 5000 });
}
async function quiet() {
await settle(); expect(coordinator!.nextWakeAt()).toBeNull();
const reads = vi.spyOn(db, "execute"), selects = vi.spyOn(db, "select"), deletes = vi.spyOn(db, "delete");
await settle(60 * 60_000);
expect(reads).not.toHaveBeenCalled(); expect(selects).not.toHaveBeenCalled(); expect(deletes).not.toHaveBeenCalled();
reads.mockRestore(); selects.mockRestore(); deletes.mockRestore();
}
const token = (expiresAt?: Date) => gateway.createNamedGatewayToken({ companyId, gatewayId: namedId, body: { name: "Expiry", expiresAt } });
const session = (service = gateway) => service.createSession({ companyId, agentId, runId, ttlMs: 5 * 60_000 });
it("stays quiet with configuration and a nonexpiring token, then expires real tokens and sessions by deadline", async () => {
await start(); await quiet();
const retained = await token(); await quiet();
const active = await session();
const due = new Date(Date.now() + 3 * 60_000);
const expiring = await token(due);
await settle();
expect(coordinator!.nextWakeAt()).toBe(due.getTime());
const reads = vi.spyOn(db, "execute"), selects = vi.spyOn(db, "select");
await settle(30_000);
expect(reads).not.toHaveBeenCalled(); expect(selects).not.toHaveBeenCalled();
reads.mockRestore(); selects.mockRestore();
await gateway.revokeNamedGatewayToken({ companyId, tokenId: expiring.id });
await settle(2 * 60_000);
expect((await db.select().from(toolMcpGatewayTokens)).map(row => row.id)).toEqual([retained.id]);
expect(coordinator!.nextWakeAt()).toBe(active.expiresAt.getTime());
await settle(2 * 60_000);
expect(await db.select().from(toolGatewaySessions)).toEqual([]);
await quiet();
});
it.each([
{ kind: "session", rollback: false }, { kind: "session", rollback: true },
{ kind: "token", rollback: false }, { kind: "token", rollback: true },
])("respects $kind outer transaction settlement (rollback $rollback)", async ({ kind, rollback }) => {
await start(); await quiet();
let release!: () => void, written!: () => void;
const gate = new Promise<void>(resolve => { release = resolve; });
const staged = new Promise<void>(resolve => { written = resolve; });
const rollbackError = new Error("rollback fixture");
const writing = db.transaction(async tx => {
const scoped = createToolGatewayService(tx as unknown as typeof db, { toolActionSigningSecret: "expiry-test-only" });
if (kind === "session") await session(scoped);
else await scoped.createNamedGatewayToken({ companyId, gatewayId: namedId,
body: { name: "Outer transaction", expiresAt: new Date(Date.now() + 5 * 60_000) } });
written(); await gate; if (rollback) throw rollbackError;
}).catch(error => { if (error !== rollbackError) throw error; });
await Promise.race([staged, writing]);
try {
await vi.advanceTimersByTimeAsync(60_000);
expect(await db.select().from(toolGatewaySessions)).toEqual([]);
expect(await db.select().from(toolMcpGatewayTokens)).toEqual([]);
} finally { release(); }
await writing; await settle();
if (rollback) await quiet();
else {
expect(coordinator!.nextWakeAt()).toBe(await gateway.nextCredentialExpiryAt());
await settle(5 * 60_000); await quiet();
}
});
it("coalesces sequential credential writes within the former cleanup cadence", async () => {
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: false });
const cleanup = vi.spyOn(gateway, "cleanupExpiredSessions");
const next = vi.spyOn(gateway, "nextCredentialExpiryAt");
await start();
const firstPassAt = Date.now();
for (let i = 0; i < 10; i++) {
if (i % 2 === 0) await session();
else await token(new Date(Date.now() + 5 * 60_000));
await vi.advanceTimersByTimeAsync(1);
// Let each old scan finish so this measures sequential writes, not overlapping calls.
await db.execute(sql`select pg_sleep(0.01)`);
}
expect(cleanup).toHaveBeenCalledTimes(1);
expect(next).toHaveBeenCalledTimes(1);
expect(coordinator!.nextWakeAt()).toBe(firstPassAt + 30_000);
await vi.advanceTimersByTimeAsync(30_000);
await vi.waitFor(() => expect(next).toHaveBeenCalledTimes(2));
await db.execute(sql`select pg_sleep(0.01)`);
expect(cleanup).toHaveBeenCalledTimes(2);
expect(coordinator!.nextWakeAt()).toBeGreaterThan(Date.now());
await settle(5 * 60_000); await quiet();
});
it("keeps a loaded expiry deadline ahead of credential notification coalescing", async () => {
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: false });
const deadline = Date.now() + 500;
await token(new Date(deadline));
await start();
expect(coordinator!.nextWakeAt()).toBe(deadline);
await session();
await vi.advanceTimersByTimeAsync(1);
await db.execute(sql`select pg_sleep(0.01)`);
expect(coordinator!.nextWakeAt()).toBe(deadline);
await vi.advanceTimersByTimeAsync(499);
await vi.waitFor(async () => expect(await db.select().from(toolMcpGatewayTokens)).toEqual([]));
expect(await db.select().from(toolGatewaySessions)).toHaveLength(1);
await settle(5 * 60_000); await quiet();
});
it("rejects a newly expired credential while physical cleanup is coalesced", async () => {
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: false });
const cleanup = vi.spyOn(gateway, "cleanupExpiredSessions");
await start();
const at = Date.now();
const active = await gateway.createSession({ companyId, agentId, runId, ttlMs: 1000 });
await vi.advanceTimersByTimeAsync(1000);
await db.execute(sql`select pg_sleep(0.01)`);
expect(cleanup).toHaveBeenCalledTimes(1);
expect(await db.select().from(toolGatewaySessions)).toHaveLength(1);
await expect(gateway.listToolsForSession(active.token)).rejects.toMatchObject({ status: 401, reasonCode: "session_expired" });
expect(coordinator!.nextWakeAt()).toBe(at + 30_000);
await vi.advanceTimersByTimeAsync(29_000);
await vi.waitFor(async () => expect(await db.select().from(toolGatewaySessions)).toEqual([]));
await quiet();
});
it.each(["sessions", "tokens", "both"] as const)("drains the full %s backlog without a registered scheduler", async kind => {
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: false });
const now = new Date();
const futureSession = await session();
const futureToken = await token(new Date(now.getTime() + 60_000));
const nonexpiringToken = await token();
if (kind !== "tokens") await db.insert(toolGatewaySessions).values(Array.from({ length: 1001 }, () => ({
companyId, agentId, runId, tokenHash: randomUUID(), expiresAt: new Date(now.getTime() - 1),
})));
if (kind !== "sessions") await db.insert(toolMcpGatewayTokens).values(Array.from({ length: 1001 }, () => ({
companyId, gatewayId: namedId, name: "Expired", tokenHash: randomUUID(), expiresAt: new Date(now.getTime() - 1),
})));
await expect(gateway.cleanupExpiredSessions({ now })).resolves.toEqual({ deletedCount: kind === "tokens" ? 0 : 1001 });
expect((await db.select().from(toolGatewaySessions)).map(row => row.id)).toEqual([futureSession.id]);
expect((await db.select().from(toolMcpGatewayTokens)).map(row => row.id).sort()).toEqual([futureToken.id, nonexpiringToken.id].sort());
expect(coordinator).toBeUndefined();
expect(idleWorkSnapshot().active).toBe(0);
});
it("bounds session cleanup and schedules the remaining overdue batch", async () => {
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: false });
await db.insert(toolGatewaySessions).values(Array.from({ length: 501 }, () => ({
companyId, agentId, runId, tokenHash: randomUUID(), expiresAt: new Date(Date.now() - 1),
})));
const future = await session();
const cleanup = vi.spyOn(gateway, "cleanupExpiredSessions");
await start();
expect(cleanup).toHaveResolvedWith({ deletedCount: 500 });
expect(await db.select().from(toolGatewaySessions)).toHaveLength(2);
expect(coordinator!.nextWakeAt()).toBe(Date.now() + 30_000);
await vi.advanceTimersByTimeAsync(29_999);
expect(cleanup).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(1);
await vi.waitFor(async () => expect((await db.select().from(toolGatewaySessions)).map(row => row.id)).toEqual([future.id]));
await settle(5 * 60_000); await quiet();
});
it("reconstructs saved token expiry on startup", async () => {
const due = new Date(Date.now() + 5 * 60_000);
await token(due);
await start(); await settle();
expect(coordinator!.nextWakeAt()).toBe(due.getTime());
await settle(5 * 60_000);
expect(await db.select().from(toolMcpGatewayTokens)).toEqual([]);
await quiet();
});
it("drains an overdue persisted backlog across bounded batches", async () => {
await db.insert(toolMcpGatewayTokens).values(Array.from({ length: 501 }, (_, index) => ({
companyId, gatewayId: namedId, name: `Expired ${index}`, tokenHash: randomUUID(),
tokenPrefix: "expired", expiresAt: new Date(Date.now() - 1),
})));
await start();
expect(coordinator!.nextWakeAt()).not.toBeNull();
await quiet();
expect(await db.select().from(toolMcpGatewayTokens)).toEqual([]);
});
it("retains cleanup through drain without permitting expired authentication", async () => {
await start(); await quiet(); draining = true;
const active = await session();
const expiredToken = await token(new Date(Date.now() + 3 * 60_000));
await settle();
const reads = vi.spyOn(db, "execute"), selects = vi.spyOn(db, "select");
await settle(10 * 60_000);
expect(reads).not.toHaveBeenCalled(); expect(selects).not.toHaveBeenCalled();
reads.mockRestore(); selects.mockRestore();
expect(await db.select().from(toolGatewaySessions)).toHaveLength(1);
await expect(gateway.listToolsForSession(active.token)).rejects.toMatchObject({ status: 401, reasonCode: "session_expired" });
await expect(gateway.initializeNamedGatewayProtocol({ gatewayId: namedId, bearerToken: expiredToken.token }))
.rejects.toMatchObject({ status: 401 });
draining = false; await settle();
expect(await db.select().from(toolGatewaySessions)).toEqual([]);
expect(await db.select().from(toolMcpGatewayTokens)).toEqual([]);
await quiet();
});
});
@@ -71,6 +71,7 @@ const {
toolGateway: {
sweepActionReviews: vi.fn(async () => ({ scanned: 0 })),
cleanupExpiredSessions: vi.fn(async () => ({ deletedCount: 0 })),
nextCredentialExpiryAt: vi.fn(async () => null),
},
toolActionDeliveries: { hasPending: vi.fn(async () => false), sweepPending: vi.fn(async () => ({ scanned: 0, delivered: 0 })) },
},
+6 -4
View File
@@ -1,3 +1,4 @@
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { createHash, randomUUID } from "node:crypto";
import { createServer, type IncomingMessage } from "node:http";
import express from "express";
@@ -65,8 +66,8 @@ import {
startEmbeddedPostgresTestDatabase,
} from "./helpers/embedded-postgres.js";
const externalTestDatabaseUrl = process.env.PAPERCLIP_TOOL_GATEWAY_TEST_DATABASE_URL;
const embeddedPostgresSupport = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
const externalTestDatabase = resolveExternalTestDatabaseUrl("PAPERCLIP_TOOL_GATEWAY_TEST_DATABASE_URL");
const embeddedPostgresSupport = externalTestDatabase ? { supported: true } : await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
const testToolActionSigningSecret = "test-tool-action-signing-secret";
@@ -554,8 +555,9 @@ describeEmbeddedPostgres("tool gateway acceptance", () => {
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
beforeAll(async () => {
if (!externalTestDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-tool-gateway-");
db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString);
if (externalTestDatabase) { db = createDb(externalTestDatabase); return; }
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-tool-gateway-");
db = createDb(tempDb.connectionString);
}, 90_000);
afterEach(async () => {
+5 -2
View File
@@ -1,3 +1,4 @@
import { registerGatewayCredentialExpiryWork } from "./services/gateway-credential-expiry-work.js";
import { registerSecretProposalExpiryWork } from "./services/secret-proposal-expiry-work.js";
import { registerCustomImageCleanupWork } from "./services/custom-image-cleanup-work.js";
import { registerWorkspaceRecoveryWork } from "./services/workspace-recovery-work.js";
@@ -1464,7 +1465,10 @@ async function startServerWithDatabaseTeardown(
};
await timeStartupRecovery("tool_reviews", () => app.locals.toolGateway.sweepActionReviews()).catch((err: unknown) => logger.error({ err }, "startup tool review recovery failed"));
await timeStartupRecovery("gateway_sessions", () => app.locals.toolGateway.cleanupExpiredSessions()).catch((err: unknown) => logger.error({ err }, "startup gateway token cleanup failed"));
await timeStartupRecovery("gateway_sessions", () => config.heartbeatSchedulerEnabled
? registerGatewayCredentialExpiryWork(deliveryWork, app.locals.toolGateway,
() => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready
: app.locals.toolGateway.cleanupExpiredSessions()).catch((err: unknown) => logger.error({ err }, "startup gateway token cleanup failed"));
await timeStartupRecovery("delivery_workers", () => Promise.all([budgetEnforcementWorker, accountingWorker, fastResponseWorker, ...deliveryWorkers].map(worker => worker.ready)));
scheduleGitHubConnectionEventPoll();
scheduleGitHubConnectionContinuitySweep();
@@ -1891,7 +1895,6 @@ async function startServerWithDatabaseTeardown(
}));
trackHeartbeatSchedulerWork(app.locals.toolGateway.sweepActionReviews().catch((err: unknown) => logger.error({ err }, "tool review recovery failed")));
trackHeartbeatSchedulerWork(app.locals.toolGateway.cleanupExpiredSessions().catch((err: unknown) => logger.error({ err }, "gateway token cleanup failed")));
if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
if (!(await heartbeat.resolveSchedulingSuppression()).suppressed) {
+8
View File
@@ -9,11 +9,14 @@ export function registerDeadlineWork(
next: () => Promise<number | null>;
canRun: () => boolean;
retryMs: number;
// Optional batching latency for notifications; already-known deadlines still preempt it.
notificationCoalesceMs?: number;
},
) {
let requested = 1, reconciled = 0;
let next: number | null = null;
let earliest = 0;
let coalesceUntil = 0;
return coordinator.register(queue, {
retryMs: task.retryMs,
pausePassiveMaintenance: true,
@@ -21,6 +24,10 @@ export function registerDeadlineWork(
run: async () => {
if (!task.canRun()) { requested++; return; }
if (Date.now() < earliest) { next = earliest; return; }
if (Date.now() < coalesceUntil && (next === null || next > Date.now())) {
next = next === null ? coalesceUntil : Math.min(next, coalesceUntil);
return;
}
const generation = ++requested;
earliest = Date.now() + task.retryMs;
await task.run();
@@ -28,6 +35,7 @@ export function registerDeadlineWork(
// Back off failed or still-overdue work, not a successfully read future deadline.
if (deadline === null || deadline > Date.now()) earliest = 0;
next = deadline === null ? null : Math.max(earliest, deadline);
coalesceUntil = Date.now() + (task.notificationCoalesceMs ?? 0);
reconciled = generation;
},
hasPending: async () => requested !== reconciled,
@@ -9,6 +9,7 @@ export const DELIVERY_QUEUES = {
executionStatus: "execution-status",
executionDisposition: "execution-disposition",
environmentCleanup: "environment-cleanup",
gatewayCredentialExpiry: "gateway-credential-expiry",
secretProposalExpiry: "secret-proposal-expiry",
customImageExpiry: "custom-image-expiry",
historicalCleanup: "historical-cleanup",
@@ -0,0 +1,18 @@
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
import { registerDeadlineWork } from "./deadline-work.js";
import type { createToolGatewayService } from "./tool-gateway.js";
export function registerGatewayCredentialExpiryWork(
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
gateway: Pick<ReturnType<typeof createToolGatewayService>, "cleanupExpiredSessions" | "nextCredentialExpiryAt">,
canRun: () => boolean,
) {
return registerDeadlineWork(coordinator, DELIVERY_QUEUES.gatewayCredentialExpiry, {
retryMs: 30_000,
notificationCoalesceMs: 30_000,
canRun,
run: () => gateway.cleanupExpiredSessions({ singleBatch: true }),
next: () => gateway.nextCredentialExpiryAt(),
});
}
+97 -64
View File
@@ -9210,45 +9210,48 @@ export function createToolGatewayService(
const token = generateNamedGatewayToken(tokenId);
const tokenPrefix = `pcgw_${tokenId.slice(0, 8)}`;
const now = new Date();
const [row] = await db
.insert(toolMcpGatewayTokens)
.values({
id: tokenId,
companyId: input.companyId,
gatewayId: input.gatewayId,
name: input.body.name,
tokenHash: hashGatewayToken(token),
tokenPrefix,
subjectType: input.body.subjectType ?? "gateway_client",
subjectId: input.body.subjectId ?? null,
clientLabel: input.body.clientLabel,
ownerNote: input.body.ownerNote,
allowedActions: input.body.allowedActions ?? [
"tools/list",
"tools/call",
"resources/list",
"resources/read",
"prompts/list",
"prompts/get",
],
expiresAt: input.body.expiresAt ?? null,
expiryOverrideReason: input.body.expiryOverrideReason ?? null,
expiryOverrideByAgentId:
input.actor?.agentId && input.body.expiryOverrideReason
? input.actor.agentId
: null,
expiryOverrideByUserId:
input.actor?.userId && input.body.expiryOverrideReason
? input.actor.userId
: null,
expiryOverrideAt: input.body.expiryOverrideReason ? now : null,
responsibleUserId: input.managementContext?.actor.type === "agent" ? input.managementContext.responsibleUserId : null,
createdByAgentId: input.actor?.agentId ?? null,
createdByUserId: input.actor?.userId ?? null,
createdAt: now,
updatedAt: now,
})
.returning();
const [row] = await db.transaction(async tx => {
if (input.body.expiresAt) await notifyDeliveryWork(tx, DELIVERY_QUEUES.gatewayCredentialExpiry);
return tx
.insert(toolMcpGatewayTokens)
.values({
id: tokenId,
companyId: input.companyId,
gatewayId: input.gatewayId,
name: input.body.name,
tokenHash: hashGatewayToken(token),
tokenPrefix,
subjectType: input.body.subjectType ?? "gateway_client",
subjectId: input.body.subjectId ?? null,
clientLabel: input.body.clientLabel,
ownerNote: input.body.ownerNote,
allowedActions: input.body.allowedActions ?? [
"tools/list",
"tools/call",
"resources/list",
"resources/read",
"prompts/list",
"prompts/get",
],
expiresAt: input.body.expiresAt ?? null,
expiryOverrideReason: input.body.expiryOverrideReason ?? null,
expiryOverrideByAgentId:
input.actor?.agentId && input.body.expiryOverrideReason
? input.actor.agentId
: null,
expiryOverrideByUserId:
input.actor?.userId && input.body.expiryOverrideReason
? input.actor.userId
: null,
expiryOverrideAt: input.body.expiryOverrideReason ? now : null,
responsibleUserId: input.managementContext?.actor.type === "agent" ? input.managementContext.responsibleUserId : null,
createdByAgentId: input.actor?.agentId ?? null,
createdByUserId: input.actor?.userId ?? null,
createdAt: now,
updatedAt: now,
})
.returning();
});
if (input.managementContext) await logActivity(db, { companyId: input.companyId, ...input.managementContext.audit,
action: "tool_gateway.token_created", entityType: "tool_mcp_gateway", entityId: input.gatewayId,
details: { tokenId: row.id, subjectType: row.subjectType, allowedActions: row.allowedActions } });
@@ -9373,18 +9376,21 @@ export function createToolGatewayService(
expiresAt: new Date(now.getTime() + sessionTtlMs(input.ttlMs)),
};
await db.insert(toolGatewaySessions).values({
id: session.id,
companyId: session.companyId,
agentId: input.agentId,
runId: input.runId,
issueId: session.issueId,
projectId: session.projectId,
tokenHash: hashGatewayToken(token),
expiresAt: session.expiresAt,
createdAt: session.createdAt,
updatedAt: session.createdAt,
} as any);
await db.transaction(async tx => {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.gatewayCredentialExpiry);
await tx.insert(toolGatewaySessions).values({
id: session.id,
companyId: session.companyId,
agentId: input.agentId,
runId: input.runId,
issueId: session.issueId,
projectId: session.projectId,
tokenHash: hashGatewayToken(token),
expiresAt: session.expiresAt,
createdAt: session.createdAt,
updatedAt: session.createdAt,
} as any);
});
await writeAudit({
session,
@@ -11860,23 +11866,50 @@ export function createToolGatewayService(
return { ...sessionView, revokedAt: session!.revokedAt ?? now };
},
async cleanupExpiredSessions(input: { now?: Date } = {}) {
async nextCredentialExpiryAt(): Promise<number | null> {
const [row] = await db.execute<{ at: string | Date | null }>(sql`
select min(expires_at) as at from (
(select ${toolMcpGatewayTokens.expiresAt} as expires_at from ${toolMcpGatewayTokens}
where ${toolMcpGatewayTokens.expiresAt} is not null
order by ${toolMcpGatewayTokens.expiresAt} limit 1)
union all
(select ${toolGatewaySessions.expiresAt} as expires_at from ${toolGatewaySessions}
order by ${toolGatewaySessions.expiresAt} limit 1)
) expiries
`);
return row?.at == null ? null : new Date(row.at).getTime();
},
async cleanupExpiredSessions(input: { now?: Date; singleBatch?: boolean } = {}) {
const now = input.now ?? new Date();
const expiredTokens = await db.select({ id: toolMcpGatewayTokens.id })
.from(toolMcpGatewayTokens)
.where(lte(toolMcpGatewayTokens.expiresAt, now))
.orderBy(asc(toolMcpGatewayTokens.expiresAt), asc(toolMcpGatewayTokens.id))
.limit(500);
if (expiredTokens.length > 0) {
await db.delete(toolMcpGatewayTokens)
let deletedCount = 0;
while (true) {
const expiredTokens = await db.select({ id: toolMcpGatewayTokens.id })
.from(toolMcpGatewayTokens)
.where(lte(toolMcpGatewayTokens.expiresAt, now))
.orderBy(asc(toolMcpGatewayTokens.expiresAt), asc(toolMcpGatewayTokens.id))
.limit(500);
if (expiredTokens.length > 0) {
await db.delete(toolMcpGatewayTokens)
.where(and(inArray(toolMcpGatewayTokens.id, expiredTokens.map((token) => token.id)), lte(toolMcpGatewayTokens.expiresAt, now)))
.returning({ id: toolMcpGatewayTokens.id });
}
const expiredSessions = await db.select({ id: toolGatewaySessions.id })
.from(toolGatewaySessions)
.where(lte(toolGatewaySessions.expiresAt, now))
.orderBy(asc(toolGatewaySessions.expiresAt), asc(toolGatewaySessions.id))
.limit(500);
if (expiredSessions.length > 0) {
const rows = await db.delete(toolGatewaySessions)
.where(and(inArray(toolGatewaySessions.id, expiredSessions.map((session) => session.id)), lte(toolGatewaySessions.expiresAt, now)))
.returning({ id: toolGatewaySessions.id });
deletedCount += rows.length;
}
// The worker schedules the next page; one-shot startup/manual callers
// must finish the backlog without relying on another scheduled pass.
if (input.singleBatch || (expiredTokens.length < 500 && expiredSessions.length < 500)) break;
}
const rows = await db
.delete(toolGatewaySessions)
.where(lte(toolGatewaySessions.expiresAt, now))
.returning({ id: toolGatewaySessions.id });
return { deletedCount: rows.length };
return { deletedCount };
},
async listRuntimeSlots(companyId?: string) {