diff --git a/tests/runner-e2e/chat-catalog-descriptors.json b/tests/runner-e2e/chat-catalog-descriptors.json index 4b385befc09..07eb133aef7 100644 --- a/tests/runner-e2e/chat-catalog-descriptors.json +++ b/tests/runner-e2e/chat-catalog-descriptors.json @@ -808,6 +808,7 @@ "chat-catalog-domains/external-app-review.ts", "chat-catalog-domains/external-app-review-outcomes.ts", "chat-catalog-domains/external-app-review-probe.ts", + "chat-catalog-domains/external-app-review-diagnostic.ts", "../../packages/shared/src/validators/issue.ts", "chat-catalog-domains/external-provider.ts", "chat-catalog-domains/identity-support.ts", diff --git a/tests/runner-e2e/chat-catalog-domains/external-app-review-diagnostic.test.ts b/tests/runner-e2e/chat-catalog-domains/external-app-review-diagnostic.test.ts new file mode 100644 index 00000000000..98c8fccd3d7 --- /dev/null +++ b/tests/runner-e2e/chat-catalog-domains/external-app-review-diagnostic.test.ts @@ -0,0 +1,96 @@ +import { randomUUID } from "node:crypto"; +import { mkdtemp, writeFile, rm } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { createRequire } from "node:module"; +const { eq }: typeof import("../../../packages/db/node_modules/drizzle-orm/index.js") = + createRequire(new URL("../../../packages/db/package.json", import.meta.url))("drizzle-orm"); +import { afterAll, beforeAll, expect, it } from "vitest"; +import { createDb, agents, companies, issues, heartbeatRuns, issueThreadInteractions, toolInvocations, + toolActionRequests, toolActionDeliveries, agentWakeupRequests } from "../../../packages/db/src/index.js"; +import { startEmbeddedPostgresTestDatabase } from "../../../server/src/__tests__/helpers/embedded-postgres.js"; +import { summarizeToolValue } from "../../../server/src/services/tool-content-guards.js"; +import { observeReviewAdmission, readReviewAdmissionSnapshot, reviewAdmissionSummary, type ReviewAdmissionBinding } from "./external-app-review-diagnostic.js"; + +let database: Awaited>, db: ReturnType; +beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("review-admission-observer-"); db = createDb(database.connectionString); }, 30000); +afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); }); +const privateCanary = "synthetic-private-content-never-export"; +async function seed(wrongSourceIssue = false): Promise { + const b = Object.fromEntries(["companyId", "issueId", "agentId", "actionRequestId", "invocationId", "sourceRunId", "interactionId"].map(key => [key, randomUUID()])) as unknown as ReviewAdmissionBinding; + await db.insert(companies).values({ id: b.companyId, name: "Read-only review observer", issuePrefix: b.companyId.slice(0, 8).toUpperCase() }); + await db.insert(agents).values({ id: b.agentId, companyId: b.companyId, name: "Observer fixture", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: b.issueId, companyId: b.companyId, title: privateCanary, assigneeAgentId: b.agentId, status: "in_review" }); + const sourceIssueId = wrongSourceIssue ? randomUUID() : b.issueId; + if (wrongSourceIssue) await db.insert(issues).values({ id: sourceIssueId, companyId: b.companyId, title: "Unrelated issue", assigneeAgentId: b.agentId }); + await db.insert(heartbeatRuns).values({ id: b.sourceRunId, companyId: b.companyId, agentId: b.agentId, + issueId: sourceIssueId, status: "succeeded", runtimeMode: "native", nativePhase: "committed", contextSnapshot: { issueId: sourceIssueId, private: privateCanary } }); + await db.insert(issueThreadInteractions).values({ id: b.interactionId, companyId: b.companyId, issueId: b.issueId, + sourceRunId: b.sourceRunId, kind: "request_confirmation", status: "accepted", payload: { version: 1, prompt: privateCanary } }); + await db.insert(toolInvocations).values({ id: b.invocationId, companyId: b.companyId, agentId: b.agentId, + issueId: b.issueId, runId: b.sourceRunId, toolName: "controlled_write", status: "failed", errorMessage: privateCanary }); + await db.insert(toolActionRequests).values({ id: b.actionRequestId, companyId: b.companyId, issueId: b.issueId, + interactionId: b.interactionId, invocationId: b.invocationId, requestedByAgentId: b.agentId, status: "failed", + canonicalArgumentsHash: "a".repeat(64), canonicalArgumentsSummary: summarizeToolValue({ text: privateCanary }) }); + return b; +} + +it("observes missing/queued/native-running/settled states without writing or reading unrelated ownership", async () => { + const b = await seed(), foreign = await seed(); + await db.insert(agentWakeupRequests).values({ companyId: foreign.companyId, agentId: foreign.agentId, + source: "automation", status: "queued", payload: { toolActionRequestIds: [b.actionRequestId], private: privateCanary } }); + expect(await readReviewAdmissionSnapshot(db, { ...b, companyId: foreign.companyId })).toEqual({ available: false, reason: "binding_mismatch" }); + expect(await readReviewAdmissionSnapshot(db, { ...b, sourceRunId: foreign.sourceRunId })).toEqual({ available: false, reason: "binding_mismatch" }); + const wrongSource = await seed(true); + expect(await readReviewAdmissionSnapshot(db, wrongSource)).toEqual({ available: false, reason: "binding_mismatch" }); + expect(reviewAdmissionSummary(await readReviewAdmissionSnapshot(db, b))).toMatchObject({ delivery: "missing", wakes: 0, boundRuns: 0 }); + await db.insert(toolActionDeliveries).values({ actionRequestId: b.actionRequestId, companyId: b.companyId, issueId: b.issueId, interactionId: b.interactionId }); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: b.companyId, agentId: b.agentId, + source: "automation", status: "queued", coalescedCount: 2, payload: { toolActionRequestIds: [b.actionRequestId], private: privateCanary } }).returning(); + expect(reviewAdmissionSummary(await readReviewAdmissionSnapshot(db, b))).toMatchObject({ delivery: "pending", wakes: 1, wakeStates: { queued: 1 }, coalescedWakes: 1, boundRuns: 0 }); + const [run] = await db.insert(heartbeatRuns).values({ companyId: b.companyId, agentId: b.agentId, issueId: b.issueId, + wakeupRequestId: wake.id, status: "queued", runtimeMode: "native", contextSnapshot: { toolActionRequestIds: [b.actionRequestId], private: privateCanary } }).returning(); + await db.update(agentWakeupRequests).set({ runId: run.id }).where(eq(agentWakeupRequests.id, wake.id)); + expect(reviewAdmissionSummary(await readReviewAdmissionSnapshot(db, b))).toMatchObject({ boundRuns: 1, nativeRunStates: { queued: 1 } }); + await db.update(heartbeatRuns).set({ status: "running", nativePhase: "observed" }).where(eq(heartbeatRuns.id, run.id)); + expect(reviewAdmissionSummary(await readReviewAdmissionSnapshot(db, b))).toMatchObject({ nativeRunStates: { running: 1 } }); + await db.update(heartbeatRuns).set({ status: "succeeded", nativePhase: "committed" }).where(eq(heartbeatRuns.id, run.id)); + await db.update(toolActionDeliveries).set({ deliveredAt: new Date() }).where(eq(toolActionDeliveries.actionRequestId, b.actionRequestId)); + const before = await db.select().from(toolActionDeliveries).where(eq(toolActionDeliveries.actionRequestId, b.actionRequestId)); + const snapshot = await readReviewAdmissionSnapshot(db, b); + expect(reviewAdmissionSummary(snapshot)).toMatchObject({ delivery: "acknowledged", nativeRunStates: { succeeded: 1 } }); + expect(JSON.stringify(snapshot)).not.toContain(privateCanary); + expect(JSON.stringify(reviewAdmissionSummary(snapshot))).not.toContain(b.companyId); + expect(await db.select().from(toolActionDeliveries).where(eq(toolActionDeliveries.actionRequestId, b.actionRequestId))).toEqual(before); + await db.update(agentWakeupRequests).set({ status: privateCanary }).where(eq(agentWakeupRequests.id, wake.id)); + const closed = reviewAdmissionSummary(await readReviewAdmissionSnapshot(db, b)); + expect(closed).toMatchObject({ wakeStates: { other: 1 } }); + expect(JSON.stringify(closed)).not.toContain(privateCanary); + await db.insert(agentWakeupRequests).values(Array.from({ length: 8 }, () => ({ companyId: b.companyId, agentId: b.agentId, + source: "automation", payload: { toolActionRequestIds: [b.actionRequestId] } }))); + const bounded = await readReviewAdmissionSnapshot(db, b); + expect(bounded.available && bounded.wakes).toHaveLength(8); + expect(bounded).toMatchObject({ rowLimitReached: true }); +}); + +it("keeps absent owned configuration explicitly unavailable", async () => { + const b = await seed(); + expect(await observeReviewAdmission(b, {})).toEqual({ available: false, reason: "owned_config_unavailable" }); +}); + +it("uses the launcher's owned config and rejects a config outside that root", async () => { + const b = await seed(), root = await mkdtemp(path.join(os.tmpdir(), "review-admission-config-")); + const filename = path.join(root, "config.json"); + try { + await writeFile(filename, JSON.stringify({ $meta: { version: 1, updatedAt: new Date().toISOString(), source: "onboard" }, + database: { mode: "postgres", connectionString: database.connectionString }, logging: { mode: "file" }, server: {}, telemetry: {} }), { mode: 0o600 }); + const snapshot = await observeReviewAdmission(b, { PAPERCLIP_CONFIG: filename, PAPERCLIP_RUNNER_E2E_TEMP_ROOT: root }); + expect(snapshot).toMatchObject({ available: true, binding: b, wakes: [], runs: [] }); + expect(await observeReviewAdmission(b, { PAPERCLIP_CONFIG: filename, PAPERCLIP_RUNNER_E2E_TEMP_ROOT: path.join(root, "missing") })) + .toEqual({ available: false, reason: "observation_unavailable" }); + const sibling = await mkdtemp(path.join(os.tmpdir(), "review-admission-other-")); + try { expect(await observeReviewAdmission(b, { PAPERCLIP_CONFIG: filename, PAPERCLIP_RUNNER_E2E_TEMP_ROOT: sibling })) + .toEqual({ available: false, reason: "owned_config_unavailable" }); } + finally { await rm(sibling, { recursive: true, force: true }); } + } finally { await rm(root, { recursive: true, force: true }); } +}); diff --git a/tests/runner-e2e/chat-catalog-domains/external-app-review-diagnostic.ts b/tests/runner-e2e/chat-catalog-domains/external-app-review-diagnostic.ts new file mode 100644 index 00000000000..4edd5ed51a4 --- /dev/null +++ b/tests/runner-e2e/chat-catalog-domains/external-app-review-diagnostic.ts @@ -0,0 +1,87 @@ +import { readFile, realpath } from "node:fs/promises"; +import path from "node:path"; +import { createRequire } from "node:module"; +import { createDb, type Db } from "../../../packages/db/src/index.js"; +import { agents, issues, heartbeatRuns, toolActionRequests, toolInvocations, toolActionDeliveries, agentWakeupRequests } from "../../../packages/db/src/index.js"; +import { paperclipConfigSchema } from "../../../packages/shared/src/config-schema.js"; +const { and, eq, or, inArray, sql }: typeof import("../../../packages/db/node_modules/drizzle-orm/index.js") = + createRequire(new URL("../../../packages/db/package.json", import.meta.url))("drizzle-orm"); + +export interface ReviewAdmissionBinding { + companyId: string; issueId: string; agentId: string; actionRequestId: string; + invocationId: string; sourceRunId: string; interactionId: string; +} +/** Failure evidence only. No payloads, tool arguments, provider text or credentials. */ +export async function readReviewAdmissionSnapshot(db: Db, b: ReviewAdmissionBinding) { + if (Object.keys(b).length !== 7 || Object.values(b).some(value => !/^[a-f0-9]{8}(?:-[a-f0-9]{4}){3}-[a-f0-9]{12}$/i.test(value))) + return { available: false as const, reason: "invalid_binding" as const }; + return db.transaction(async tx => { + await tx.execute(sql`set local statement_timeout = '3000ms'`); + const [owner] = await tx.select({ actionStatus: toolActionRequests.status, invocationStatus: toolInvocations.status, + sourceStatus: heartbeatRuns.status, sourceNativePhase: heartbeatRuns.nativePhase, + issueStatus: issues.status, conversationState: issues.conversationState, agentStatus: agents.status, + }).from(toolActionRequests).innerJoin(toolInvocations, eq(toolInvocations.id, toolActionRequests.invocationId)) + .innerJoin(heartbeatRuns, eq(heartbeatRuns.id, toolInvocations.runId)).innerJoin(issues, eq(issues.id, toolActionRequests.issueId)) + .innerJoin(agents, eq(agents.id, toolInvocations.agentId)).where(and( + eq(toolActionRequests.id, b.actionRequestId), eq(toolActionRequests.companyId, b.companyId), + eq(toolActionRequests.issueId, b.issueId), eq(toolActionRequests.interactionId, b.interactionId), eq(toolActionRequests.requestedByAgentId, b.agentId), + eq(toolInvocations.id, b.invocationId), eq(toolInvocations.companyId, b.companyId), eq(toolInvocations.issueId, b.issueId), + eq(toolInvocations.agentId, b.agentId), eq(heartbeatRuns.id, b.sourceRunId), eq(heartbeatRuns.companyId, b.companyId), + eq(heartbeatRuns.agentId, b.agentId), + or(eq(heartbeatRuns.issueId, b.issueId), eq(heartbeatRuns.nativeIssueId, b.issueId), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${b.issueId}`), + eq(issues.companyId, b.companyId), eq(agents.companyId, b.companyId), + )).limit(1); + if (!owner) return { available: false as const, reason: "binding_mismatch" as const }; + const [delivery] = await tx.select({ createdAt: toolActionDeliveries.createdAt, deliveredAt: toolActionDeliveries.deliveredAt }) + .from(toolActionDeliveries).where(and(eq(toolActionDeliveries.companyId, b.companyId), eq(toolActionDeliveries.actionRequestId, b.actionRequestId), + eq(toolActionDeliveries.issueId, b.issueId), eq(toolActionDeliveries.interactionId, b.interactionId))).limit(1); + const wakeRows = await tx.select({ id: agentWakeupRequests.id, status: agentWakeupRequests.status, + runId: agentWakeupRequests.runId, coalescedCount: agentWakeupRequests.coalescedCount, + requestedAt: agentWakeupRequests.requestedAt, finishedAt: agentWakeupRequests.finishedAt, + }).from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, b.companyId), eq(agentWakeupRequests.agentId, b.agentId), + or(eq(agentWakeupRequests.idempotencyKey, `tool-action-response:${b.actionRequestId}`), + sql`${agentWakeupRequests.payload}->'toolActionRequestIds' @> ${JSON.stringify([b.actionRequestId])}::jsonb`))) + .orderBy(agentWakeupRequests.createdAt, agentWakeupRequests.id).limit(8); + const wakes = wakeRows.slice(0, 8), ids = wakes.flatMap(wake => wake.runId ? [wake.runId] : []); + const runRows = await tx.select({ id: heartbeatRuns.id, status: heartbeatRuns.status, runtimeMode: heartbeatRuns.runtimeMode, + nativePhase: heartbeatRuns.nativePhase, wakeupRequestId: heartbeatRuns.wakeupRequestId, + startedAt: heartbeatRuns.startedAt, finishedAt: heartbeatRuns.finishedAt, + }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, b.companyId), eq(heartbeatRuns.agentId, b.agentId), + or(eq(heartbeatRuns.issueId, b.issueId), eq(heartbeatRuns.nativeIssueId, b.issueId), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${b.issueId}`), + or(ids.length ? inArray(heartbeatRuns.id, ids) : sql`false`, + sql`${heartbeatRuns.contextSnapshot}->'toolActionRequestIds' @> ${JSON.stringify([b.actionRequestId])}::jsonb`))) + .orderBy(heartbeatRuns.createdAt, heartbeatRuns.id).limit(8); + return { available: true as const, observedAt: new Date().toISOString(), binding: b, owner, + delivery: delivery ?? null, wakes, runs: runRows.slice(0, 8), rowLimitReached: wakeRows.length === 8 || runRows.length === 8 }; + }, { isolationLevel: "repeatable read", accessMode: "read only" }); +} + +/** Same owned-config database pattern as controller-recovery-observer.ts. */ +export async function observeReviewAdmission(b: ReviewAdmissionBinding, env: NodeJS.ProcessEnv = process.env) { + let db: Db | undefined; + try { + if (!env.PAPERCLIP_CONFIG || !env.PAPERCLIP_RUNNER_E2E_TEMP_ROOT) return { available: false as const, reason: "owned_config_unavailable" as const }; + const [root, filename] = await Promise.all([realpath(env.PAPERCLIP_RUNNER_E2E_TEMP_ROOT), realpath(env.PAPERCLIP_CONFIG)]); + if (!filename.startsWith(`${root}${path.sep}`)) return { available: false as const, reason: "owned_config_unavailable" as const }; + const config = paperclipConfigSchema.parse(JSON.parse(await readFile(filename, "utf8"))); + const url = config.database.mode === "postgres" ? config.database.connectionString + : `postgres://paperclip:paperclip@127.0.0.1:${config.database.embeddedPostgresPort}/paperclip`; + if (!url) return { available: false as const, reason: "owned_config_unavailable" as const }; + db = createDb(url, { maxConnections: 1, connectTimeoutSeconds: 3 }); + return await readReviewAdmissionSnapshot(db, b); + } catch { return { available: false as const, reason: "observation_unavailable" as const }; } + finally { await db?.$client.end({ timeout: 1 }).catch(() => {}); } +} + +export function reviewAdmissionSummary(snapshot: Awaited>) { + if (!snapshot.available) return { available: false, reason: snapshot.reason }; + const states = ["queued", "running", "claimed", "completed", "coalesced", "deferred", "deferred_issue_execution", "succeeded", "failed", "cancelled", "timed_out", "skipped", "interrupted", "other"]; + const counts = (values: string[]) => Object.fromEntries(states.map(state => [state, + values.filter(value => (states.includes(value) ? value : "other") === state).length])); + return { available: true, delivery: !snapshot.delivery ? "missing" : snapshot.delivery.deliveredAt ? "acknowledged" : "pending", + wakes: snapshot.wakes.length, wakeStates: counts(snapshot.wakes.map(row => row.status)), + coalescedWakes: snapshot.wakes.filter(row => row.coalescedCount > 0).length, + boundRuns: snapshot.wakes.filter(wake => snapshot.runs.some(run => run.id === wake.runId)).length, + runStates: counts(snapshot.runs.map(row => row.status)), + nativeRunStates: counts(snapshot.runs.filter(row => row.runtimeMode === "native").map(row => row.status)), rowLimitReached: snapshot.rowLimitReached }; +} diff --git a/tests/runner-e2e/chat-catalog-domains/external-app-review.ts b/tests/runner-e2e/chat-catalog-domains/external-app-review.ts index bf977d1abed..49af0878e89 100644 --- a/tests/runner-e2e/chat-catalog-domains/external-app-review.ts +++ b/tests/runner-e2e/chat-catalog-domains/external-app-review.ts @@ -13,6 +13,7 @@ import { } from "./external-provider.js"; import { probeReviewApproval } from "./external-app-review-probe.js"; import { reviewOutcomeReply } from "./external-app-review-outcomes.js"; +import { observeReviewAdmission, reviewAdmissionSummary } from "./external-app-review-diagnostic.js"; import { openStandingChat } from "./identity-support.js"; import { definition, @@ -356,6 +357,7 @@ export const appReviewExternalCases = [ "chat-catalog-domains/external-app-review.ts", "chat-catalog-domains/external-app-review-outcomes.ts", "chat-catalog-domains/external-app-review-probe.ts", + "chat-catalog-domains/external-app-review-diagnostic.ts", "../../packages/shared/src/validators/issue.ts", "chat-catalog-domains/external-provider.ts", "chat-catalog-domains/identity-support.ts", @@ -544,8 +546,21 @@ export const appReviewExternalCases = [ ); await option.click(); } - const runs = await settled(ctx, index * 2 + 2), - reply = runs.find( + let runs: Awaited>; + try { runs = await settled(ctx, index * 2 + 2); } + catch (error) { + // Observe only the exact failed review phase. This cannot resume, + // settle or regrade it; the original wait failure remains authoritative. + try { + const snapshot = await observeReviewAdmission({ companyId: String(state.companyId), issueId: String(state.sourceId), + agentId: String(state.leadId), actionRequestId: phase.actionRequestId, invocationId: phase.invocationId, + interactionId: phase.interactionId, sourceRunId: phase.requestRunId }); + await ctx.input.evidence(`chat-catalog-app-review-admission-${index + 1}.json`, snapshot).catch(() => {}); + console.log("App review admission diagnostic", JSON.stringify(reviewAdmissionSummary(snapshot))); + } catch { /* Diagnostic failure cannot replace the original failure. */ } + throw error; + } + const reply = runs.find( (row) => row.contextSnapshot?.source === "tool_action_review" && Array.isArray(row.contextSnapshot.toolActionRequestIds) &&