mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 19:28:46 +01:00
feat(server): bind idle credential eligibility to incoming wake proof (#18639)
Reviewed and verified at d1e3afa4af.
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
0b87308d5d
commit
5877d751c0
31 files changed
+2071
-53
No files matched your search
@@ -11,6 +11,7 @@ import { retainGitHubReviewReceipt } from "../services/chat-channels/github/task
|
||||
import { settleUnrecoverableExecutions } from "../services/execution-recovery-resolution.js";
|
||||
import { recordRunPresentationDecision } from "../services/heartbeat/run-completion.js";
|
||||
import { environmentService } from "../services/environments.js";
|
||||
import { reconcilePersistedRuntimeServicesOnStartup, readWorkspaceRuntimeIdleWork } from "../services/workspace-runtime.js";
|
||||
import { createAdmissionTransactionScope, createWakeAdmissionWriter } from "../modules/wake-queue/adapters/postgres.js";
|
||||
import { chatFailedRetryNoticeWork } from "../services/chat-failed-retry-notice-work.js";
|
||||
import { issueTreeControlService } from "../services/issue-tree-control.js";
|
||||
@@ -33,6 +34,10 @@ import * as githubTaskPreparation from "../services/chat-channels/github/task-pr
|
||||
import { budgetPolicies } from "@paperclipai/db";
|
||||
import { budgetService } from "../services/budgets.js";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
import { readIdleLocalWork } from "../services/idle-local-work.js";
|
||||
import { incomingWakeWorkQuery } from "../services/idle-incoming-wake.js";
|
||||
import { createChatRuntime } from "../services/chat-runtime.js";
|
||||
import { createChatSdkRuntime } from "../services/chat-sdk-runtime.js";
|
||||
import { readFreshWorkspaceReadiness } from "../services/fresh-workspace-continuation.js";
|
||||
import { requestTaskRetry, reconcileTaskChatRetryReceipts } from "../services/task-retry.js";
|
||||
import { taskInterruptionPresentation } from "../services/task-interruption-presentation.js";
|
||||
@@ -809,6 +814,8 @@ class FakeChatSdkRuntime {
|
||||
return this.endpoints.get(endpointId) ?? null;
|
||||
}
|
||||
|
||||
hasUnsettledLifecycle() { return false; }
|
||||
|
||||
async replaceEndpoint(options: CreateChatSdkEndpointRuntimeOptions) {
|
||||
this.replaceCount += 1;
|
||||
this.configurations.set(options.endpointId, options);
|
||||
@@ -1128,8 +1135,20 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
// Do not expire or time out its connections on those virtual clock jumps.
|
||||
db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString, {
|
||||
idleTimeoutSeconds: 0, connectTimeoutSeconds: 0, maxLifetimeSeconds: 0,
|
||||
...(historyTestDatabaseUrl ? { maxConnections: 4 } : {}),
|
||||
...(externalTestDatabaseUrl ? { maxConnections: 4 } : {}),
|
||||
});
|
||||
// The composed local-probe inventory includes the workspace owner. Match
|
||||
// server startup on this fresh fixture; never adopt or start a leftover.
|
||||
const [runtimeState] = await db.execute<{ configured: boolean }>(sql`SELECT
|
||||
EXISTS (SELECT 1 FROM workspace_runtime_services)
|
||||
OR EXISTS (SELECT 1 FROM project_workspaces)
|
||||
OR EXISTS (SELECT 1 FROM execution_workspaces) AS configured`);
|
||||
expect(runtimeState.configured).toBe(false);
|
||||
expect(await reconcilePersistedRuntimeServicesOnStartup(db)).toEqual({
|
||||
reconciled: 0, adopted: 0, stopped: 0, backfilled: 0, restarted: 0,
|
||||
restartFailed: 0, exposureReservationDrift: [],
|
||||
});
|
||||
expect(readWorkspaceRuntimeIdleWork()).toBe("none");
|
||||
}, 90_000);
|
||||
|
||||
afterAll(async () => {
|
||||
@@ -4185,10 +4204,14 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
.toEqual({ next: null, active: 0 });
|
||||
expect(f.responseWrites).toHaveLength(baseline);
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
|
||||
const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
|
||||
const readStacks: string[] = [], select = db.select;
|
||||
const reads = vi.spyOn(db, "select").mockImplementation((...args) => {
|
||||
if (readStacks.length < 5) readStacks.push(new Error("Unexpected idle database read").stack ?? "Stack unavailable");
|
||||
return Reflect.apply(select, db, args);
|
||||
}), writes = vi.spyOn(db, "transaction");
|
||||
try {
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
|
||||
expect(reads, readStacks.join("\n\n")).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
|
||||
expect(f.responseWrites).toHaveLength(baseline);
|
||||
} finally { reads.mockRestore(); writes.mockRestore(); vi.useRealTimers(); }
|
||||
const settlePresentation = async (tx: Parameters<Parameters<Db["transaction"]>[0]>[0]) => {
|
||||
@@ -29957,13 +29980,23 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
}),
|
||||
);
|
||||
await initializationEntered;
|
||||
expect(runtime.hasUnsettledLifecycle()).toBe(false); // Fake registry tail is already settled.
|
||||
expect(await registeredIdleProbeState()).toBe("present");
|
||||
await service.configure(endpoint.id, { action: "pause" }, "owner-user");
|
||||
let shutdownSettled = false;
|
||||
const shutdown = service.shutdown().then(() => { shutdownSettled = true; });
|
||||
await new Promise<void>(resolve => setImmediate(resolve));
|
||||
expect(shutdownSettled).toBe(false);
|
||||
expect(await registeredIdleProbeState()).toBe("present");
|
||||
releaseInitialization();
|
||||
|
||||
await expect(initializingWebhook).rejects.toMatchObject({
|
||||
status: 409,
|
||||
details: { code: "chat_endpoint_runtime_superseded" },
|
||||
});
|
||||
await shutdown;
|
||||
expect(shutdownSettled).toBe(true);
|
||||
expect(await registeredIdleProbeState()).toBe("none");
|
||||
expect(runtime.endpoints.has(endpoint.id)).toBe(false);
|
||||
await expect(service.get(endpoint.id)).resolves.toMatchObject({
|
||||
status: "paused",
|
||||
@@ -49515,7 +49548,12 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
.then((rows) => rows.length);
|
||||
expect(deliveryCount).toBeGreaterThan(0);
|
||||
|
||||
const endpointBlocksSleep = async () => (await db.execute(sql`
|
||||
SELECT EXISTS (${incomingWakeWorkQuery("chat_endpoints")} AND endpoint.id = ${endpoint.id}::uuid) AS blocked`))[0].blocked;
|
||||
if (provider === "github") expect(await endpointBlocksSleep()).toBe(true);
|
||||
await service.configure(endpoint.id, { action: "remove" }, "owner-user");
|
||||
if (provider === "github") expect(await endpointBlocksSleep()).toBe(false);
|
||||
else expect(await endpointBlocksSleep()).toBe(true); // No persistent/provider coverage expansion.
|
||||
|
||||
await expect(service.get(endpoint.id)).resolves.toMatchObject({
|
||||
status: "archived",
|
||||
@@ -49557,6 +49595,44 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
},
|
||||
);
|
||||
|
||||
async function registeredIdleProbeState() {
|
||||
let state: string | undefined;
|
||||
await readIdleLocalWork({ diagnostic: (reason, value) => { if (reason === "local.registered_probes") state = value; } });
|
||||
return state;
|
||||
}
|
||||
|
||||
it.each([
|
||||
{ name: "SDK", create: createChatSdkRuntime },
|
||||
{ name: "shared", create: createChatRuntime },
|
||||
])("retains the service idle probe after failed $name retirement and shutdown until owner retry", async ({ create }) => {
|
||||
const fixture = await seedCompany();
|
||||
const f = await configuredGitHubEndpoint(fixture);
|
||||
const registry = create();
|
||||
const first = await registry.replaceEndpoint(f.runtime.configurations.get(f.endpoint.id)!);
|
||||
const shutdown = vi.spyOn(first, "shutdown")
|
||||
.mockRejectedValueOnce(new Error("remove retirement uncertain"))
|
||||
.mockRejectedValueOnce(new Error("shutdown retirement uncertain"))
|
||||
.mockResolvedValue(undefined);
|
||||
const service = chatChannelService(db, { runtime: registry, fetch: f.providerFetch, publicBaseUrl: "https://paperclip.example" });
|
||||
fixtureServices.add(service);
|
||||
expect(await registeredIdleProbeState()).toBe("none");
|
||||
await service.configure(f.endpoint.id, { action: "remove" }, "owner-user");
|
||||
expect(registry.get(f.endpoint.id)).toBeNull();
|
||||
expect(registry.list()).toEqual([]);
|
||||
expect(registry.hasUnsettledLifecycle()).toBe(true);
|
||||
const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction");
|
||||
try {
|
||||
expect(await registeredIdleProbeState()).toBe("present");
|
||||
expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled();
|
||||
} finally { select.mockRestore(); transaction.mockRestore(); }
|
||||
await expect(service.shutdown()).rejects.toThrow("shutdown retirement uncertain");
|
||||
expect(await registeredIdleProbeState()).toBe("present");
|
||||
await service.shutdown();
|
||||
expect(shutdown).toHaveBeenCalledTimes(3);
|
||||
expect(registry.hasUnsettledLifecycle()).toBe(false);
|
||||
expect(await registeredIdleProbeState()).toBe("none");
|
||||
});
|
||||
|
||||
it("retries credential cleanup after archival made a remove partially complete", async () => {
|
||||
const fixture = await seedCompany();
|
||||
const { endpoint, service } = await configuredSlackEndpoint(fixture);
|
||||
@@ -60587,7 +60663,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
const commentsBefore = await db
|
||||
.select({ id: issueComments.id })
|
||||
.from(issueComments)
|
||||
.where(eq(issueComments.issueId, conversation.issueId));
|
||||
.where(eq(issueComments.issueId, conversation.issueId))
|
||||
.orderBy(asc(issueComments.id));
|
||||
const wakesBefore = wakeup.mock.calls.length;
|
||||
const reactionsBefore = providerRuntime.reactions.length;
|
||||
const postsBefore = providerRuntime.posts.length;
|
||||
@@ -60617,7 +60694,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
await db
|
||||
.select({ id: issueComments.id })
|
||||
.from(issueComments)
|
||||
.where(eq(issueComments.issueId, conversation.issueId)),
|
||||
.where(eq(issueComments.issueId, conversation.issueId))
|
||||
.orderBy(asc(issueComments.id)),
|
||||
).toEqual(commentsBefore);
|
||||
expect(wakeup).toHaveBeenCalledTimes(wakesBefore);
|
||||
expect(cancelRun).not.toHaveBeenCalled();
|
||||
|
||||
@@ -14,7 +14,7 @@ function createService(db: Db) {
|
||||
const runtimeShutdown = vi.fn().mockResolvedValue(undefined);
|
||||
const service = chatChannelService(db, {
|
||||
heartbeat: { wakeup },
|
||||
runtime: { shutdown: runtimeShutdown } as unknown as ChatChannelServiceOptions["runtime"],
|
||||
runtime: { shutdown: runtimeShutdown, hasUnsettledLifecycle: () => false } as unknown as ChatChannelServiceOptions["runtime"],
|
||||
});
|
||||
services.add(service);
|
||||
return { service, wakeup, runtimeShutdown };
|
||||
|
||||
@@ -0,0 +1,700 @@
|
||||
import { matchesApiWakeAdmission } from "../services/idle-api-wake-admission.js";
|
||||
import { generateKeyPairSync, randomUUID, sign } 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 {
|
||||
agents,
|
||||
agentApiKeys,
|
||||
authUsers,
|
||||
boardApiKeys,
|
||||
companies,
|
||||
companyMemberships,
|
||||
createDb,
|
||||
toolApplications,
|
||||
toolConnections,
|
||||
connectionGrants,
|
||||
chatEndpoints,
|
||||
chatGitHubRegistrations,
|
||||
chatSlackRegistrations,
|
||||
chatEndpointLeases,
|
||||
chatActions,
|
||||
chatDeliveries,
|
||||
chatPublications,
|
||||
chatConversations,
|
||||
issues,
|
||||
type Db,
|
||||
} from "@paperclipai/db";
|
||||
import { setCloudRuntimeConfigSnapshot } from "../services/cloud-runtime-config-snapshot.js";
|
||||
import { actorMiddleware } from "../middleware/auth.js";
|
||||
import {
|
||||
idleAdmissionMiddleware,
|
||||
trackIdleRequestHandlers,
|
||||
} from "../middleware/idle-admission.js";
|
||||
import { errorHandler } from "../middleware/error-handler.js";
|
||||
import { companyRoutes } from "../routes/companies.js";
|
||||
import { agentService } from "../services/agents.js";
|
||||
import { boardAuthService } from "../services/board-auth.js";
|
||||
import { readIdleSleepSafety } from "../services/idle-sleep-safety.js";
|
||||
import {
|
||||
readIdleIncomingWakeProof,
|
||||
incomingWakeFingerprint,
|
||||
incomingWakeWorkQuery,
|
||||
} from "../services/idle-incoming-wake.js";
|
||||
import { registerConnectionEventWake } from "../services/connection-event-wake.js";
|
||||
import { idleApiActivity } from "../services/idle-api-activity.js";
|
||||
import { startTaskDrain, stopTaskDrain } from "../services/task-admission.js";
|
||||
import {
|
||||
verifyCloudControlAssertion,
|
||||
CLOUD_CONTROL_AUDIENCE,
|
||||
CLOUD_CONTROL_JWS_TYPE,
|
||||
CLOUD_RUNTIME_IDENTITY_ISSUER,
|
||||
} from "../services/cloud-runtime-identity.js";
|
||||
import type {
|
||||
IdleIncomingWakeCapability,
|
||||
VerifiedIdleCapabilities,
|
||||
} from "../services/idle-incoming-wake-capability.js";
|
||||
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import {
|
||||
getEmbeddedPostgresTestSupport,
|
||||
startEmbeddedPostgresTestDatabase,
|
||||
} from "./helpers/embedded-postgres.js";
|
||||
const url = resolveExternalTestDatabaseUrl(
|
||||
"PAPERCLIP_INCOMING_WAKE_TEST_DATABASE_URL",
|
||||
);
|
||||
const support = url
|
||||
? { supported: true }
|
||||
: await getEmbeddedPostgresTestSupport();
|
||||
const ownerId = "00000000-0000-4000-8000-000000000001";
|
||||
const now = Date.now();
|
||||
const hold = () => ({
|
||||
ownerId,
|
||||
draining: true,
|
||||
startedAt: new Date(now - 1000),
|
||||
expiresAt: new Date(now + 300_000),
|
||||
activeRuns: 0,
|
||||
pendingWakes: 0,
|
||||
});
|
||||
const capability: IdleIncomingWakeCapability = {
|
||||
version: 1,
|
||||
ownerId,
|
||||
generation: "a".repeat(64),
|
||||
apiKeyProof: { version: 1 as const, challenge: "c".repeat(64) },
|
||||
routes: ["tenant-api-http-v2"],
|
||||
};
|
||||
function verified(incomingWake = capability): VerifiedIdleCapabilities {
|
||||
const pair = generateKeyPairSync("ed25519"),
|
||||
kid = randomUUID(),
|
||||
issued = Math.floor(Date.now() / 1000);
|
||||
const encode = (v: unknown) =>
|
||||
Buffer.from(JSON.stringify(v)).toString("base64url");
|
||||
const header = encode({ alg: "EdDSA", typ: CLOUD_CONTROL_JWS_TYPE, kid });
|
||||
const payload = encode({
|
||||
v: 1,
|
||||
iss: CLOUD_RUNTIME_IDENTITY_ISSUER,
|
||||
aud: CLOUD_CONTROL_AUDIENCE,
|
||||
sub: "fixture-stack",
|
||||
action: "task-drain:read",
|
||||
requestId: randomUUID(),
|
||||
iat: issued,
|
||||
exp: issued + 60,
|
||||
idleCapabilities: { incomingWake },
|
||||
});
|
||||
const token = `${header}.${payload}.${sign(null, Buffer.from(`${header}.${payload}`), pair.privateKey).toString("base64url")}`;
|
||||
return verifyCloudControlAssertion({
|
||||
compactJws: token,
|
||||
expectedAction: "task-drain:read",
|
||||
env: {
|
||||
PAPERCLIP_CLOUD_STACK_ID: "fixture-stack",
|
||||
PAPERCLIP_CLOUD_RUNTIME_IDENTITY_JWKS: JSON.stringify({
|
||||
keys: [
|
||||
{
|
||||
...pair.publicKey.export({ format: "jwk" }),
|
||||
kid,
|
||||
use: "sig",
|
||||
alg: "EdDSA",
|
||||
},
|
||||
],
|
||||
}),
|
||||
},
|
||||
}).idleCapabilities as VerifiedIdleCapabilities;
|
||||
}
|
||||
|
||||
(support.supported ? describe : describe.skip)(
|
||||
"scoped incoming wake PostgreSQL contract",
|
||||
() => {
|
||||
let db: Db;
|
||||
let database:
|
||||
Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
beforeAll(async () => {
|
||||
if (!url)
|
||||
database = await startEmbeddedPostgresTestDatabase(
|
||||
"idle-incoming-wake-",
|
||||
);
|
||||
db = createDb(url ?? database!.connectionString, { maxConnections: 4 });
|
||||
await db.execute(sql`SELECT 1 FROM agent_api_keys LIMIT 1`); // Coordinator migrates the fresh fixture; no embedded fallback.
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN", "fixture-only");
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_STACK_ID", "fixture-stack");
|
||||
setCloudRuntimeConfigSnapshot({ version: 1, revision: 0, config: null });
|
||||
});
|
||||
afterAll(async () => {
|
||||
setCloudRuntimeConfigSnapshot(null);
|
||||
stopTaskDrain();
|
||||
vi.restoreAllMocks();
|
||||
vi.unstubAllEnvs();
|
||||
await db?.$client.end({ timeout: 5 });
|
||||
await database?.cleanup();
|
||||
});
|
||||
// The externally supplied database is dedicated to this suite. Preserve all
|
||||
// rows during every eligibility scenario, then isolate the next scenario.
|
||||
const resetFixture = async () => {
|
||||
await db.execute(
|
||||
sql`TRUNCATE companies, "user", chat_github_registrations CASCADE`,
|
||||
);
|
||||
};
|
||||
beforeEach(resetFixture);
|
||||
afterEach(async () => {
|
||||
stopTaskDrain();
|
||||
vi.restoreAllMocks();
|
||||
await resetFixture();
|
||||
});
|
||||
|
||||
const read = (incomingWake?: VerifiedIdleCapabilities) =>
|
||||
readIdleSleepSafety(
|
||||
db,
|
||||
hold,
|
||||
() => now,
|
||||
ownerId,
|
||||
async () => "none",
|
||||
undefined,
|
||||
undefined,
|
||||
false,
|
||||
false,
|
||||
{ incomingWake },
|
||||
);
|
||||
async function seed() {
|
||||
const companyId = randomUUID(),
|
||||
agentId = randomUUID(),
|
||||
userId = randomUUID(),
|
||||
foreignId = randomUUID();
|
||||
await db.insert(companies).values([
|
||||
{
|
||||
id: companyId,
|
||||
name: "Incoming fixture",
|
||||
issuePrefix: `I${companyId.slice(0, 6)}`,
|
||||
},
|
||||
{
|
||||
id: foreignId,
|
||||
name: "Foreign fixture",
|
||||
issuePrefix: `F${foreignId.slice(0, 6)}`,
|
||||
},
|
||||
]);
|
||||
await db
|
||||
.insert(authUsers)
|
||||
.values({
|
||||
id: userId,
|
||||
name: "Fixture 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 db
|
||||
.insert(agents)
|
||||
.values({
|
||||
id: agentId,
|
||||
companyId,
|
||||
name: "Fixture agent",
|
||||
role: "engineer",
|
||||
adapterType: "process",
|
||||
status: "idle",
|
||||
});
|
||||
const board = await boardAuthService(db).createNamedBoardApiKey({
|
||||
userId,
|
||||
name: "incoming board",
|
||||
expiresAt: null,
|
||||
});
|
||||
const agent = await agentService(db).createApiKey(
|
||||
agentId,
|
||||
"incoming agent",
|
||||
{ kind: "standard" },
|
||||
{ responsibleUserId: userId },
|
||||
);
|
||||
return { companyId, foreignId, userId, agentId, board, agent };
|
||||
}
|
||||
it("retains real saved keys and exempts them only for the verified exact owner/profile", async () => {
|
||||
const f = await seed();
|
||||
expect((await read()).backgroundWork).toBe("present");
|
||||
expect(await read(verified())).toMatchObject({
|
||||
version: 3,
|
||||
backgroundWork: "none",
|
||||
incomingWake: capability,
|
||||
});
|
||||
const receipt = (await read(verified())).apiWakeAdmission!;
|
||||
expect(receipt.keys).toHaveLength(2);
|
||||
expect(matchesApiWakeAdmission(receipt, { stackId: "fixture-stack", token: f.board.token, now })).toBe(true);
|
||||
expect(matchesApiWakeAdmission(receipt, { stackId: "fixture-stack", token: f.agent.token, now })).toBe(true);
|
||||
const serialized = JSON.stringify(receipt);
|
||||
expect(serialized).not.toContain(f.board.token);
|
||||
expect(serialized).not.toContain(f.agent.token);
|
||||
await db.update(boardApiKeys).set({ revokedAt: new Date() }).where(eq(boardApiKeys.id, f.board.id));
|
||||
const changedReceipt = (await read(verified())).apiWakeAdmission!;
|
||||
expect(changedReceipt.generation).not.toBe(receipt.generation);
|
||||
expect(matchesApiWakeAdmission(changedReceipt, { stackId: "fixture-stack", token: f.board.token, now })).toBe(false);
|
||||
expect(
|
||||
(await read(verified({ ...capability, ownerId: randomUUID() })))
|
||||
.backgroundWork,
|
||||
).toBe("unknown");
|
||||
expect(
|
||||
await db
|
||||
.select()
|
||||
.from(agentApiKeys)
|
||||
.where(eq(agentApiKeys.id, f.agent.id)),
|
||||
).toHaveLength(1);
|
||||
expect(
|
||||
await db
|
||||
.select()
|
||||
.from(boardApiKeys)
|
||||
.where(eq(boardApiKeys.id, f.board.id)),
|
||||
).toHaveLength(1);
|
||||
});
|
||||
it("delivers through actual tenant auth/routes, denies current revoked/expired/foreign authority and holds only successful API use", async () => {
|
||||
const f = await seed();
|
||||
let monotonic = 1;
|
||||
vi.spyOn(performance, "now").mockImplementation(() => monotonic);
|
||||
let release: (() => void) | undefined, arrived: (() => void) | undefined;
|
||||
const app = express();
|
||||
// Named Cloud wake boundary: preserve request until the simulated compute handoff finishes.
|
||||
app.use(async (_req, _res, next) => {
|
||||
if (release) {
|
||||
await new Promise<void>((resolve) => {
|
||||
release = resolve;
|
||||
arrived?.();
|
||||
});
|
||||
}
|
||||
next();
|
||||
});
|
||||
app.use(idleAdmissionMiddleware);
|
||||
app.use(express.json());
|
||||
app.use(
|
||||
actorMiddleware(db, {
|
||||
deploymentMode: "authenticated",
|
||||
resolveSession: async () => null,
|
||||
}),
|
||||
);
|
||||
app.use("/api/companies", companyRoutes(db));
|
||||
app.use(errorHandler);
|
||||
trackIdleRequestHandlers(app);
|
||||
expect(idleApiActivity.active()).toBe(false);
|
||||
expect(
|
||||
(
|
||||
await request(app)
|
||||
.get(`/api/companies/${f.companyId}`)
|
||||
.set("Authorization", "Bearer unknown")
|
||||
).status,
|
||||
).toBe(401);
|
||||
expect(idleApiActivity.active()).toBe(false);
|
||||
expect(
|
||||
(
|
||||
await request(app)
|
||||
.get(`/api/companies/${f.foreignId}`)
|
||||
.set("Authorization", `Bearer ${f.board.token}`)
|
||||
).status,
|
||||
).toBe(403);
|
||||
expect(idleApiActivity.active()).toBe(false);
|
||||
const own = await request(app)
|
||||
.patch(`/api/companies/${f.companyId}`)
|
||||
.set("Authorization", `Bearer ${f.board.token}`)
|
||||
.send({ name: "Persisted after wake" });
|
||||
expect(own.status).toBe(200);
|
||||
expect(idleApiActivity.active()).toBe(true);
|
||||
expect(
|
||||
(
|
||||
await db.select().from(companies).where(eq(companies.id, f.companyId))
|
||||
)[0]!.name,
|
||||
).toBe("Persisted after wake");
|
||||
monotonic += 300_000;
|
||||
expect(
|
||||
(
|
||||
await request(app)
|
||||
.get(`/api/companies/${f.companyId}`)
|
||||
.set("Authorization", `Bearer ${f.agent.token}`)
|
||||
).status,
|
||||
).toBe(200);
|
||||
monotonic += 300_001;
|
||||
expect(idleApiActivity.active()).toBe(true);
|
||||
monotonic += 299_999;
|
||||
expect(idleApiActivity.active()).toBe(false);
|
||||
const arrival = new Promise<void>((resolve) => {
|
||||
arrived = resolve;
|
||||
});
|
||||
release = () => {};
|
||||
const held = request(app)
|
||||
.patch(`/api/companies/${f.companyId}`)
|
||||
.set("Authorization", `Bearer ${f.board.token}`)
|
||||
.send({ name: "Must not commit" })
|
||||
.then((r) => r);
|
||||
await arrival;
|
||||
await db
|
||||
.update(boardApiKeys)
|
||||
.set({ revokedAt: new Date() })
|
||||
.where(eq(boardApiKeys.id, f.board.id));
|
||||
release!();
|
||||
release = undefined;
|
||||
expect((await held).status).toBe(401);
|
||||
expect(idleApiActivity.active()).toBe(false);
|
||||
const expired = await boardAuthService(db).createNamedBoardApiKey({
|
||||
userId: f.userId,
|
||||
name: "expired",
|
||||
expiresAt: new Date(0),
|
||||
});
|
||||
expect(
|
||||
(
|
||||
await request(app)
|
||||
.get(`/api/companies/${f.companyId}`)
|
||||
.set("Authorization", `Bearer ${expired.token}`)
|
||||
).status,
|
||||
).toBe(401);
|
||||
expect(idleApiActivity.active()).toBe(false);
|
||||
startTaskDrain({ purpose: "idle", ttlMs: 60_000 });
|
||||
expect(
|
||||
(
|
||||
await request(app)
|
||||
.patch(`/api/companies/${f.companyId}`)
|
||||
.set("Authorization", `Bearer ${f.agent.token}`)
|
||||
.send({ name: "Draining" })
|
||||
).status,
|
||||
).toBe(503);
|
||||
expect(idleApiActivity.active()).toBe(false);
|
||||
stopTaskDrain();
|
||||
expect(
|
||||
(
|
||||
await db.select().from(companies).where(eq(companies.id, f.companyId))
|
||||
)[0]!.name,
|
||||
).toBe("Persisted after wake");
|
||||
});
|
||||
// Fault injection below supplies only malformed/incomplete persisted states;
|
||||
// the positive archived transition is covered by the real configure(remove)
|
||||
// producer in chat-channels.integration.test.ts, without deleting history.
|
||||
async function removedGitHubFixture() {
|
||||
const f = await seed(), applicationId = randomUUID(), connectionId = randomUUID(), endpointId = randomUUID();
|
||||
await db.insert(toolApplications).values({ id: applicationId, companyId: f.companyId, name: "Removed GitHub", type: "chat" });
|
||||
await db.insert(toolConnections).values({ id: connectionId, companyId: f.companyId, applicationId,
|
||||
name: "Removed GitHub", uid: randomUUID(), transport: "chat_sdk", connectionPurpose: "channel",
|
||||
status: "archived", enabled: false, credentialSecretRefs: [] });
|
||||
await db.insert(chatEndpoints).values({ id: endpointId, companyId: f.companyId, connectionId,
|
||||
assignedAgentId: f.agentId, provider: "github", publicId: randomUUID(), status: "archived",
|
||||
archivedAt: new Date(), setup: { step: "complete", runtimeGeneration: 1 } });
|
||||
const blocked = async () => (await db.execute(sql`SELECT EXISTS (
|
||||
${incomingWakeWorkQuery("chat_endpoints")} AND endpoint.id = ${endpointId}::uuid) AS blocked`))[0].blocked;
|
||||
return { ...f, applicationId, connectionId, endpointId, blocked };
|
||||
}
|
||||
it.each(["stamp", "active", "paused", "revoked", "persistent_provider", "enabled_connection", "active_connection",
|
||||
"non_sdk_connection", "credentials", "malformed_credentials", "malformed_setup", "missing_generation", "fractional_generation",
|
||||
"lease", "expired_lease", "pending_registration", "unconsumed_registration", "wrong_registry"])(
|
||||
"keeps incomplete removed GitHub proof blocked: %s", async fault => {
|
||||
const f = await removedGitHubFixture();
|
||||
if (fault === "stamp") await db.update(chatEndpoints).set({ archivedAt: null }).where(eq(chatEndpoints.id, f.endpointId));
|
||||
if (["active", "paused", "revoked"].includes(fault)) await db.update(chatEndpoints)
|
||||
.set({ status: fault as "active" | "paused" | "revoked" }).where(eq(chatEndpoints.id, f.endpointId));
|
||||
if (fault === "persistent_provider") await db.update(chatEndpoints).set({ provider: "discord" }).where(eq(chatEndpoints.id, f.endpointId));
|
||||
if (fault === "enabled_connection") await db.update(toolConnections).set({ enabled: true }).where(eq(toolConnections.id, f.connectionId));
|
||||
if (fault === "active_connection") await db.update(toolConnections).set({ status: "active" }).where(eq(toolConnections.id, f.connectionId));
|
||||
if (fault === "non_sdk_connection") await db.update(toolConnections).set({ transport: "voice" }).where(eq(toolConnections.id, f.connectionId));
|
||||
if (fault === "credentials" || fault === "malformed_credentials") await db.execute(sql`UPDATE tool_connections
|
||||
SET credential_secret_refs = ${fault === "credentials" ? '[{"secretId":"retained"}]' : '{}'}::jsonb WHERE id = ${f.connectionId}::uuid`);
|
||||
if (fault === "malformed_setup") await db.execute(sql`UPDATE chat_endpoints SET setup = '"unknown"'::jsonb WHERE id = ${f.endpointId}::uuid`);
|
||||
if (fault === "missing_generation" || fault === "fractional_generation") await db.update(chatEndpoints)
|
||||
.set({ setup: fault === "missing_generation" ? { step: "complete" } : { step: "complete", runtimeGeneration: 1.5 } })
|
||||
.where(eq(chatEndpoints.id, f.endpointId));
|
||||
if (fault === "lease" || fault === "expired_lease") await db.insert(chatEndpointLeases).values({ companyId: f.companyId,
|
||||
endpointId: f.endpointId, leaseKey: "credentials", token: randomUUID(), expiresAt: new Date(fault === "lease" ? Date.now() + 60_000 : 0) });
|
||||
if (fault === "pending_registration" || fault === "unconsumed_registration") await db.insert(chatGitHubRegistrations).values({
|
||||
companyId: f.companyId, endpointId: f.endpointId, userId: f.userId, stateHash: randomUUID(), trustedOrigin: "https://tenant.example.test",
|
||||
status: fault === "pending_registration" ? "pending" : "completed", expiresAt: new Date(Date.now() + 60_000) });
|
||||
if (fault === "wrong_registry") await db.insert(chatSlackRegistrations).values({ endpointId: f.endpointId, companyId: f.companyId,
|
||||
requestId: randomUUID(), status: "uncertain", manifest: {}, manifestHash: "a".repeat(64), callbackUri: "https://tenant.example.test",
|
||||
createdByUserId: f.userId, secretIds: { clientSecret: randomUUID() } });
|
||||
expect(await f.blocked()).toBe(true);
|
||||
expect(await db.select({ id: chatEndpoints.id }).from(chatEndpoints).where(eq(chatEndpoints.id, f.endpointId))).toHaveLength(1);
|
||||
});
|
||||
it("retains tenant boundaries and rejects unfamiliar archived state at the schema boundary", async () => {
|
||||
const f = await removedGitHubFixture();
|
||||
await expect(db.execute(sql`UPDATE chat_endpoints SET connection_id = ${randomUUID()}::uuid WHERE id = ${f.endpointId}::uuid`)).rejects.toThrow();
|
||||
const foreignApplication = randomUUID(), foreignConnection = randomUUID();
|
||||
await db.insert(toolApplications).values({ id: foreignApplication, companyId: f.foreignId, name: "Foreign", type: "chat" });
|
||||
await db.insert(toolConnections).values({ id: foreignConnection, companyId: f.foreignId, applicationId: foreignApplication,
|
||||
name: "Foreign", uid: randomUUID(), transport: "chat_sdk", connectionPurpose: "channel", status: "archived", enabled: false });
|
||||
await expect(db.execute(sql`UPDATE chat_endpoints SET connection_id = ${foreignConnection}::uuid WHERE id = ${f.endpointId}::uuid`)).rejects.toThrow();
|
||||
await expect(db.execute(sql`UPDATE chat_endpoints SET status = 'unknown' WHERE id = ${f.endpointId}::uuid`)).rejects.toThrow();
|
||||
expect((await db.select().from(chatEndpoints).where(eq(chatEndpoints.id, f.endpointId)))[0].connectionId).toBe(f.connectionId);
|
||||
});
|
||||
it("does not let removed GitHub endpoint history discharge pending delivery, publication or action obligations", async () => {
|
||||
const f = await removedGitHubFixture();
|
||||
await db.insert(chatGitHubRegistrations).values({ companyId: f.companyId, endpointId: f.endpointId, userId: f.userId,
|
||||
stateHash: randomUUID(), trustedOrigin: "https://tenant.example.test", status: "completed", consumedAt: new Date(), expiresAt: new Date(0) });
|
||||
expect(await f.blocked()).toBe(false);
|
||||
const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Retained chat debt", status: "done" }).returning();
|
||||
const [conversation] = await db.insert(chatConversations).values({ companyId: f.companyId, endpointId: f.endpointId,
|
||||
issueId: issue.id, externalConversationId: "github:retained", externalLabel: "retained", state: "endpoint_removed" }).returning();
|
||||
await db.insert(chatDeliveries).values({ companyId: f.companyId, endpointId: f.endpointId, conversationId: conversation.id,
|
||||
providerEventId: "retained", deduplicationKey: "retained", eventKind: "message", normalizedEvent: {}, state: "retry", nextAttemptAt: new Date() });
|
||||
await db.insert(chatPublications).values({ companyId: f.companyId, endpointId: f.endpointId, conversationId: conversation.id,
|
||||
issueId: issue.id, idempotencyKey: "retained", payload: { text: "Retained" }, state: "retry", nextAttemptAt: new Date() });
|
||||
await db.insert(chatActions).values({ companyId: f.companyId, endpointId: f.endpointId, conversationId: conversation.id,
|
||||
kind: "provider_effect", providerActionId: "retained", payload: {}, status: "received" });
|
||||
const report = await readIdleSleepSafety(db, hold, () => now, ownerId, async () => "none", undefined, undefined,
|
||||
false, false, { incomingWake: verified(), collectDiagnostics: true });
|
||||
expect(report.backgroundWork).toBe("present");
|
||||
expect(report.diagnostics?.checks).toEqual(expect.arrayContaining([
|
||||
{ reason: "chat_endpoints", state: "none" }, { reason: "chat_deliveries", state: "present" },
|
||||
{ reason: "chat_publications", state: "present" }, { reason: "chat_actions", state: "present" },
|
||||
]));
|
||||
expect(await db.select().from(chatDeliveries)).toHaveLength(1);
|
||||
expect(await db.select().from(chatPublications)).toHaveLength(1);
|
||||
expect(await db.select().from(chatActions)).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("requires current broker instance and exact owner binding, preserving continuity and unsupported transport blockers", async () => {
|
||||
const f = await seed();
|
||||
const applicationId = randomUUID(),
|
||||
connectionId = randomUUID(),
|
||||
grantId = randomUUID();
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_INSTANCE_ID", "fixture-instance");
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_SIGN_PRIVATE_KEY", "fixture");
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_SEAL_PRIVATE_KEY", "fixture");
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_ENVIRONMENT", "test");
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_BASE_URL", "http://127.0.0.1");
|
||||
const botConnectionId = randomUUID(),
|
||||
endpointId = randomUUID(),
|
||||
registration = "b".repeat(43);
|
||||
const botApplication = randomUUID();
|
||||
await db
|
||||
.insert(toolApplications)
|
||||
.values({
|
||||
id: botApplication,
|
||||
companyId: f.companyId,
|
||||
name: "GitHub bot",
|
||||
type: "chat",
|
||||
});
|
||||
await db
|
||||
.insert(toolConnections)
|
||||
.values({
|
||||
id: botConnectionId,
|
||||
companyId: f.companyId,
|
||||
applicationId: botApplication,
|
||||
name: "GitHub bot",
|
||||
uid: randomUUID(),
|
||||
transport: "chat_sdk",
|
||||
connectionPurpose: "channel",
|
||||
status: "active",
|
||||
enabled: true,
|
||||
config: { provider: "github" },
|
||||
});
|
||||
await db
|
||||
.insert(chatEndpoints)
|
||||
.values({
|
||||
id: endpointId,
|
||||
companyId: f.companyId,
|
||||
connectionId: botConnectionId,
|
||||
assignedAgentId: f.agentId,
|
||||
provider: "github",
|
||||
publicId: "p".repeat(43),
|
||||
status: "active",
|
||||
activatedAt: new Date(),
|
||||
providerAccountId: "fixture-app",
|
||||
botExternalId: "fixture-bot",
|
||||
setup: {
|
||||
step: "complete",
|
||||
github: {
|
||||
stage: "test",
|
||||
registrationStatus: "completed",
|
||||
cloudRegistrationId: registration,
|
||||
},
|
||||
},
|
||||
});
|
||||
await db
|
||||
.insert(chatGitHubRegistrations)
|
||||
.values({
|
||||
companyId: f.companyId,
|
||||
endpointId,
|
||||
userId: f.userId,
|
||||
stateHash: randomUUID(),
|
||||
trustedOrigin: "https://tenant.example.test",
|
||||
status: "completed",
|
||||
expiresAt: new Date(0),
|
||||
handoff: {
|
||||
cloudId: registration,
|
||||
redemptionId: randomUUID(),
|
||||
returnState: randomUUID(),
|
||||
},
|
||||
});
|
||||
const botTuple = [
|
||||
`github-app:${registration}`,
|
||||
f.companyId,
|
||||
f.userId,
|
||||
"github.bot",
|
||||
`gateway:${registration}`,
|
||||
registration,
|
||||
registration,
|
||||
];
|
||||
const stop = registerConnectionEventWake(db, () => {});
|
||||
const botCap: IdleIncomingWakeCapability = {
|
||||
...capability,
|
||||
routes: ["github-broker-events-v1", "tenant-api-http-v2"],
|
||||
githubBroker: {
|
||||
instanceId: "fixture-instance",
|
||||
bindingsFingerprint: incomingWakeFingerprint([botTuple]),
|
||||
},
|
||||
};
|
||||
expect((await read(verified())).backgroundWork).toBe("present");
|
||||
expect(await read(verified(botCap))).toMatchObject({
|
||||
version: 3,
|
||||
backgroundWork: "none",
|
||||
});
|
||||
const botProof = await readIdleIncomingWakeProof(db, {
|
||||
capabilities: verified(botCap),
|
||||
ownerId,
|
||||
receiverOwner: db,
|
||||
});
|
||||
expect(botProof?.endpointIds).toEqual([endpointId]);
|
||||
expect(botProof?.connectionIds).toEqual([botConnectionId]);
|
||||
expect(
|
||||
await db
|
||||
.select()
|
||||
.from(chatEndpoints)
|
||||
.where(eq(chatEndpoints.id, endpointId)),
|
||||
).toHaveLength(1);
|
||||
|
||||
await db
|
||||
.insert(toolApplications)
|
||||
.values({
|
||||
id: applicationId,
|
||||
companyId: f.companyId,
|
||||
name: "GitHub",
|
||||
type: "mcp_http",
|
||||
});
|
||||
await db
|
||||
.insert(toolConnections)
|
||||
.values({
|
||||
id: connectionId,
|
||||
companyId: f.companyId,
|
||||
applicationId,
|
||||
name: "GitHub",
|
||||
uid: randomUUID(),
|
||||
transport: "mcp_remote",
|
||||
connectionPurpose: "tool",
|
||||
status: "active",
|
||||
enabled: true,
|
||||
config: {
|
||||
sourceTemplateKey: "github",
|
||||
oauth: { connectorProfile: "github.code" },
|
||||
},
|
||||
});
|
||||
const github = {
|
||||
userId: "42",
|
||||
login: "fixture",
|
||||
installationCount: 1,
|
||||
repositoryCount: 1,
|
||||
repositorySelection: "all" as const,
|
||||
installationIds: ["123"],
|
||||
installationOwnerLogins: ["fixture"],
|
||||
lastAccessRefreshAt: new Date().toISOString(),
|
||||
};
|
||||
await db
|
||||
.insert(connectionGrants)
|
||||
.values({
|
||||
id: grantId,
|
||||
companyId: f.companyId,
|
||||
connectionId,
|
||||
kind: "user",
|
||||
subjectUserId: f.userId,
|
||||
providerTenant: { github },
|
||||
status: "active",
|
||||
});
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_INSTANCE_ID", "fixture-instance");
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_SIGN_PRIVATE_KEY", "fixture");
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_SEAL_PRIVATE_KEY", "fixture");
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_ENVIRONMENT", "test");
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_BASE_URL", "http://127.0.0.1");
|
||||
const brokerCap: IdleIncomingWakeCapability = {
|
||||
...capability,
|
||||
routes: ["github-broker-events-v1", "tenant-api-http-v2"],
|
||||
githubBroker: {
|
||||
instanceId: "fixture-instance",
|
||||
bindingsFingerprint: incomingWakeFingerprint([
|
||||
botTuple,
|
||||
[
|
||||
`${grantId}_123`,
|
||||
f.companyId,
|
||||
f.userId,
|
||||
"github.code",
|
||||
"123",
|
||||
connectionId,
|
||||
grantId,
|
||||
],
|
||||
]),
|
||||
},
|
||||
};
|
||||
try {
|
||||
const proof = await readIdleIncomingWakeProof(db, {
|
||||
capabilities: verified(brokerCap),
|
||||
ownerId,
|
||||
receiverOwner: db,
|
||||
});
|
||||
expect(proof?.requiredContinuityConnectionIds).toEqual([connectionId]);
|
||||
expect(proof?.connectionIds).toEqual([botConnectionId]);
|
||||
expect((await read(verified(brokerCap))).backgroundWork).toBe(
|
||||
"present",
|
||||
); // transport alone never discharges real credential deadline
|
||||
await db
|
||||
.update(connectionGrants)
|
||||
.set({
|
||||
providerTenant: {
|
||||
github: {
|
||||
...github,
|
||||
bindingSync: { id: randomUUID(), removeInstallationIds: [] },
|
||||
},
|
||||
},
|
||||
})
|
||||
.where(eq(connectionGrants.id, grantId));
|
||||
expect((await read(verified(brokerCap))).backgroundWork).toBe(
|
||||
"unknown",
|
||||
);
|
||||
await db
|
||||
.update(connectionGrants)
|
||||
.set({ status: "revoked" })
|
||||
.where(eq(connectionGrants.id, grantId));
|
||||
expect((await read(verified(brokerCap))).backgroundWork).toBe(
|
||||
"unknown",
|
||||
);
|
||||
await db
|
||||
.update(toolConnections)
|
||||
.set({ transport: "local_stdio" })
|
||||
.where(eq(toolConnections.id, connectionId));
|
||||
expect((await read(verified())).backgroundWork).toBe("present");
|
||||
} finally {
|
||||
stop();
|
||||
}
|
||||
});
|
||||
},
|
||||
);
|
||||
@@ -1,3 +1,4 @@
|
||||
import { generateKeyPairSync, randomUUID, sign } from "node:crypto";
|
||||
import express from "express";
|
||||
import request from "supertest";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
@@ -81,21 +82,24 @@ function localAdminActor() {
|
||||
|
||||
describe("instance settings routes", () => {
|
||||
const routeModules = hoistModuleGraph(registerModuleMocks, async () => {
|
||||
const [{ errorHandler }, { instanceSettingsRoutes }] = await Promise.all([
|
||||
const [{ errorHandler }, { instanceSettingsRoutes }, { cloudControlMiddleware }, controlIdentity] = await Promise.all([
|
||||
vi.importActual<typeof import("../middleware/index.js")>("../middleware/index.js"),
|
||||
vi.importActual<typeof import("../routes/instance-settings.js")>("../routes/instance-settings.js"),
|
||||
vi.importActual<typeof import("../middleware/cloud-control.js")>("../middleware/cloud-control.js"),
|
||||
vi.importActual<typeof import("../services/cloud-runtime-identity.js")>("../services/cloud-runtime-identity.js"),
|
||||
]);
|
||||
return { errorHandler, instanceSettingsRoutes };
|
||||
return { errorHandler, instanceSettingsRoutes, cloudControlMiddleware, controlIdentity };
|
||||
});
|
||||
|
||||
function createApp(actor: any, pluginWorkers?: PluginWorkerManager, prepareBackup?: () => Promise<boolean>) {
|
||||
const { errorHandler, instanceSettingsRoutes } = routeModules.value;
|
||||
function createApp(actor: any, pluginWorkers?: PluginWorkerManager, prepareBackup?: () => Promise<boolean>, verifyControl = false) {
|
||||
const { errorHandler, instanceSettingsRoutes, cloudControlMiddleware } = routeModules.value;
|
||||
const app = express();
|
||||
app.use(express.json());
|
||||
app.use((req, _res, next) => {
|
||||
req.actor = actor;
|
||||
next();
|
||||
});
|
||||
if (verifyControl) app.use(cloudControlMiddleware());
|
||||
app.use("/api", instanceSettingsRoutes(mockDb as any, pluginWorkers, prepareBackup));
|
||||
app.use(errorHandler);
|
||||
return app;
|
||||
@@ -1101,7 +1105,7 @@ describe("instance settings routes", () => {
|
||||
await request(createApp(adminActor)).post("/api/instance/task-drain").send({ purpose: "idle", ttlMs: 60_000 }).expect(200);
|
||||
expect(mockHeartbeatService.computeTaskDrain).toHaveBeenCalledWith({ purpose: "idle", ttlMs: 60_000 });
|
||||
await request(createApp(adminActor)).get(`/api/instance/task-drain?idleSleepSafety=1&ownerId=${ownerId}`).expect(200);
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenLastCalledWith(mockDb, expect.any(Function), Date.now, ownerId, expect.any(Function), undefined, undefined, false, false);
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenLastCalledWith(mockDb, expect.any(Function), Date.now, ownerId, expect.any(Function), undefined, undefined, false, false, { incomingWake: undefined, deadlineCapability: undefined });
|
||||
});
|
||||
|
||||
it.each([{}, { ttlMs: null }, { ttlMs: 1 }, { ttlMs: 300_001 }])("rejects an unbounded or invalid idle TTL: %j", async body => {
|
||||
@@ -1128,7 +1132,7 @@ describe("instance settings routes", () => {
|
||||
const res = await request(createApp(adminActor)).get("/api/instance/task-drain?idleSleepSafety=1");
|
||||
expect(res.status).toBe(200);
|
||||
expect(res.body).toEqual({ ...idleStatus, idleSleepSafety: report });
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, undefined, expect.any(Function), undefined, undefined, false, false);
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, undefined, expect.any(Function), undefined, undefined, false, false, { incomingWake: undefined, deadlineCapability: undefined });
|
||||
expect(mockReadIdleSleepSafety.mock.calls[0][1]()).toEqual(idleStatus);
|
||||
});
|
||||
|
||||
@@ -1142,7 +1146,7 @@ describe("instance settings routes", () => {
|
||||
.get("/api/instance/task-drain?idleSleepSafety=1&idleSleepDiagnostics=1&ownerId=fixture-owner").expect(200);
|
||||
expect(response.body.idleSleepSafety.diagnostics.checks).toEqual([{ reason: "agent_api_keys", state: "present" }]);
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now,
|
||||
"fixture-owner", expect.any(Function), undefined, undefined, false, false, { collectDiagnostics: true });
|
||||
"fixture-owner", expect.any(Function), undefined, undefined, false, false, { incomingWake: undefined, deadlineCapability: undefined, collectDiagnostics: true });
|
||||
mockReadIdleSleepSafety.mockClear();
|
||||
for (const actor of [nonAdminActor, { type: "agent", agentId: "agent-1", companyId: "company-1", source: "api_key" }]) {
|
||||
await request(createApp(actor))
|
||||
@@ -1164,7 +1168,93 @@ describe("instance settings routes", () => {
|
||||
await request(createApp({ ...adminActor, source: "cloud_control" }))
|
||||
.get(`/api/instance/task-drain?idleSleepSafety=1&scheduledWake=${version}&ownerId=fixture-owner`).expect(200);
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now,
|
||||
"fixture-owner", expect.any(Function), undefined, undefined, true, version === "2");
|
||||
"fixture-owner", expect.any(Function), undefined, undefined, true, version === "2", { incomingWake: undefined, deadlineCapability: undefined });
|
||||
});
|
||||
|
||||
describe("signed incoming capability handoff", () => {
|
||||
const createControlApp = (actor: any) => createApp(actor, undefined, undefined, true);
|
||||
const keyPair = generateKeyPairSync("ed25519");
|
||||
const publicJwk = { ...keyPair.publicKey.export({ format: "jwk" }), kid: "idle-route-test", alg: "EdDSA", use: "sig" };
|
||||
const ownerId = "d0b833f4-4098-42de-8420-1907f3aa4895";
|
||||
const catalog = {
|
||||
incomingWake: { version: 1, ownerId, generation: "a".repeat(64), apiKeyProof: { version: 1, challenge: "c".repeat(64) }, routes: ["tenant-api-http-v2"] },
|
||||
deadlineCatalog: "domain_deadlines_v1",
|
||||
};
|
||||
const path = `/api/instance/task-drain?idleSleepSafety=1&ownerId=${ownerId}`;
|
||||
function assertion(claims: Record<string, unknown> = {}) {
|
||||
const identity = routeModules.value.controlIdentity;
|
||||
const encode = (value: unknown) => Buffer.from(JSON.stringify(value)).toString("base64url");
|
||||
const iat = Math.floor(Date.now() / 1000);
|
||||
const header = encode({ alg: "EdDSA", typ: identity.CLOUD_CONTROL_JWS_TYPE, kid: publicJwk.kid });
|
||||
const payload = encode({ v: 1, iss: identity.CLOUD_RUNTIME_IDENTITY_ISSUER, aud: identity.CLOUD_CONTROL_AUDIENCE,
|
||||
sub: "stack-idle-route-test", action: "task-drain:read", requestId: randomUUID(), iat, exp: iat + 60,
|
||||
idleCapabilities: catalog, ...claims });
|
||||
return `${header}.${payload}.${sign(null, Buffer.from(`${header}.${payload}`, "ascii"), keyPair.privateKey).toString("base64url")}`;
|
||||
}
|
||||
beforeEach(() => {
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_RUNTIME_IDENTITY_JWKS", JSON.stringify({ keys: [publicJwk] }));
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_STACK_ID", "stack-idle-route-test");
|
||||
routeModules.value.controlIdentity.resetCloudControlReplayFenceForTests();
|
||||
mockHeartbeatService.getTaskDrainStatus.mockReturnValue(idleStatus);
|
||||
mockReadIdleSleepSafety.mockResolvedValue({ version: 3, backgroundWork: "none" });
|
||||
});
|
||||
afterEach(() => vi.unstubAllEnvs());
|
||||
|
||||
it("passes only the immutable signed catalog through real middleware and the production route", async () => {
|
||||
const token = assertion();
|
||||
const response = await request(createControlApp({ type: "none" }))
|
||||
.get(`${path}&incomingWake=${encodeURIComponent(JSON.stringify({ ...catalog.incomingWake, generation: "b".repeat(64) }))}`)
|
||||
.set("x-paperclip-cloud-control", token)
|
||||
.send({ idleCapabilities: { incomingWake: { ...catalog.incomingWake, ownerId: randomUUID() } } }).expect(200);
|
||||
expect(response.body.idleSleepSafety).toEqual({ version: 3, backgroundWork: "none" });
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenCalledExactlyOnceWith(mockDb, expect.any(Function), Date.now,
|
||||
ownerId, expect.any(Function), undefined, undefined, false, false,
|
||||
{ incomingWake: catalog, deadlineCapability: catalog });
|
||||
const options = mockReadIdleSleepSafety.mock.calls[0][9];
|
||||
expect(options.incomingWake).toBe(options.deadlineCapability);
|
||||
expect(Object.isFrozen(options.incomingWake)).toBe(true);
|
||||
expect(Object.isFrozen(options.incomingWake.incomingWake)).toBe(true);
|
||||
expect(Object.isFrozen(options.incomingWake.incomingWake.routes)).toBe(true);
|
||||
await request(createControlApp({ type: "none" })).get(path).set("x-paperclip-cloud-control", token).expect(401);
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it.each([
|
||||
["wrong stack", () => ({ sub: "other-stack" })],
|
||||
["wrong action", () => ({ action: "task-drain:start" })],
|
||||
["expired", () => ({ iat: Math.floor(Date.now() / 1000) - 600, exp: Math.floor(Date.now() / 1000) - 300 })],
|
||||
["unsupported catalog", () => ({ idleCapabilities: { incomingWake: { ...catalog.incomingWake, routes: ["unverified-path"] } } })],
|
||||
])("rejects %s before safety inspection", async (_name, claims) => {
|
||||
await request(createControlApp(adminActor)).get(path).set("x-paperclip-cloud-control", assertion(claims())).expect(401);
|
||||
expect(mockReadIdleSleepSafety).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("rejects a tampered catalog signature before safety inspection", async () => {
|
||||
const [header, payload, signature] = assertion().split(".");
|
||||
const changed = JSON.parse(Buffer.from(payload!, "base64url").toString());
|
||||
changed.idleCapabilities.incomingWake.generation = "b".repeat(64);
|
||||
const tampered = `${header}.${Buffer.from(JSON.stringify(changed)).toString("base64url")}.${signature}`;
|
||||
await request(createControlApp(adminActor)).get(path).set("x-paperclip-cloud-control", tampered).expect(401);
|
||||
expect(mockReadIdleSleepSafety).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("leaves legacy signed reads without a catalog conservative", async () => {
|
||||
await request(createControlApp({ type: "none" })).get(path)
|
||||
.set("x-paperclip-cloud-control", assertion({ idleCapabilities: undefined })).expect(200);
|
||||
expect(mockReadIdleSleepSafety.mock.calls[0][9]).toEqual({ incomingWake: undefined, deadlineCapability: undefined });
|
||||
});
|
||||
|
||||
it("cannot install capabilities from tenant, actor, query or body values", async () => {
|
||||
for (const source of ["session", "cloud_tenant", "local_implicit", "cloud_control"]) {
|
||||
const actor = { ...adminActor, source, idleCapabilities: catalog, incomingWake: catalog };
|
||||
await request(createControlApp(actor)).get(`${path}&incomingWake=${encodeURIComponent(JSON.stringify(catalog))}`)
|
||||
.send({ idleCapabilities: catalog }).expect(200);
|
||||
expect(mockReadIdleSleepSafety.mock.lastCall?.[9]).toEqual({ incomingWake: undefined, deadlineCapability: undefined });
|
||||
}
|
||||
mockReadIdleSleepSafety.mockClear();
|
||||
await request(createControlApp(nonAdminActor)).get(path).send({ idleCapabilities: catalog }).expect(403);
|
||||
expect(mockReadIdleSleepSafety).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
it("uses the live worker manager for inspection and releases only the owned hold", async () => {
|
||||
@@ -1199,7 +1289,7 @@ describe("instance settings routes", () => {
|
||||
expect(mockReadIdleSleepSafety).not.toHaveBeenCalled();
|
||||
await request(app).get("/api/instance/task-drain?idleSleepSafety=1&ownerId=fixture-owner").expect(200);
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now,
|
||||
"fixture-owner", expect.any(Function), undefined, backup, false, false);
|
||||
"fixture-owner", expect.any(Function), undefined, backup, false, false, { incomingWake: undefined, deadlineCapability: undefined });
|
||||
mockReadIdleSleepSafety.mockClear();
|
||||
await request(createApp(nonAdminActor, undefined, backup)).get("/api/instance/task-drain?idleSleepSafety=1").expect(403);
|
||||
expect(mockReadIdleSleepSafety).not.toHaveBeenCalled();
|
||||
@@ -1221,7 +1311,7 @@ describe("instance settings routes", () => {
|
||||
const control = createApp({ ...adminActor, source: "cloud_control" }, undefined, backup);
|
||||
await request(control).get("/api/instance/task-drain?idleSleepSafety=1&ownerId=fixture-owner").expect(200);
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now,
|
||||
"fixture-owner", expect.any(Function), undefined, backup, false, false);
|
||||
"fixture-owner", expect.any(Function), undefined, backup, false, false, { incomingWake: undefined, deadlineCapability: undefined });
|
||||
} finally {
|
||||
delete process.env.PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN;
|
||||
}
|
||||
|
||||
@@ -8,6 +8,7 @@ import { describe, expect, it, vi } from "vitest";
|
||||
import { setupLiveEventsWebSocketServer } from "../realtime/live-events-ws.js";
|
||||
import { publishLiveEvent } from "../services/live-events.js";
|
||||
import { taskToastEligible } from "../realtime/task-toast-eligibility.js";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
|
||||
vi.mock("../realtime/task-toast-eligibility.js", () => ({ taskToastEligible: vi.fn(async () => false) }));
|
||||
|
||||
@@ -39,6 +40,48 @@ function database(onPredicate: (kind: "where" | "join", condition: SQL) => void
|
||||
} };
|
||||
}
|
||||
describe("authenticated activity notifications", () => {
|
||||
it("retains live delivery idle work until the identifier read settles after the last socket closes", async () => {
|
||||
const baseline = idleWorkSnapshot().active;
|
||||
let release!: () => void, entered!: () => void;
|
||||
const pending = new Promise<void>(resolve => { release = resolve; });
|
||||
const started = new Promise<void>(resolve => { entered = resolve; });
|
||||
const db = database(), select = db.select;
|
||||
db.select = fields => {
|
||||
const query = select(fields);
|
||||
if (fields.identifier) {
|
||||
const limit = query.limit;
|
||||
query.limit = async () => { entered(); await pending; return limit(); };
|
||||
}
|
||||
return query;
|
||||
};
|
||||
const server = createServer();
|
||||
const wss = setupLiveEventsWebSocketServer(server, db as never, { deploymentMode: "local_trusted" });
|
||||
const closed = new Promise<void>(resolve => wss.on("connection", socket => socket.on("close", resolve)));
|
||||
server.listen(0, "127.0.0.1");
|
||||
await once(server, "listening");
|
||||
const socket = new WebSocket(`ws://127.0.0.1:${(server.address() as { port: number }).port}/api/companies/company/events/ws`);
|
||||
try {
|
||||
await once(socket, "open");
|
||||
await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(baseline));
|
||||
// System activity has no actor lookup: this gate is the new identifier SQL.
|
||||
publishLiveEvent({ companyId: "company", type: "activity.logged", payload: {
|
||||
action: "issue.updated", entityType: "issue", entityId: "visible", actorType: "system",
|
||||
} });
|
||||
await started;
|
||||
socket.terminate();
|
||||
await closed;
|
||||
expect(wss.clients.size).toBe(0);
|
||||
expect(idleWorkSnapshot().active).toBe(baseline + 1);
|
||||
release();
|
||||
await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(baseline));
|
||||
} finally {
|
||||
release();
|
||||
socket.terminate();
|
||||
await new Promise<void>(resolve => server.close(() => resolve()));
|
||||
await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(baseline));
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["local_trusted", "authenticated"] as const)("keeps authorized notifications when optional identifier reads fail (%s)", async deploymentMode => {
|
||||
const db = database();
|
||||
const select = db.select;
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { idleApiActivity } from "../services/idle-api-activity.js";
|
||||
import { createHash, timingSafeEqual } from "node:crypto";
|
||||
import type { Request, RequestHandler } from "express";
|
||||
import { and, eq, isNull, sql } from "drizzle-orm";
|
||||
@@ -43,7 +44,7 @@ import { ensureHumanRoleDefaultGrants } from "../services/principal-access-compa
|
||||
import { forbidden, unauthorized, unprocessable } from "../errors.js";
|
||||
|
||||
export { isCloudManagedInstance } from "../services/cloud-instance.js";
|
||||
import { cloudTenantPrimaryCompanyId } from "../services/cloud-instance.js";
|
||||
import { cloudTenantPrimaryCompanyId, isCloudManagedInstance } from "../services/cloud-instance.js";
|
||||
|
||||
function hashToken(token: string) {
|
||||
return createHash("sha256").update(token).digest("hex");
|
||||
@@ -329,6 +330,9 @@ export function actorMiddleware(db: Db, opts: ActorMiddlewareOptions): RequestHa
|
||||
runId: runIdHeader || undefined,
|
||||
source: "board_key",
|
||||
};
|
||||
if (isCloudManagedInstance() && (access.isInstanceAdmin || access.companyIds.length)) {
|
||||
_res.once("finish", () => { if (_res.statusCode >= 200 && _res.statusCode < 400) idleApiActivity.record(); });
|
||||
}
|
||||
next();
|
||||
return;
|
||||
}
|
||||
@@ -521,7 +525,9 @@ export function actorMiddleware(db: Db, opts: ActorMiddlewareOptions): RequestHa
|
||||
runId: runIdHeader || undefined,
|
||||
source: "agent_key",
|
||||
};
|
||||
|
||||
if (isCloudManagedInstance() && req.actor.onBehalfOfMemberships?.length) {
|
||||
_res.once("finish", () => { if (_res.statusCode >= 200 && _res.statusCode < 400) idleApiActivity.record(); });
|
||||
}
|
||||
next();
|
||||
};
|
||||
return async (req, res, next) => {
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import type { Request } from "express";
|
||||
import type { VerifiedIdleCapabilities } from "../services/idle-incoming-wake-capability.js";
|
||||
import type { RequestHandler } from "express";
|
||||
import { logger } from "./logger.js";
|
||||
import {
|
||||
@@ -6,6 +8,12 @@ import {
|
||||
type CloudControlAction,
|
||||
} from "../services/cloud-runtime-identity.js";
|
||||
|
||||
const idleCapabilities = new WeakMap<Request, VerifiedIdleCapabilities>();
|
||||
/** Only this middleware's verified assertion can install request capabilities. */
|
||||
export function verifiedIdleCapabilities(req: Request): VerifiedIdleCapabilities | undefined {
|
||||
return idleCapabilities.get(req);
|
||||
}
|
||||
|
||||
/** Endpoint → method → the one action a control assertion must name to take it. */
|
||||
const ACTIONS_BY_ENDPOINT: Record<string, Record<string, CloudControlAction>> = {
|
||||
"/api/instance/connector-events/wake": { POST: "connector-events:wake" },
|
||||
@@ -50,7 +58,8 @@ export function cloudControlMiddleware(): RequestHandler {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
verifyCloudControlAssertion({ compactJws: assertion, expectedAction });
|
||||
const claims = verifyCloudControlAssertion({ compactJws: assertion, expectedAction });
|
||||
if (claims.idleCapabilities) idleCapabilities.set(req, claims.idleCapabilities as VerifiedIdleCapabilities);
|
||||
} catch (error) {
|
||||
logger.warn({ err: error }, "Rejected Cloud control assertion");
|
||||
res.status(401).json({ error: "invalid_cloud_control_assertion" });
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import type { LiveEvent } from "@paperclipai/shared";
|
||||
import { createLiveEventDelivery, LIVE_DELIVERY_MAX_BYTES, LIVE_DELIVERY_MAX_ITEMS } from "./live-event-delivery.js";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
|
||||
const event = (seq = 1, runId = "run-1"): LiveEvent => ({ id: seq, companyId: "company", type: "heartbeat.run.log",
|
||||
createdAt: "2026-10-24T00:00:00Z", payload: { runId, agentId: "agent", seq, chunk: "output" } });
|
||||
@@ -116,7 +117,7 @@ it.each([false, true])("preserves the overflow deadline across resubscription un
|
||||
await vi.advanceTimersByTimeAsync(1_000);
|
||||
if (settled) expect(f.socket.close).not.toHaveBeenCalled();
|
||||
else expect(f.socket.close).toHaveBeenCalledWith(1013, "persistent event backlog");
|
||||
} finally { f.delivery.dispose(); }
|
||||
} finally { f.delivery.dispose(); release(event()); await settle(); }
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
});
|
||||
it("recovers failed delivery reads through resync rather than silently claiming healthy streaming", async () => {
|
||||
@@ -146,6 +147,7 @@ it("rejects unbounded interest sets and stale revision commands", () => {
|
||||
it.each([false, true].flatMap(negotiated => ["authorize", "decorate"].map(stage => ({ negotiated, stage }))))(
|
||||
"closes a stalled $stage read below the queue bound (negotiated=$negotiated)", async ({ negotiated, stage }) => {
|
||||
vi.useFakeTimers();
|
||||
const baseline = idleWorkSnapshot().active;
|
||||
let release!: (value: LiveEvent) => void;
|
||||
const pending = new Promise<LiveEvent>(resolve => { release = resolve; });
|
||||
const f = fixture(stage === "authorize" ? vi.fn(() => pending) : undefined);
|
||||
@@ -168,11 +170,13 @@ it.each([false, true].flatMap(negotiated => ["authorize", "decorate"].map(stage
|
||||
expect(f.socket.close).toHaveBeenCalledWith(1013, "persistent event backlog");
|
||||
expect(f.delivery.metrics().retainedItems).toBe(1);
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
expect(idleWorkSnapshot().active).toBe(baseline + 1);
|
||||
release(source); await settle();
|
||||
expect(idleWorkSnapshot().active).toBe(baseline);
|
||||
expect(f.delivery.metrics()).toMatchObject({ retainedItems: 0, retainedBytes: 0 });
|
||||
expect(f.frames.filter(frame => frame.type.startsWith("heartbeat."))).toEqual([]);
|
||||
expect(f.authorize).toHaveBeenCalledOnce();
|
||||
} finally { f.delivery.dispose(); other.delivery.dispose(); }
|
||||
} finally { f.delivery.dispose(); other.delivery.dispose(); release(event()); await settle(); }
|
||||
});
|
||||
|
||||
it.each(["resolve", "dispose"])("clears a pending-read deadline on %s without idle work", async ending => {
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import type { LiveEvent } from "@paperclipai/shared";
|
||||
import { beginIdleTrackedWork } from "../services/task-admission.js";
|
||||
|
||||
export const LIVE_DELIVERY_MAX_ITEMS = 1_000;
|
||||
export const LIVE_DELIVERY_MAX_BYTES = 1024 * 1024;
|
||||
@@ -90,6 +91,7 @@ export function createLiveEventDelivery(input: {
|
||||
async function drain() {
|
||||
if (draining) return;
|
||||
draining = true;
|
||||
const finishIdleWork = beginIdleTrackedWork();
|
||||
try {
|
||||
while (queued.length && isOpen()) {
|
||||
const item = queued.shift()!;
|
||||
@@ -121,7 +123,10 @@ export function createLiveEventDelivery(input: {
|
||||
retainedBytes -= item.bytes; retainedItems -= 1;
|
||||
}
|
||||
}
|
||||
} finally { draining = false; resync(); }
|
||||
} finally {
|
||||
try { draining = false; resync(); }
|
||||
finally { finishIdleWork(); }
|
||||
}
|
||||
}
|
||||
return {
|
||||
receive(raw: string) {
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { verifiedIdleCapabilities } from "../middleware/cloud-control.js";
|
||||
import { trackIdleWork } from "../services/task-admission.js";
|
||||
import { Router, type Request } from "express";
|
||||
import { companies, type Db } from "@paperclipai/db";
|
||||
@@ -336,7 +337,8 @@ export function instanceSettingsRoutes(db: Db, pluginWorkers?: PluginWorkerManag
|
||||
typeof req.query.ownerId === "string" ? req.query.ownerId : undefined,
|
||||
diagnostic => readIdleLocalWork({ backupCheckpoint: Boolean(prepareIdleDatabaseBackup), diagnostic }),
|
||||
pluginWorkers?.inspectIdleSleep?.bind(pluginWorkers), prepareIdleDatabaseBackup, scheduledRoutines, scheduledIssueMonitors,
|
||||
...(req.query.idleSleepDiagnostics === "1" ? [{ collectDiagnostics: true }] as const : [] as const));
|
||||
{ incomingWake: verifiedIdleCapabilities(req), deadlineCapability: verifiedIdleCapabilities(req),
|
||||
...(req.query.idleSleepDiagnostics === "1" ? { collectDiagnostics: true } : {}) });
|
||||
res.json({ ...heartbeat.getTaskDrainStatus(), idleSleepSafety });
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -136,6 +136,7 @@ import { notifyChatDeliveryWork, notifyChatPublicationWork } from "./chat-work-n
|
||||
|
||||
import { voiceSessionStore } from "./voice/voice-session-store.js";
|
||||
import { createChatRuntime, type ChatRuntime } from "./chat-runtime.js";
|
||||
import { registerIdleLocalWorkProbe } from "./idle-local-work.js";
|
||||
import type { ChatEndpointRuntime } from "./voice/voice-runtime.js";
|
||||
|
||||
import { fastResponseService, fastResponseSourceCurrent } from "./fast-responses.js";
|
||||
@@ -517,6 +518,11 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
version: string;
|
||||
}
|
||||
>();
|
||||
// SDK initialize() and the final authority read continue after the registry's
|
||||
// replacement tail settles. Neither that work nor a failed retirement may
|
||||
// disappear just because the archived endpoint is no longer returned by get().
|
||||
const unregisterRuntimeIdleProbe = registerIdleLocalWorkProbe(() =>
|
||||
runtimeInitializations.size > 0 || runtime.hasUnsettledLifecycle() ? "present" : "none");
|
||||
const { recordCurrentMicrosoftTeamsRoute } = createTeamsRoutes(db, { runtime, runtimeCallbackEndpoint });
|
||||
const persistence = createChatSdkStatePersistence(db);
|
||||
const fetchImpl = options.fetch ?? globalThis.fetch;
|
||||
@@ -3622,12 +3628,17 @@ return {
|
||||
// Stop every provider runtime before releasing leader rows. This gives a
|
||||
// standby an immediate takeover path without overlapping Gateway sockets.
|
||||
await runtime.shutdown();
|
||||
await Promise.allSettled([...runtimeInitializations.values()].map(initialization => initialization.promise));
|
||||
await slackRegistration.close();
|
||||
await Promise.all(
|
||||
ownedDiscordGateways.map((ownership) =>
|
||||
releaseDiscordGatewayLeaseRow(ownership),
|
||||
),
|
||||
);
|
||||
if (runtimeInitializations.size > 0 || runtime.hasUnsettledLifecycle()) {
|
||||
throw new Error("Chat runtime lifecycle is still unsettled after shutdown");
|
||||
}
|
||||
unregisterRuntimeIdleProbe();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -23,6 +23,11 @@ export class ChatRuntime {
|
||||
return [...this.endpoints.values()];
|
||||
}
|
||||
|
||||
/** Retirement failures retain their owner even after get/list hide it. */
|
||||
hasUnsettledLifecycle(): boolean {
|
||||
return this.retiringEndpoints.size > 0 || this.lifecycleTails.size > 0;
|
||||
}
|
||||
|
||||
private enqueueLifecycle<T>(
|
||||
endpointId: string,
|
||||
operation: () => Promise<T>,
|
||||
|
||||
@@ -241,6 +241,8 @@ import {
|
||||
type CreateChatSdkEndpointRuntimeOptions,
|
||||
} from "./chat-sdk-runtime.js";
|
||||
|
||||
import { createChatRuntime } from "./chat-runtime.js";
|
||||
|
||||
const persistence: ChatSdkStatePersistence = {
|
||||
async compareAndSet() {
|
||||
return true;
|
||||
@@ -1152,6 +1154,31 @@ describe("Chat SDK endpoint runtime", () => {
|
||||
await registry.shutdown();
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ name: "SDK", create: createChatSdkRuntime },
|
||||
{ name: "shared", create: createChatRuntime },
|
||||
])("keeps $name lifecycle work visible while hidden retirement fails, until its owner retries", async ({ create }) => {
|
||||
const registry = create();
|
||||
const first = await registry.replaceEndpoint(registryOptions());
|
||||
expect(registry.hasUnsettledLifecycle()).toBe(false);
|
||||
const held = deferred();
|
||||
const shutdown = vi.spyOn(first, "shutdown").mockReturnValueOnce(held.promise).mockResolvedValue(undefined);
|
||||
const removal = registry.removeEndpoint("endpoint-1");
|
||||
const result = removal.catch((error: Error) => error.message);
|
||||
expect(registry.hasUnsettledLifecycle()).toBe(true);
|
||||
await vi.waitFor(() => expect(shutdown).toHaveBeenCalledOnce());
|
||||
expect(registry.get("endpoint-1")).toBeNull();
|
||||
expect(registry.list()).toEqual([]);
|
||||
expect(registry.hasUnsettledLifecycle()).toBe(true);
|
||||
held.reject(new Error("retirement uncertain"));
|
||||
expect(await result).toBe("retirement uncertain");
|
||||
expect(registry.hasUnsettledLifecycle()).toBe(true);
|
||||
await registry.removeEndpoint("endpoint-1");
|
||||
expect(shutdown).toHaveBeenCalledTimes(2);
|
||||
expect(registry.hasUnsettledLifecycle()).toBe(false);
|
||||
await registry.shutdown();
|
||||
});
|
||||
|
||||
it("retains failed retirement ownership and only constructs a replacement after an explicit successful retry", async () => {
|
||||
const registry = createChatSdkRuntime();
|
||||
const first = await registry.replaceEndpoint(registryOptions());
|
||||
|
||||
@@ -3195,6 +3195,11 @@ export class ChatSdkRuntime {
|
||||
return [...this.endpoints.values()];
|
||||
}
|
||||
|
||||
/** Retirement failures retain their owner even after get/list hide it. */
|
||||
hasUnsettledLifecycle(): boolean {
|
||||
return this.retiringEndpoints.size > 0 || this.lifecycleTails.size > 0;
|
||||
}
|
||||
|
||||
private enqueueLifecycle<T>(
|
||||
endpointId: string,
|
||||
operation: () => Promise<T>,
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { parseIdleCapabilities, type IdleCapabilities } from "./idle-incoming-wake-capability.js";
|
||||
import { createPublicKey, timingSafeEqual, verify, type JsonWebKey } from "node:crypto";
|
||||
import { eq } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
@@ -491,6 +492,7 @@ export type CloudControlClaims = {
|
||||
sub: string;
|
||||
action: CloudControlAction;
|
||||
requestId: string;
|
||||
idleCapabilities?: IdleCapabilities;
|
||||
iat: number;
|
||||
exp: number;
|
||||
};
|
||||
@@ -572,6 +574,10 @@ export function verifyCloudControlAssertion(input: {
|
||||
) {
|
||||
throw new Error("Cloud control assertion is expired or has an invalid lifetime");
|
||||
}
|
||||
if (payload.idleCapabilities !== undefined) {
|
||||
if (payload.action !== "task-drain:read") throw new Error("Idle capabilities require task-drain:read");
|
||||
payload.idleCapabilities = parseIdleCapabilities(payload.idleCapabilities);
|
||||
}
|
||||
// Consumed LAST, only after every other check passed: a rejected
|
||||
// assertion must not burn its request id, or an attacker could deny a
|
||||
// legitimate call by replaying a mangled copy of it first.
|
||||
|
||||
@@ -10,3 +10,6 @@ export function wakeConnectionEvents(owner: object): boolean {
|
||||
wake();
|
||||
return true;
|
||||
}
|
||||
|
||||
/** Inspection never emits a wake or polls the broker. */
|
||||
export const hasConnectionEventWakeReceiver = (owner: object): boolean => receivers.has(owner);
|
||||
@@ -146,7 +146,7 @@ function normalizeLeasedPayload(event: LeasedEvent): Record<string, unknown> {
|
||||
return compact(base);
|
||||
}
|
||||
|
||||
function bindingRows(rows: Array<{
|
||||
export function githubConnectionEventBindings(rows: Array<{
|
||||
grant: typeof connectionGrants.$inferSelect;
|
||||
connection: typeof toolConnections.$inferSelect;
|
||||
}>): GitHubBinding[] {
|
||||
@@ -259,7 +259,7 @@ export function githubConnectionEventService(
|
||||
eq(toolConnections.enabled, true),
|
||||
sql`exists (select 1 from companies where id = ${connectionGrants.companyId} and status <> 'archived')`,
|
||||
));
|
||||
return bindingRows(rows);
|
||||
return githubConnectionEventBindings(rows);
|
||||
}
|
||||
|
||||
async function applyPullRequestEvent(database: Db, companyId: string, event: LeasedEvent, effects: Array<(committedDb: Db) => Promise<void>>) {
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { IdleApiActivity } from "./idle-api-activity.js";
|
||||
|
||||
describe("authenticated API idle activity", () => {
|
||||
it("is passive without successful authentication and expires without timers or dependency work", () => {
|
||||
let monotonic = 1;
|
||||
const activity = new IdleApiActivity(() => monotonic);
|
||||
expect(activity.active()).toBe(false);
|
||||
activity.record();
|
||||
expect(activity.active()).toBe(true);
|
||||
monotonic += 599_999;
|
||||
expect(activity.active()).toBe(true);
|
||||
monotonic++;
|
||||
expect(activity.active()).toBe(false);
|
||||
});
|
||||
it("extends only from the actual latest successful use and ignores wall clock changes", () => {
|
||||
let monotonic = 1;
|
||||
const activity = new IdleApiActivity(() => monotonic);
|
||||
activity.record();
|
||||
monotonic += 300_000;
|
||||
activity.record();
|
||||
monotonic += 300_001;
|
||||
expect(activity.active()).toBe(true);
|
||||
monotonic += 299_999;
|
||||
expect(activity.active()).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,12 @@
|
||||
/** Authenticated API use is real activity; rejecting an unknown key is not. */
|
||||
export class IdleApiActivity {
|
||||
private until = 0;
|
||||
constructor(private readonly now: () => number = () => performance.now()) {}
|
||||
record(): void {
|
||||
this.until = Math.max(this.until, this.now() + 10 * 60_000);
|
||||
}
|
||||
active(): boolean {
|
||||
return this.now() < this.until;
|
||||
}
|
||||
}
|
||||
export const idleApiActivity = new IdleApiActivity();
|
||||
@@ -0,0 +1,33 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { apiWakeFingerprint, createApiWakeAdmission, parseApiWakeAdmission, matchesApiWakeAdmission } from "./idle-api-wake-admission.js";
|
||||
|
||||
const token = "api-wake-vector-token";
|
||||
const keyHash = createHash("sha256").update(token).digest("hex");
|
||||
const input = { stackId: "vector-stack", ownerId: "00000000-0000-4000-8000-000000000001", capabilityGeneration: "a".repeat(64), challenge: "c".repeat(64) };
|
||||
const fingerprint = apiWakeFingerprint({ ...input, keyHash });
|
||||
const receipt = createApiWakeAdmission({ ...input, keys: [{ fingerprint, expiresAt: null }] });
|
||||
describe("API wake receipt wire vectors", () => {
|
||||
it("uses exact domain-separated vectors and exports neither token nor database hash", () => {
|
||||
expect(fingerprint).toBe("6478041b90f4665eeca8403186fefedc456bf1322f38dab9fb89212eecaafd43");
|
||||
expect(JSON.stringify(receipt)).not.toContain(token);
|
||||
expect(JSON.stringify(receipt)).not.toContain(keyHash);
|
||||
expect(matchesApiWakeAdmission(receipt, { stackId: input.stackId, token, now: new Date("2099-01-01").getTime() })).toBe(true);
|
||||
expect(matchesApiWakeAdmission(receipt, { stackId: "other", token, now: 0 })).toBe(false);
|
||||
expect(matchesApiWakeAdmission(receipt, { stackId: input.stackId, token: "unknown", now: 0 })).toBe(false);
|
||||
});
|
||||
it("enforces board expiry and canonical unique bounded complete receipts", () => {
|
||||
const changedChallenge = createApiWakeAdmission({ ...input, challenge: "b".repeat(64), keys: receipt.keys });
|
||||
expect(matchesApiWakeAdmission(changedChallenge, { stackId: input.stackId, token, now: 0 })).toBe(false);
|
||||
const expiring = createApiWakeAdmission({ ...input, keys: [{ fingerprint, expiresAt: new Date(1000).toISOString() }] });
|
||||
expect(matchesApiWakeAdmission(expiring, { stackId: input.stackId, token, now: 999 })).toBe(true);
|
||||
expect(matchesApiWakeAdmission(expiring, { stackId: input.stackId, token, now: 1000 })).toBe(false);
|
||||
for (const invalid of [
|
||||
{ ...receipt, generation: "b".repeat(64) }, { ...receipt, challenge: "b".repeat(64) },
|
||||
{ ...receipt, extra: true }, { ...receipt, keys: [...receipt.keys, ...receipt.keys] },
|
||||
{ ...receipt, keys: Array.from({ length: 1001 }, () => receipt.keys[0]) },
|
||||
{ ...receipt, keys: [{ fingerprint, expiresAt: "2099-01-01" }] },
|
||||
{ ...receipt, keys: [{ fingerprint: "d".repeat(64), expiresAt: null }, { fingerprint: "a".repeat(64), expiresAt: null }] },
|
||||
]) expect(() => parseApiWakeAdmission(invalid)).toThrow();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,56 @@
|
||||
import { createHash } from "node:crypto";
|
||||
|
||||
export type ApiWakeClaim = { generation: string; fingerprint: string };
|
||||
export interface ApiWakeAdmission {
|
||||
version: 1;
|
||||
stackId: string;
|
||||
ownerId: string;
|
||||
capabilityGeneration: string;
|
||||
challenge: string;
|
||||
generation: string;
|
||||
keys: { fingerprint: string; expiresAt: string | null }[];
|
||||
}
|
||||
const digest = (value: unknown): value is string => typeof value === "string" && /^[a-f0-9]{64}$/.test(value);
|
||||
const object = (value: unknown): value is Record<string, unknown> => Boolean(value && typeof value === "object" && !Array.isArray(value));
|
||||
const exact = (value: Record<string, unknown>, keys: string[]) => Object.keys(value).length === keys.length && Object.keys(value).every(key => keys.includes(key));
|
||||
const generation = (value: Omit<ApiWakeAdmission, "generation">) => createHash("sha256").update(JSON.stringify([
|
||||
"paperclip-api-wake-receipt-v1", value.version, value.stackId, value.ownerId, value.capabilityGeneration, value.challenge,
|
||||
value.keys.map(key => [key.fingerprint, key.expiresAt]),
|
||||
])).digest("hex");
|
||||
export function apiWakeFingerprint(input: { stackId: string; ownerId: string; challenge: string; keyHash: string }): string {
|
||||
if (!digest(input.keyHash) || !digest(input.challenge)) throw new Error("Invalid API wake fingerprint input");
|
||||
return createHash("sha256").update(JSON.stringify(["paperclip-api-wake-v1", input.stackId, input.ownerId, input.challenge, input.keyHash])).digest("hex");
|
||||
}
|
||||
/** Strict bounded canonical wire contract. Parsing never grants tenant permissions. */
|
||||
export function parseApiWakeAdmission(value: unknown): ApiWakeAdmission {
|
||||
if (!object(value) || !exact(value, ["version", "stackId", "ownerId", "capabilityGeneration", "challenge", "generation", "keys"])
|
||||
|| value.version !== 1 || typeof value.stackId !== "string" || !/^[A-Za-z0-9_-]{1,200}$/.test(value.stackId)
|
||||
|| typeof value.ownerId !== "string" || !/^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i.test(value.ownerId)
|
||||
|| !digest(value.capabilityGeneration) || !digest(value.challenge) || !digest(value.generation)
|
||||
|| !Array.isArray(value.keys) || value.keys.length > 1000) throw new Error("Invalid API wake receipt");
|
||||
let previous = "";
|
||||
for (const key of value.keys) {
|
||||
if (!object(key) || !exact(key, ["fingerprint", "expiresAt"]) || !digest(key.fingerprint) || key.fingerprint <= previous
|
||||
|| !(key.expiresAt === null || typeof key.expiresAt === "string" && key.expiresAt.length <= 30 && Number.isFinite(Date.parse(key.expiresAt))
|
||||
&& new Date(key.expiresAt).toISOString() === key.expiresAt)) throw new Error("Invalid API wake receipt key");
|
||||
previous = key.fingerprint;
|
||||
}
|
||||
const parsed = value as unknown as ApiWakeAdmission;
|
||||
if (generation(parsed) !== parsed.generation) throw new Error("API wake receipt generation changed");
|
||||
return { version: 1, stackId: parsed.stackId, ownerId: parsed.ownerId, capabilityGeneration: parsed.capabilityGeneration,
|
||||
challenge: parsed.challenge, generation: parsed.generation,
|
||||
keys: parsed.keys.map(key => ({ fingerprint: key.fingerprint, expiresAt: key.expiresAt })) };
|
||||
}
|
||||
export function createApiWakeAdmission(input: Omit<ApiWakeAdmission, "version" | "generation">): ApiWakeAdmission {
|
||||
const canonical = { version: 1 as const, ...input, keys: [...input.keys].sort((a, b) => a.fingerprint.localeCompare(b.fingerprint)) };
|
||||
return parseApiWakeAdmission({ ...canonical, generation: generation(canonical) });
|
||||
}
|
||||
export function authorizeApiWakeAdmission(receipt: ApiWakeAdmission, input: { stackId: string; token: string; now: number }): ApiWakeClaim | undefined {
|
||||
if (receipt.stackId !== input.stackId || input.token.length < 1 || input.token.length > 8192 || /\s/.test(input.token)) return;
|
||||
const fingerprint = apiWakeFingerprint({ ...receipt, keyHash: createHash("sha256").update(input.token).digest("hex") });
|
||||
return receipt.keys.some(key => key.fingerprint === fingerprint && (key.expiresAt === null || Date.parse(key.expiresAt) > input.now))
|
||||
? { generation: receipt.generation, fingerprint } : undefined;
|
||||
}
|
||||
export function matchesApiWakeAdmission(receipt: ApiWakeAdmission, input: { stackId: string; token: string; now: number }): boolean {
|
||||
return authorizeApiWakeAdmission(receipt, input) !== undefined;
|
||||
}
|
||||
@@ -0,0 +1,113 @@
|
||||
import { generateKeyPairSync, sign } from "node:crypto";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { parseIdleCapabilities } from "./idle-incoming-wake-capability.js";
|
||||
import {
|
||||
verifyCloudControlAssertion,
|
||||
resetCloudControlReplayFenceForTests,
|
||||
CLOUD_CONTROL_AUDIENCE,
|
||||
CLOUD_CONTROL_JWS_TYPE,
|
||||
CLOUD_RUNTIME_IDENTITY_ISSUER,
|
||||
} from "./cloud-runtime-identity.js";
|
||||
const incoming = {
|
||||
version: 1 as const,
|
||||
ownerId: "00000000-0000-4000-8000-000000000001",
|
||||
generation: "a".repeat(64),
|
||||
apiKeyProof: { version: 1 as const, challenge: "c".repeat(64) },
|
||||
routes: ["tenant-api-http-v2" as const],
|
||||
};
|
||||
const invalid: unknown[] = [
|
||||
null,
|
||||
{},
|
||||
{ other: true },
|
||||
{ deadlineCatalog: "other" },
|
||||
{ incomingWake: { ...incoming, version: 2 } },
|
||||
{ incomingWake: { ...incoming, routes: ["all"] } },
|
||||
{
|
||||
incomingWake: {
|
||||
...incoming,
|
||||
routes: ["tenant-api-http-v2", "tenant-api-http-v2"],
|
||||
},
|
||||
},
|
||||
{ incomingWake: { ...incoming, ownerId: "other" } },
|
||||
{ incomingWake: { ...incoming, extra: true } },
|
||||
{
|
||||
incomingWake: {
|
||||
...incoming,
|
||||
githubBroker: {
|
||||
instanceId: "instance",
|
||||
bindingsFingerprint: "b".repeat(64),
|
||||
},
|
||||
},
|
||||
},
|
||||
{ incomingWake: { ...incoming, routes: ["github-broker-events-v1"] } },
|
||||
];
|
||||
|
||||
describe("signed idle capabilities", () => {
|
||||
it.each(invalid)("rejects unknown/partial capabilities: %j", (value) => {
|
||||
expect(() => parseIdleCapabilities(value)).toThrow();
|
||||
});
|
||||
it("detaches and freezes the exact known catalogs", () => {
|
||||
const source = {
|
||||
incomingWake: { ...incoming, routes: [...incoming.routes] },
|
||||
deadlineCatalog: "domain_deadlines_v1",
|
||||
};
|
||||
const parsed = parseIdleCapabilities(source);
|
||||
source.incomingWake.routes.length = 0;
|
||||
expect(parsed.incomingWake?.routes).toEqual(["tenant-api-http-v2"]);
|
||||
expect(Object.isFrozen(parsed.incomingWake?.routes)).toBe(true);
|
||||
});
|
||||
it("binds capability to signed stack, action and replay fence", () => {
|
||||
resetCloudControlReplayFenceForTests();
|
||||
const pair = generateKeyPairSync("ed25519");
|
||||
const now = new Date("2099-01-01T00:00:00Z");
|
||||
const jwk = {
|
||||
...pair.publicKey.export({ format: "jwk" }),
|
||||
kid: "idle",
|
||||
use: "sig",
|
||||
alg: "EdDSA",
|
||||
};
|
||||
const env = {
|
||||
PAPERCLIP_CLOUD_STACK_ID: "stack",
|
||||
PAPERCLIP_CLOUD_RUNTIME_IDENTITY_JWKS: JSON.stringify({ keys: [jwk] }),
|
||||
};
|
||||
const encode = (v: unknown) =>
|
||||
Buffer.from(JSON.stringify(v)).toString("base64url");
|
||||
const token = (overrides: object = {}) => {
|
||||
const header = encode({
|
||||
alg: "EdDSA",
|
||||
typ: CLOUD_CONTROL_JWS_TYPE,
|
||||
kid: "idle",
|
||||
});
|
||||
const body = encode({
|
||||
v: 1,
|
||||
iss: CLOUD_RUNTIME_IDENTITY_ISSUER,
|
||||
aud: CLOUD_CONTROL_AUDIENCE,
|
||||
sub: "stack",
|
||||
action: "task-drain:read",
|
||||
requestId: "same",
|
||||
iat: now.getTime() / 1000,
|
||||
exp: now.getTime() / 1000 + 60,
|
||||
idleCapabilities: {
|
||||
incomingWake: incoming,
|
||||
deadlineCatalog: "domain_deadlines_v1",
|
||||
},
|
||||
...overrides,
|
||||
});
|
||||
return `${header}.${body}.${sign(null, Buffer.from(`${header}.${body}`), pair.privateKey).toString("base64url")}`;
|
||||
};
|
||||
const verify = (compactJws: string) =>
|
||||
verifyCloudControlAssertion({
|
||||
compactJws,
|
||||
env,
|
||||
now,
|
||||
expectedAction: "task-drain:read",
|
||||
});
|
||||
expect(() => verify(token({ sub: "foreign" }))).toThrow();
|
||||
expect(() => verify(token({ action: "task-drain:start" }))).toThrow();
|
||||
expect(() =>
|
||||
verify(token({ idleCapabilities: { deadlineCatalog: "unknown" } })),
|
||||
).toThrow();
|
||||
expect(verify(token()).idleCapabilities?.incomingWake).toEqual(incoming);
|
||||
expect(() => verify(token())).toThrow(/already been used/);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,93 @@
|
||||
/** Closed Cloud-owned ingress catalog. No credential material belongs in this proof. */
|
||||
export interface IdleIncomingWakeCapability {
|
||||
version: 1;
|
||||
ownerId: string;
|
||||
generation: string;
|
||||
routes: ("tenant-api-http-v2" | "github-broker-events-v1")[];
|
||||
apiKeyProof?: { version: 1; challenge: string };
|
||||
githubBroker?: { instanceId: string; bindingsFingerprint: string };
|
||||
}
|
||||
export interface IdleCapabilities {
|
||||
incomingWake?: IdleIncomingWakeCapability;
|
||||
deadlineCatalog?: "domain_deadlines_v1";
|
||||
}
|
||||
const object = (v: unknown): v is Record<string, unknown> =>
|
||||
Boolean(v && typeof v === "object" && !Array.isArray(v));
|
||||
const exact = (v: Record<string, unknown>, keys: string[]) =>
|
||||
Object.keys(v).every((k) => keys.includes(k));
|
||||
const digest = (v: unknown): v is string =>
|
||||
typeof v === "string" && /^[a-f0-9]{64}$/.test(v);
|
||||
export function parseIdleCapabilities(value: unknown): IdleCapabilities {
|
||||
if (
|
||||
!object(value) ||
|
||||
!exact(value, ["incomingWake", "deadlineCatalog"]) ||
|
||||
!Object.keys(value).length
|
||||
)
|
||||
throw new Error("Invalid idle capabilities");
|
||||
if (
|
||||
value.deadlineCatalog !== undefined &&
|
||||
value.deadlineCatalog !== "domain_deadlines_v1"
|
||||
)
|
||||
throw new Error("Unknown idle deadline catalog");
|
||||
if (value.incomingWake !== undefined) {
|
||||
const v = value.incomingWake;
|
||||
if (
|
||||
!object(v) ||
|
||||
!exact(v, [
|
||||
"version",
|
||||
"ownerId",
|
||||
"generation",
|
||||
"routes",
|
||||
"githubBroker",
|
||||
"apiKeyProof",
|
||||
]) ||
|
||||
v.version !== 1 ||
|
||||
typeof v.ownerId !== "string" ||
|
||||
!/^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i.test(
|
||||
v.ownerId,
|
||||
) ||
|
||||
!digest(v.generation) ||
|
||||
!Array.isArray(v.routes) ||
|
||||
v.routes.length < 1 ||
|
||||
v.routes.length > 2 ||
|
||||
v.routes.some(
|
||||
(r) => !["tenant-api-http-v2", "github-broker-events-v1"].includes(r),
|
||||
) ||
|
||||
new Set(v.routes).size !== v.routes.length ||
|
||||
[...v.routes].sort().join() !== v.routes.join()
|
||||
)
|
||||
throw new Error("Invalid incoming wake capability");
|
||||
if (v.routes.includes("tenant-api-http-v2")) {
|
||||
const proof = v.apiKeyProof;
|
||||
if (!object(proof) || Object.keys(proof).length !== 2 || !exact(proof, ["version", "challenge"])
|
||||
|| proof.version !== 1 || !digest(proof.challenge)) throw new Error("Invalid API wake challenge");
|
||||
} else if (v.apiKeyProof !== undefined) throw new Error("Unexpected API wake challenge");
|
||||
if (v.routes.includes("github-broker-events-v1")) {
|
||||
const b = v.githubBroker;
|
||||
if (
|
||||
!object(b) ||
|
||||
!exact(b, ["instanceId", "bindingsFingerprint"]) ||
|
||||
typeof b.instanceId !== "string" ||
|
||||
!/^[A-Za-z0-9_-]{1,200}$/.test(b.instanceId) ||
|
||||
!digest(b.bindingsFingerprint)
|
||||
)
|
||||
throw new Error("Invalid GitHub wake binding");
|
||||
} else if (v.githubBroker !== undefined)
|
||||
throw new Error("Unexpected GitHub wake binding");
|
||||
}
|
||||
// Return a detached immutable snapshot; request callers cannot mutate a verified claim.
|
||||
const parsed = JSON.parse(JSON.stringify(value)) as IdleCapabilities;
|
||||
if (parsed.incomingWake) {
|
||||
Object.freeze(parsed.incomingWake.routes);
|
||||
if (parsed.incomingWake.apiKeyProof) Object.freeze(parsed.incomingWake.apiKeyProof);
|
||||
if (parsed.incomingWake.githubBroker)
|
||||
Object.freeze(parsed.incomingWake.githubBroker);
|
||||
Object.freeze(parsed.incomingWake);
|
||||
}
|
||||
return Object.freeze(parsed);
|
||||
}
|
||||
|
||||
declare const verified: unique symbol;
|
||||
export type VerifiedIdleCapabilities = IdleCapabilities & {
|
||||
readonly [verified]: true;
|
||||
};
|
||||
@@ -0,0 +1,184 @@
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import {
|
||||
readIdleIncomingWakeProof,
|
||||
incomingWakeFingerprint,
|
||||
} from "./idle-incoming-wake.js";
|
||||
import { registerConnectionEventWake } from "./connection-event-wake.js";
|
||||
import type { VerifiedIdleCapabilities } from "./idle-incoming-wake-capability.js";
|
||||
|
||||
const ownerId = "00000000-0000-4000-8000-000000000001";
|
||||
function fixture(count: number) {
|
||||
const endpoints = Array.from({ length: count }, (_, i) => ({
|
||||
endpoint: {
|
||||
id: `endpoint-${i}`,
|
||||
companyId: "company",
|
||||
status: "active",
|
||||
activatedAt: new Date(0),
|
||||
botExternalId: `bot-${i}`,
|
||||
providerAccountId: `app-${i}`,
|
||||
setup: {
|
||||
step: "complete",
|
||||
github: {
|
||||
cloudRegistrationId: String(i).padStart(43, "a"),
|
||||
registrationStatus: "completed",
|
||||
},
|
||||
},
|
||||
},
|
||||
connection: { id: `connection-${i}` },
|
||||
}));
|
||||
const registrations = endpoints.map(({ endpoint }) => ({
|
||||
endpointId: endpoint.id,
|
||||
companyId: endpoint.companyId,
|
||||
userId: "owner",
|
||||
cloudId: endpoint.setup.github.cloudRegistrationId,
|
||||
}));
|
||||
const tuples = registrations.map((row) => [
|
||||
`github-app:${row.cloudId}`,
|
||||
row.companyId,
|
||||
row.userId,
|
||||
"github.bot",
|
||||
`gateway:${row.cloudId}`,
|
||||
row.cloudId,
|
||||
row.cloudId,
|
||||
]);
|
||||
const capability = {
|
||||
incomingWake: {
|
||||
version: 1,
|
||||
ownerId,
|
||||
generation: "a".repeat(64),
|
||||
routes: ["github-broker-events-v1"],
|
||||
githubBroker: {
|
||||
instanceId: "fixture-instance",
|
||||
bindingsFingerprint: incomingWakeFingerprint(tuples),
|
||||
},
|
||||
},
|
||||
} as VerifiedIdleCapabilities;
|
||||
return { endpoints, registrations, capability };
|
||||
}
|
||||
/** Named DB boundary for cardinality/elapsed guards; PostgreSQL fixture covers real SQL separately. */
|
||||
function database(replies: Array<() => unknown[]>) {
|
||||
let index = 0;
|
||||
const select = vi.fn(() => {
|
||||
const reply = replies[index++];
|
||||
const query = {
|
||||
from: () => query,
|
||||
innerJoin: () => query,
|
||||
where: () => query,
|
||||
limit: async () => reply?.() ?? [],
|
||||
};
|
||||
return query;
|
||||
});
|
||||
const execute = vi.fn(async () => []);
|
||||
return {
|
||||
db: { select, execute } as unknown as Pick<Db, "select" | "execute">,
|
||||
select,
|
||||
execute,
|
||||
};
|
||||
}
|
||||
const stops: Array<() => void> = [];
|
||||
function input(
|
||||
db: object,
|
||||
capabilities: VerifiedIdleCapabilities,
|
||||
now?: () => number,
|
||||
) {
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_INSTANCE_ID", "fixture-instance");
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_SIGN_PRIVATE_KEY", "fixture");
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_SEAL_PRIVATE_KEY", "fixture");
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_ENVIRONMENT", "test");
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_BASE_URL", "http://127.0.0.1");
|
||||
stops.push(registerConnectionEventWake(db, () => {}));
|
||||
return { ownerId, receiverOwner: db, capabilities, now };
|
||||
}
|
||||
afterEach(() => {
|
||||
for (const stop of stops.splice(0)) stop();
|
||||
vi.unstubAllEnvs();
|
||||
});
|
||||
|
||||
describe("bounded incoming wake owner projection", () => {
|
||||
it("uses three bounded selects for a thousand bots and retains exact per-bot authority", async () => {
|
||||
const f = fixture(1000),
|
||||
sql = database([() => [], () => f.endpoints, () => f.registrations]);
|
||||
const result = await readIdleIncomingWakeProof(
|
||||
sql.db,
|
||||
input(sql.db, f.capability),
|
||||
);
|
||||
expect(sql.select).toHaveBeenCalledTimes(3);
|
||||
expect(sql.execute).toHaveBeenCalledTimes(4);
|
||||
expect(result?.endpointIds).toHaveLength(1000);
|
||||
expect(result?.connectionIds).toHaveLength(1000);
|
||||
});
|
||||
it.each(["duplicate", "foreign", "missing"])(
|
||||
"rejects %s registration authority",
|
||||
async (fault) => {
|
||||
const f = fixture(1);
|
||||
let registrations = f.registrations;
|
||||
if (fault === "duplicate")
|
||||
registrations = [...registrations, ...registrations];
|
||||
if (fault === "foreign")
|
||||
registrations = registrations.map((r) => ({
|
||||
...r,
|
||||
companyId: "foreign",
|
||||
}));
|
||||
if (fault === "missing") registrations = [];
|
||||
const sql = database([() => [], () => f.endpoints, () => registrations]);
|
||||
await expect(
|
||||
readIdleIncomingWakeProof(sql.db, input(sql.db, f.capability)),
|
||||
).rejects.toThrow("registration authority");
|
||||
},
|
||||
);
|
||||
it("stops the complete inspection when its monotonic budget expires before another query", async () => {
|
||||
const f = fixture(1);
|
||||
let now = 0;
|
||||
const sql = database([
|
||||
() => {
|
||||
now = 2001;
|
||||
return [];
|
||||
},
|
||||
() => f.endpoints,
|
||||
() => f.registrations,
|
||||
]);
|
||||
await expect(
|
||||
readIdleIncomingWakeProof(
|
||||
sql.db,
|
||||
input(sql.db, f.capability, () => now),
|
||||
),
|
||||
).rejects.toThrow("budget exhausted");
|
||||
expect(sql.select).toHaveBeenCalledTimes(1);
|
||||
expect(sql.execute).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
it("fails closed at row limits before expanding work or issuing registration queries", async () => {
|
||||
const f = fixture(1001),
|
||||
sql = database([() => [], () => f.endpoints, () => f.registrations]);
|
||||
await expect(
|
||||
readIdleIncomingWakeProof(sql.db, input(sql.db, f.capability)),
|
||||
).rejects.toThrow("Too many GitHub endpoints");
|
||||
expect(sql.select).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe("bounded API admission projection", () => {
|
||||
const api = { incomingWake: { version: 1, ownerId, generation: "a".repeat(64), routes: ["tenant-api-http-v2"],
|
||||
apiKeyProof: { version: 1, challenge: "c".repeat(64) } } } as VerifiedIdleCapabilities;
|
||||
it.each(["overflow", "duplicate", "malformed"])("refuses complete eligibility for %s key state", async kind => {
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_STACK_ID", "fixture-stack");
|
||||
const row = { keyHash: "a".repeat(64), expiresAt: null };
|
||||
const rows = kind === "overflow" ? Array.from({ length: 1001 }, () => row)
|
||||
: kind === "duplicate" ? [row, row] : [{ keyHash: "not-a-digest", expiresAt: null }];
|
||||
const execute = vi.fn().mockResolvedValueOnce([]).mockResolvedValueOnce(rows);
|
||||
await expect(readIdleIncomingWakeProof({ execute } as unknown as Pick<Db, "select" | "execute">,
|
||||
{ ownerId, receiverOwner: {}, capabilities: api })).rejects.toThrow();
|
||||
expect(execute).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
it("has an explicit complete empty-key receipt and shares the inspection time budget", async () => {
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_STACK_ID", "fixture-stack");
|
||||
const execute = vi.fn(async () => []);
|
||||
const proof = await readIdleIncomingWakeProof({ execute } as unknown as Pick<Db, "select" | "execute">,
|
||||
{ ownerId, receiverOwner: {}, capabilities: api });
|
||||
expect(proof?.apiWakeAdmission?.keys).toEqual([]);
|
||||
let clock = 0;
|
||||
const slow = vi.fn(async () => { clock += 1100; return []; });
|
||||
await expect(readIdleIncomingWakeProof({ execute: slow } as unknown as Pick<Db, "select" | "execute">,
|
||||
{ ownerId, receiverOwner: {}, capabilities: api, now: () => clock })).rejects.toThrow(/budget exhausted/);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,359 @@
|
||||
import { apiWakeFingerprint, createApiWakeAdmission, type ApiWakeAdmission } from "./idle-api-wake-admission.js";
|
||||
import { createHash } from "node:crypto";
|
||||
import { and, eq, inArray, sql, type SQL } from "drizzle-orm";
|
||||
import {
|
||||
chatEndpoints,
|
||||
chatGitHubRegistrations,
|
||||
connectionGrants,
|
||||
toolConnections,
|
||||
type Db,
|
||||
} from "@paperclipai/db";
|
||||
import type {
|
||||
IdleIncomingWakeCapability,
|
||||
VerifiedIdleCapabilities,
|
||||
} from "./idle-incoming-wake-capability.js";
|
||||
import { githubConnectionEventBindings } from "./github-connection-events.js";
|
||||
import { paperclipCloudConnectorConfigFromEnv } from "./paperclip-cloud-connector.js";
|
||||
import { hasConnectionEventWakeReceiver } from "./connection-event-wake.js";
|
||||
|
||||
export interface IdleIncomingWakeProof {
|
||||
capability: IdleIncomingWakeCapability;
|
||||
apiWakeAdmission?: ApiWakeAdmission;
|
||||
connectionIds: string[];
|
||||
endpointIds: string[];
|
||||
/** A broker enrollment cannot discharge these credential continuity obligations. */
|
||||
requiredContinuityConnectionIds: string[];
|
||||
}
|
||||
export const incomingWakeFingerprint = (bindings: string[][]) =>
|
||||
createHash("sha256")
|
||||
.update(
|
||||
JSON.stringify(
|
||||
bindings
|
||||
.map((b) => [...b])
|
||||
.sort((a, b) =>
|
||||
JSON.stringify(a) < JSON.stringify(b)
|
||||
? -1
|
||||
: JSON.stringify(a) > JSON.stringify(b)
|
||||
? 1
|
||||
: 0,
|
||||
),
|
||||
),
|
||||
)
|
||||
.digest("hex");
|
||||
|
||||
/** Read current binding authority in the safety transaction. No network call, renewal or write. */
|
||||
export async function readIdleIncomingWakeProof(
|
||||
tx: Pick<Db, "select" | "execute">,
|
||||
input: {
|
||||
capabilities?: VerifiedIdleCapabilities;
|
||||
ownerId: string;
|
||||
receiverOwner: object;
|
||||
now?: () => number;
|
||||
},
|
||||
): Promise<IdleIncomingWakeProof | undefined> {
|
||||
const capability = input.capabilities?.incomingWake;
|
||||
if (!capability) return;
|
||||
if (capability.ownerId !== input.ownerId)
|
||||
throw new Error("Incoming wake owner changed");
|
||||
const proof: IdleIncomingWakeProof = {
|
||||
capability,
|
||||
connectionIds: [],
|
||||
endpointIds: [],
|
||||
requiredContinuityConnectionIds: [],
|
||||
};
|
||||
const monotonic = input.now ?? (() => performance.now());
|
||||
const started = monotonic();
|
||||
const remaining = () => {
|
||||
const ms = 2_000 - (monotonic() - started);
|
||||
if (!Number.isFinite(ms) || ms <= 0)
|
||||
throw new Error("Incoming wake inspection budget exhausted");
|
||||
return Math.max(1, Math.floor(ms));
|
||||
};
|
||||
const beforeRead = async () => {
|
||||
// One shared owner budget, not five seconds multiplied by the row count.
|
||||
await tx.execute(
|
||||
sql`SELECT set_config('statement_timeout', ${`${remaining()}ms`}, true)`,
|
||||
);
|
||||
remaining();
|
||||
};
|
||||
if (capability.routes.includes("tenant-api-http-v2")) {
|
||||
const stackId = process.env.PAPERCLIP_CLOUD_STACK_ID;
|
||||
if (!stackId || !capability.apiKeyProof) throw new Error("API wake challenge is missing");
|
||||
await beforeRead();
|
||||
const rows = await tx.execute<{ keyHash: string; expiresAt: Date | null }>(sql`
|
||||
SELECT key_hash AS "keyHash", NULL::timestamptz AS "expiresAt" FROM agent_api_keys WHERE revoked_at IS NULL
|
||||
UNION ALL SELECT key_hash AS "keyHash", expires_at AS "expiresAt" FROM board_api_keys
|
||||
WHERE revoked_at IS NULL AND (expires_at IS NULL OR expires_at > now()) LIMIT 1001`);
|
||||
remaining();
|
||||
if (rows.length > 1000) throw new Error("Too many API keys to certify wake admission");
|
||||
proof.apiWakeAdmission = createApiWakeAdmission({ stackId, ownerId: input.ownerId,
|
||||
capabilityGeneration: capability.generation, challenge: capability.apiKeyProof.challenge,
|
||||
keys: rows.map(row => ({ fingerprint: apiWakeFingerprint({ stackId, ownerId: input.ownerId,
|
||||
challenge: capability.apiKeyProof!.challenge, keyHash: row.keyHash }),
|
||||
expiresAt: row.expiresAt === null ? null : new Date(row.expiresAt).toISOString() })),
|
||||
});
|
||||
}
|
||||
if (!capability.routes.includes("github-broker-events-v1")) {
|
||||
await tx.execute(sql`SET LOCAL statement_timeout = '5s'`);
|
||||
remaining();
|
||||
return proof;
|
||||
}
|
||||
const broker = capability.githubBroker;
|
||||
const config = paperclipCloudConnectorConfigFromEnv();
|
||||
if (
|
||||
!broker ||
|
||||
config?.instanceId !== broker.instanceId ||
|
||||
!hasConnectionEventWakeReceiver(input.receiverOwner)
|
||||
)
|
||||
throw new Error("Incoming GitHub receiver is not bound");
|
||||
await beforeRead();
|
||||
const rows = await tx
|
||||
.select({ grant: connectionGrants, connection: toolConnections })
|
||||
.from(connectionGrants)
|
||||
.innerJoin(
|
||||
toolConnections,
|
||||
and(
|
||||
eq(toolConnections.id, connectionGrants.connectionId),
|
||||
eq(toolConnections.companyId, connectionGrants.companyId),
|
||||
),
|
||||
)
|
||||
.where(
|
||||
and(
|
||||
eq(connectionGrants.status, "active"),
|
||||
eq(toolConnections.status, "active"),
|
||||
eq(toolConnections.enabled, true),
|
||||
sql`${toolConnections.config}->>'sourceTemplateKey' = 'github'`,
|
||||
sql`${toolConnections.config}->'oauth'->>'connectorProfile' = 'github.code'`,
|
||||
sql`exists (select 1 from companies where id=${connectionGrants.companyId} and status <> 'archived')`,
|
||||
),
|
||||
)
|
||||
.limit(1001);
|
||||
if (rows.length > 1000)
|
||||
throw new Error("Too many GitHub bindings to inspect");
|
||||
remaining();
|
||||
let bindingCount = 0;
|
||||
for (const { grant, connection } of rows) {
|
||||
const github = grant.providerTenant?.github;
|
||||
if (
|
||||
!github ||
|
||||
!Array.isArray(github.installationIds) ||
|
||||
!github.installationIds.length ||
|
||||
github.installationIds.length > 1000 ||
|
||||
github.bindingSync ||
|
||||
github.continuityRetry ||
|
||||
github.installationIds.some(
|
||||
(id) => typeof id !== "string" || !/^[0-9]+$/.test(id),
|
||||
)
|
||||
)
|
||||
throw new Error("Unsettled GitHub authority");
|
||||
bindingCount += github.installationIds.length;
|
||||
if (bindingCount > 1000)
|
||||
throw new Error("Too many GitHub bindings to inspect");
|
||||
proof.requiredContinuityConnectionIds.push(connection.id);
|
||||
}
|
||||
const bindings = githubConnectionEventBindings(rows);
|
||||
if (
|
||||
bindings.length !==
|
||||
rows.reduce(
|
||||
(n, r) => n + r.grant.providerTenant!.github!.installationIds.length,
|
||||
0,
|
||||
)
|
||||
)
|
||||
throw new Error("Incomplete GitHub binding");
|
||||
const tuples = bindings.map((b) => [
|
||||
b.id,
|
||||
b.companyId,
|
||||
b.subject,
|
||||
"github.code",
|
||||
b.installationId,
|
||||
b.connectionId,
|
||||
b.grantId,
|
||||
]);
|
||||
await beforeRead();
|
||||
const endpoints = await tx
|
||||
.select({ endpoint: chatEndpoints, connection: toolConnections })
|
||||
.from(chatEndpoints)
|
||||
.innerJoin(
|
||||
toolConnections,
|
||||
and(
|
||||
eq(toolConnections.id, chatEndpoints.connectionId),
|
||||
eq(toolConnections.companyId, chatEndpoints.companyId),
|
||||
),
|
||||
)
|
||||
.where(
|
||||
and(
|
||||
eq(chatEndpoints.provider, "github"),
|
||||
eq(chatEndpoints.status, "active"),
|
||||
eq(toolConnections.status, "active"),
|
||||
eq(toolConnections.enabled, true),
|
||||
eq(toolConnections.transport, "chat_sdk"),
|
||||
eq(toolConnections.connectionPurpose, "channel"),
|
||||
sql`exists (select 1 from companies where id=${chatEndpoints.companyId} and status <> 'archived')`,
|
||||
),
|
||||
)
|
||||
.limit(1001);
|
||||
if (endpoints.length > 1000)
|
||||
throw new Error("Too many GitHub endpoints to inspect");
|
||||
remaining();
|
||||
const brokerEndpointIds = endpoints
|
||||
.filter(
|
||||
({ endpoint }) =>
|
||||
typeof endpoint.setup.github?.cloudRegistrationId === "string",
|
||||
)
|
||||
.map(({ endpoint }) => endpoint.id);
|
||||
const registrationUsers = new Map<string, string[]>();
|
||||
if (brokerEndpointIds.length) {
|
||||
await beforeRead();
|
||||
const registrations = await tx
|
||||
.select({
|
||||
endpointId: chatGitHubRegistrations.endpointId,
|
||||
companyId: chatGitHubRegistrations.companyId,
|
||||
userId: chatGitHubRegistrations.userId,
|
||||
cloudId: sql<string>`${chatGitHubRegistrations.handoff}->>'cloudId'`,
|
||||
})
|
||||
.from(chatGitHubRegistrations)
|
||||
.where(
|
||||
and(
|
||||
inArray(chatGitHubRegistrations.endpointId, brokerEndpointIds),
|
||||
eq(chatGitHubRegistrations.status, "completed"),
|
||||
),
|
||||
)
|
||||
.limit(10001);
|
||||
remaining();
|
||||
if (registrations.length > 10000)
|
||||
throw new Error("Too many GitHub registrations to inspect");
|
||||
for (const registration of registrations) {
|
||||
const key = JSON.stringify([
|
||||
registration.endpointId,
|
||||
registration.companyId,
|
||||
registration.cloudId,
|
||||
]);
|
||||
const users = registrationUsers.get(key) ?? [];
|
||||
users.push(registration.userId);
|
||||
registrationUsers.set(key, users);
|
||||
}
|
||||
}
|
||||
for (const { endpoint, connection } of endpoints) {
|
||||
const registration = endpoint.setup.github?.cloudRegistrationId;
|
||||
if (!registration || !/^[A-Za-z0-9_-]{43}$/.test(registration)) continue;
|
||||
if (
|
||||
endpoint.setup.step !== "complete" ||
|
||||
endpoint.setup.github?.registrationStatus !== "completed" ||
|
||||
endpoint.setup.github.initialSetupPending ||
|
||||
endpoint.setup.github.initialRepositoryImportPending ||
|
||||
!endpoint.activatedAt ||
|
||||
!endpoint.botExternalId ||
|
||||
!endpoint.providerAccountId
|
||||
)
|
||||
throw new Error("GitHub bot setup has unfinished work");
|
||||
const users = registrationUsers.get(
|
||||
JSON.stringify([endpoint.id, endpoint.companyId, registration]),
|
||||
);
|
||||
if (users?.length !== 1 || !users[0])
|
||||
throw new Error("GitHub registration authority changed");
|
||||
tuples.push([
|
||||
`github-app:${registration}`,
|
||||
endpoint.companyId,
|
||||
users[0],
|
||||
"github.bot",
|
||||
`gateway:${registration}`,
|
||||
registration,
|
||||
registration,
|
||||
]);
|
||||
proof.connectionIds.push(connection.id);
|
||||
proof.endpointIds.push(endpoint.id);
|
||||
}
|
||||
remaining();
|
||||
if (tuples.length > 1000)
|
||||
throw new Error("Too many GitHub bindings to inspect");
|
||||
if (incomingWakeFingerprint(tuples) !== broker.bindingsFingerprint)
|
||||
throw new Error("GitHub wake bindings changed");
|
||||
remaining();
|
||||
await tx.execute(sql`SET LOCAL statement_timeout = '5s'`);
|
||||
remaining();
|
||||
return proof;
|
||||
}
|
||||
|
||||
/** Exact domain predicates selected by stable inventory reason, never SQL-text identity. */
|
||||
export function incomingWakeWorkQuery(
|
||||
reason: string,
|
||||
proof?: IdleIncomingWakeProof,
|
||||
continuityConnectionIds: readonly string[] = [],
|
||||
): SQL | undefined {
|
||||
const connectionIds = [
|
||||
...(proof?.connectionIds ?? []),
|
||||
...(proof?.requiredContinuityConnectionIds.filter((id) =>
|
||||
continuityConnectionIds.includes(id),
|
||||
) ?? []),
|
||||
];
|
||||
if (reason === "agent_api_keys")
|
||||
return proof?.apiWakeAdmission !== undefined
|
||||
? sql`SELECT 1 WHERE false`
|
||||
: sql`SELECT 1 FROM agent_api_keys WHERE revoked_at IS NULL`;
|
||||
if (reason === "board_api_keys")
|
||||
return proof?.apiWakeAdmission !== undefined
|
||||
? sql`SELECT 1 WHERE false`
|
||||
: sql`SELECT 1 FROM board_api_keys WHERE revoked_at IS NULL
|
||||
AND (expires_at IS NULL OR expires_at > now())`;
|
||||
if (reason === "chat_endpoints") {
|
||||
const ids = proof?.endpointIds ?? [];
|
||||
// Only the supported GitHub webhook path gains a removed-history proof.
|
||||
// Archival commits before cleanup, so status alone cannot discharge work.
|
||||
return sql`SELECT 1 FROM chat_endpoints endpoint WHERE (((endpoint.provider = 'github'
|
||||
AND endpoint.status = 'archived' AND endpoint.archived_at IS NOT NULL
|
||||
AND jsonb_typeof(endpoint.setup) = 'object'
|
||||
AND CASE WHEN jsonb_typeof(endpoint.setup->'runtimeGeneration') = 'number'
|
||||
THEN (endpoint.setup->>'runtimeGeneration')::numeric >= 1
|
||||
AND trunc((endpoint.setup->>'runtimeGeneration')::numeric) = (endpoint.setup->>'runtimeGeneration')::numeric
|
||||
ELSE false END
|
||||
AND EXISTS (SELECT 1 FROM tool_connections removed_connection
|
||||
WHERE removed_connection.id = endpoint.connection_id AND removed_connection.company_id = endpoint.company_id
|
||||
AND removed_connection.connection_purpose = 'channel' AND removed_connection.transport = 'chat_sdk'
|
||||
AND removed_connection.status = 'archived' AND removed_connection.enabled = false
|
||||
AND removed_connection.credential_secret_refs = '[]'::jsonb)
|
||||
AND NOT EXISTS (SELECT 1 FROM chat_endpoint_leases lease WHERE lease.endpoint_id = endpoint.id)
|
||||
AND NOT EXISTS (SELECT 1 FROM chat_slack_registrations registration WHERE registration.endpoint_id = endpoint.id)
|
||||
AND NOT EXISTS (SELECT 1 FROM chat_github_registrations registration WHERE registration.endpoint_id = endpoint.id
|
||||
AND (registration.company_id = endpoint.company_id AND registration.status = 'completed'
|
||||
AND registration.consumed_at IS NOT NULL) IS NOT TRUE))
|
||||
OR (endpoint.provider = 'slack'
|
||||
AND endpoint.status = 'draft' AND endpoint.setup->>'step' = 'provider_setup'
|
||||
AND (endpoint.setup - ARRAY['step', 'command']) = '{}'::jsonb
|
||||
AND (endpoint.setup->'command' IS NULL OR endpoint.setup->'command' = 'null'::jsonb
|
||||
OR jsonb_typeof(endpoint.setup->'command') = 'string')
|
||||
AND endpoint.activated_at IS NULL AND endpoint.provider_account_id IS NULL
|
||||
AND endpoint.bot_external_id IS NULL AND endpoint.bot_username IS NULL
|
||||
AND endpoint.last_event_at IS NULL AND endpoint.last_publication_at IS NULL
|
||||
AND endpoint.health_message IS NULL AND endpoint.last_error IS NULL
|
||||
AND NOT EXISTS (SELECT 1 FROM chat_slack_registrations registration WHERE registration.endpoint_id = endpoint.id)
|
||||
AND EXISTS (SELECT 1 FROM tool_connections draft_connection
|
||||
WHERE draft_connection.id = endpoint.connection_id AND draft_connection.company_id = endpoint.company_id
|
||||
AND draft_connection.connection_purpose = 'channel' AND draft_connection.transport = 'chat_sdk'
|
||||
AND draft_connection.status = 'draft' AND draft_connection.enabled = false))) IS NOT TRUE) AND ${
|
||||
ids.length
|
||||
? sql`endpoint.id NOT IN (${sql.join(
|
||||
ids.map((id) => sql`${id}::uuid`),
|
||||
sql`, `,
|
||||
)})`
|
||||
: sql`true`
|
||||
}`;
|
||||
}
|
||||
if (reason === "tool_connections") {
|
||||
const ids = connectionIds;
|
||||
return sql`SELECT 1 FROM tool_connections c WHERE (c.status <> 'archived' AND (
|
||||
(c.status = 'draft' AND c.enabled = false)
|
||||
OR (c.status = 'active' AND c.connection_purpose = 'ai' AND c.transport = 'runtime_auth')
|
||||
OR (c.status = 'active' AND c.connection_purpose = 'tool' AND c.transport = 'mcp_remote'
|
||||
AND NOT EXISTS (SELECT 1 FROM connection_grants g WHERE g.connection_id = c.id
|
||||
AND (g.company_id <> c.company_id OR g.provider_tenant->'github' IS NOT NULL)))
|
||||
) IS NOT TRUE) AND ${
|
||||
ids.length
|
||||
? sql`c.id NOT IN (${sql.join(
|
||||
ids.map((id) => sql`${id}::uuid`),
|
||||
sql`, `,
|
||||
)})`
|
||||
: sql`true`
|
||||
}`;
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import { idleApiActivity } from "./idle-api-activity.js";
|
||||
import { promises as fs } from "node:fs";
|
||||
import path from "node:path";
|
||||
import { resolvePaperclipInstanceRoot } from "../home-paths.js";
|
||||
@@ -83,7 +84,7 @@ export async function readIdleLocalWork(options: { backupCheckpoint?: boolean; d
|
||||
}
|
||||
let probes: IdleLocalWork;
|
||||
try {
|
||||
const local = [...localWorkProbes].map(probe => probe());
|
||||
const local = [idleApiActivity.active() ? "present" : "none", ...[...localWorkProbes].map(probe => probe())];
|
||||
probes = local.some(state => state !== "none" && state !== "present") ? "unknown"
|
||||
: local.includes("present") ? "present" : "none";
|
||||
} catch { probes = "unknown"; }
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
import type { ApiWakeAdmission } from "./idle-api-wake-admission.js";
|
||||
import { readIdleIncomingWakeProof, incomingWakeWorkQuery, type IdleIncomingWakeProof } from "./idle-incoming-wake.js";
|
||||
import type { VerifiedIdleCapabilities, IdleIncomingWakeCapability } from "./idle-incoming-wake-capability.js";
|
||||
import { idleWorkSnapshot } from "./task-admission.js";
|
||||
import { readIdleLocalWork, type IdleLocalWork } from "./idle-local-work.js";
|
||||
import { readIdleRoutineSchedule, type ScheduledWake } from "./idle-scheduled-work.js";
|
||||
@@ -38,10 +41,12 @@ export interface IdleSleepDrainStatus {
|
||||
}
|
||||
|
||||
export interface IdleSleepSafety {
|
||||
version: 1 | 2;
|
||||
version: 1 | 2 | 3;
|
||||
backgroundWork: "none" | "present" | "unknown";
|
||||
scheduledWake?: ScheduledWake;
|
||||
diagnostics?: IdleSleepDiagnostics;
|
||||
incomingWake?: IdleIncomingWakeCapability;
|
||||
apiWakeAdmission?: ApiWakeAdmission;
|
||||
}
|
||||
|
||||
export type InspectIdlePlugins = (hold: { ownerId: string; expiresAt: number }) => Promise<{
|
||||
@@ -84,9 +89,8 @@ const WORK_CHECKS = [
|
||||
workCheck("prompt_injection_invocations", `SELECT 1 FROM prompt_injection_invocations WHERE status = 'running'`),
|
||||
// Persistent machine credentials can receive work without a human opening
|
||||
// the instance. They need an ingress wake contract before we can sleep.
|
||||
workCheck("agent_api_keys", `SELECT 1 FROM agent_api_keys WHERE revoked_at IS NULL`),
|
||||
workCheck("board_api_keys", `SELECT 1 FROM board_api_keys WHERE revoked_at IS NULL
|
||||
AND (expires_at IS NULL OR expires_at > now())`),
|
||||
workCheck("agent_api_keys", incomingWakeWorkQuery("agent_api_keys")!),
|
||||
workCheck("board_api_keys", incomingWakeWorkQuery("board_api_keys")!),
|
||||
// Coalescing consumes a request into another run. Retain it as work until
|
||||
// both the request and that exact company/agent's run have finished.
|
||||
workCheck("agent_wakeup_requests", `SELECT 1 FROM agent_wakeup_requests w WHERE
|
||||
@@ -137,20 +141,7 @@ const WORK_CHECKS = [
|
||||
// An unconfigured Slack draft has no provider identity or inbound runtime.
|
||||
// Setup still wakes through normal browser admission. Retain every enabled,
|
||||
// previously active, partially connected, foreign, or unsupported endpoint.
|
||||
workCheck("chat_endpoints", `SELECT 1 FROM chat_endpoints endpoint WHERE (endpoint.provider = 'slack'
|
||||
AND endpoint.status = 'draft' AND endpoint.setup->>'step' = 'provider_setup'
|
||||
AND (endpoint.setup - ARRAY['step', 'command']) = '{}'::jsonb
|
||||
AND (endpoint.setup->'command' IS NULL OR endpoint.setup->'command' = 'null'::jsonb
|
||||
OR jsonb_typeof(endpoint.setup->'command') = 'string')
|
||||
AND endpoint.activated_at IS NULL AND endpoint.provider_account_id IS NULL
|
||||
AND endpoint.bot_external_id IS NULL AND endpoint.bot_username IS NULL
|
||||
AND endpoint.last_event_at IS NULL AND endpoint.last_publication_at IS NULL
|
||||
AND endpoint.health_message IS NULL AND endpoint.last_error IS NULL
|
||||
AND NOT EXISTS (SELECT 1 FROM chat_slack_registrations registration WHERE registration.endpoint_id = endpoint.id)
|
||||
AND EXISTS (SELECT 1 FROM tool_connections draft_connection
|
||||
WHERE draft_connection.id = endpoint.connection_id AND draft_connection.company_id = endpoint.company_id
|
||||
AND draft_connection.connection_purpose = 'channel' AND draft_connection.transport = 'chat_sdk'
|
||||
AND draft_connection.status = 'draft' AND draft_connection.enabled = false)) IS NOT TRUE`),
|
||||
workCheck("chat_endpoints", incomingWakeWorkQuery("chat_endpoints")!),
|
||||
workCheck("routines", `SELECT 1 FROM routines WHERE status NOT IN ('paused', 'archived')`),
|
||||
workCheck("routine_runs", `SELECT 1 FROM routine_runs WHERE status NOT IN ('completed', 'failed', 'skipped', 'cancelled')`),
|
||||
workCheck("plugin_jobs", `SELECT 1 FROM plugin_jobs WHERE status <> 'paused'`),
|
||||
@@ -159,13 +150,7 @@ const WORK_CHECKS = [
|
||||
// Saved outbound MCP configuration and disabled drafts have no independent
|
||||
// listener. Keep local processes, channels and GitHub webhook bindings awake;
|
||||
// accepted calls, credential recovery and delivery work have separate fences.
|
||||
workCheck("tool_connections", `SELECT 1 FROM tool_connections c WHERE c.status <> 'archived' AND (
|
||||
(c.status = 'draft' AND c.enabled = false)
|
||||
OR (c.status = 'active' AND c.connection_purpose = 'ai' AND c.transport = 'runtime_auth')
|
||||
OR (c.status = 'active' AND c.connection_purpose = 'tool' AND c.transport = 'mcp_remote'
|
||||
AND NOT EXISTS (SELECT 1 FROM connection_grants g WHERE g.connection_id = c.id
|
||||
AND (g.company_id <> c.company_id OR g.provider_tenant->'github' IS NOT NULL)))
|
||||
) IS NOT TRUE`),
|
||||
workCheck("tool_connections", incomingWakeWorkQuery("tool_connections")!),
|
||||
// An explicit board escalation has no automatic retry. Retain the visible
|
||||
// action and all execution/cleanup checks while the board decides what to do.
|
||||
workCheck("issue_recovery_actions", `SELECT 1 FROM issue_recovery_actions a WHERE a.status NOT IN ('resolved', 'cancelled')
|
||||
@@ -294,6 +279,8 @@ function sameQuietHold(before: IdleSleepDrainStatus, after: IdleSleepDrainStatus
|
||||
* Only a caller holding the exact idle owner may receive an empty report.
|
||||
*/
|
||||
export interface IdleSleepSafetyOptions {
|
||||
incomingWake?: VerifiedIdleCapabilities;
|
||||
deadlineCapability?: VerifiedIdleCapabilities;
|
||||
collectDiagnostics?: true;
|
||||
}
|
||||
|
||||
@@ -310,12 +297,15 @@ export async function readIdleSleepSafety(
|
||||
scheduledIssueMonitors = false,
|
||||
options: IdleSleepSafetyOptions = {},
|
||||
): Promise<IdleSleepSafety> {
|
||||
let incomingProof: IdleIncomingWakeProof | undefined;
|
||||
const diagnostic = options.collectDiagnostics ? createIdleSleepDiagnostics() : undefined;
|
||||
const report = (backgroundWork: IdleLocalWork, scheduledWake?: ScheduledWake): IdleSleepSafety => {
|
||||
const diagnostics = diagnostic?.report();
|
||||
if (backgroundWork === "none" && diagnostics?.checks.some(check =>
|
||||
check.state === "unknown" || check.state === "not_checked")) backgroundWork = "unknown";
|
||||
return { version: backgroundWork === "none" && scheduledWake ? 2 : 1, backgroundWork,
|
||||
return { version: backgroundWork === "none" && incomingProof ? 3 : backgroundWork === "none" && scheduledWake ? 2 : 1, backgroundWork,
|
||||
...(backgroundWork === "none" && incomingProof ? { incomingWake: incomingProof.capability,
|
||||
...(incomingProof.apiWakeAdmission ? { apiWakeAdmission: incomingProof.apiWakeAdmission } : {}) } : {}),
|
||||
...(backgroundWork === "none" && scheduledWake ? { scheduledWake } : {}),
|
||||
...(diagnostics ? { diagnostics } : {}) };
|
||||
};
|
||||
@@ -360,9 +350,14 @@ export async function readIdleSleepSafety(
|
||||
let scheduledWake: ScheduledWake | undefined;
|
||||
const blocked = await db.transaction(async (tx) => {
|
||||
await tx.execute(sql`SET LOCAL statement_timeout = '5s'`);
|
||||
if (options.incomingWake?.incomingWake) {
|
||||
if (!ownerId || ownerId !== before.ownerId) return undefined;
|
||||
incomingProof = await readIdleIncomingWakeProof(tx, { capabilities: options.incomingWake, ownerId, receiverOwner: db });
|
||||
}
|
||||
const checks = WORK_CHECKS.filter(check =>
|
||||
(!scheduledRoutines || check.reason !== "routines")
|
||||
&& (!scheduledIssueMonitors || check.reason !== "issues"));
|
||||
&& (!scheduledIssueMonitors || check.reason !== "issues"))
|
||||
.map(check => ({ ...check, query: incomingWakeWorkQuery(check.reason, incomingProof) ?? check.query }));
|
||||
checks.push(workCheck("agents.company_hold", sql`SELECT 1 FROM agents JOIN companies ON companies.id = agents.company_id
|
||||
WHERE ${companyHoldMismatch()}`));
|
||||
checks.push(workCheck("heartbeat_runs.budget_cancellation", sql`SELECT 1 FROM heartbeat_runs WHERE ${pendingBudgetRunCancellationPredicate()}`));
|
||||
@@ -447,6 +442,9 @@ export async function readIdleSleepSafety(
|
||||
diagnostic?.record("inspection.hold", "unknown");
|
||||
return report("present");
|
||||
}
|
||||
// Legacy alarm kinds cannot be silently combined with a new ingress proof.
|
||||
// The negotiated deadline catalog supplies the composed v3 alarm separately.
|
||||
if (incomingProof && scheduledWake) return report("unknown");
|
||||
return report(local, scheduledWake);
|
||||
} catch {
|
||||
// Missing migrations, unknown state, malformed configuration, timeouts and
|
||||
|
||||
Reference in new issue
Block a user