diff --git a/tests/runner-e2e/chat-catalog-domains/internal-run-brief.ts b/tests/runner-e2e/chat-catalog-domains/internal-run-brief.ts new file mode 100644 index 00000000000..d52bf591417 --- /dev/null +++ b/tests/runner-e2e/chat-catalog-domains/internal-run-brief.ts @@ -0,0 +1,40 @@ +import { createServer, type ServerResponse } from "node:http"; +import { randomBytes } from "node:crypto"; +import { listenOnFetchAllowedPort } from "../../e2e/fetch-allowed-port.js"; +import type { ChatCatalogContext } from "../chat-catalog-contract.js"; +/** A real read-only app holds its first tool response while the product retains the actual worker checkout. */ +export async function startRunBriefGate(ctx: ChatCatalogContext, workerId: string) { + const token = randomBytes(32).toString("hex"); ctx.input.privateSecrets?.push(token); + const calls: Array<{ sequence: number; authorized: boolean; operation: string; receivedAt: string }> = []; + let released = false; const pending: Array<{ response: ServerResponse; id: unknown }> = []; + const reply = (response: ServerResponse, id: unknown, result: unknown) => { if (!response.destroyed) { response.writeHead(200, { "content-type": "application/json" }); response.end(JSON.stringify({ jsonrpc: "2.0", id, result })); } }; + const brief = { content: [{ type: "text", text: "The website brief is now available: Acorn Notes needs a semantic heading and an accessible signup link. For this prerequisite, record a short implementation note and complete the task. No website file or delivery is requested yet." }] }; + const server = createServer(async (request, response) => { + try { + let size = 0; const chunks: Buffer[] = []; for await (const chunk of request) { size += chunk.length; if (size > 65536) throw new Error("Oversized app message"); chunks.push(chunk); } + const message = JSON.parse(Buffer.concat(chunks).toString("utf8") || "{}"); + const authorized = request.headers.authorization === `Bearer ${token}`; + if (!authorized) { response.writeHead(401); response.end(); return; } + if (message.method === "initialize") { reply(response, message.id, { protocolVersion: "2024-11-05", capabilities: { tools: {} }, serverInfo: { name: "Website Brief", version: "1.0.0" } }); return; } + if (message.method === "tools/list") { reply(response, message.id, { tools: [{ name: "read_website_brief", title: "Read website brief", description: "Read the authorized website task brief. The first read waits for its actual source to become available.", annotations: { readOnlyHint: true, destructiveHint: false }, inputSchema: { type: "object", properties: {}, additionalProperties: false } }] }); return; } + if (message.method === "tools/call" && message.params?.name === "read_website_brief") { + calls.push({ sequence: calls.length + 1, authorized, operation: "read_website_brief", receivedAt: new Date().toISOString() }); + if (released) reply(response, message.id, brief); else pending.push({ response, id: message.id }); return; + } + response.writeHead(204); response.end(); + } catch { response.writeHead(400); response.end(); } + }); + const port = await listenOnFetchAllowedPort(server); let connectionId: string | undefined; + try { + const root = `/api/companies/${ctx.input.fixtures.company.id}`; + const connected = await ctx.input.api.postSensitive<{ connectionId: string; actions: { readOnly: Array<{ catalogEntryId: string }> } }>(`${root}/tools/apps/connect`, { + link: `http://127.0.0.1:${port}/`, name: "Website Brief", authMode: "bearer", credentialValues: { "credentials.authorization": token }, grantKind: "organization", + }); + connectionId = connected.connectionId; const reads = connected.actions.readOnly.map(row => row.catalogEntryId); + if (reads.length !== 1) throw new Error("Website Brief discovery did not yield exactly its read-only action"); + await ctx.input.api.post(`${root}/tools/apps/${connectionId}/finish`, { enabledCatalogEntryIds: reads, askFirstCatalogEntryIds: [], access: { agentIds: [workerId] } }); + await ctx.input.api.putSensitive(`/api/tool-connections/${connectionId}/installs`, { installs: [{ targetType: "agent", targetId: workerId }] }); + return { connectionId, calls, release() { released = true; for (const item of pending.splice(0)) reply(item.response, item.id, brief); }, + async close() { for (const item of pending.splice(0)) item.response.destroy(); server.closeAllConnections(); await new Promise(resolve => server.close(() => resolve())); await ctx.input.api.delete(`/api/tool-connections/${connectionId}`, { allowNotFound: true }); } }; + } catch (error) { server.closeAllConnections(); await new Promise(resolve => server.close(() => resolve())); if (connectionId) await ctx.input.api.delete(`/api/tool-connections/${connectionId}`, { allowNotFound: true }); throw error; } +} diff --git a/tests/runner-e2e/chat-catalog-domains/internal-run-history.test.ts b/tests/runner-e2e/chat-catalog-domains/internal-run-history.test.ts new file mode 100644 index 00000000000..46d38b37208 --- /dev/null +++ b/tests/runner-e2e/chat-catalog-domains/internal-run-history.test.ts @@ -0,0 +1,21 @@ +import { describe, expect, it } from "vitest"; +import { gradeRunHistory } from "./internal-run-history.js"; +function valid() { + const task = { id: "task", title: "Website task", status: "done", assigneeAgentId: "morgan" }; + const interruptedRun = { id: "interrupted", companyId: "company", agentId: "morgan", status: "cancelled", contextSnapshot: { issueId: "task" } }; + const continuationRun = { id: "continuation", companyId: "company", agentId: "morgan", status: "succeeded", contextSnapshot: { issueId: "task" } }; + return { scenarioId: "activity.find-runs", origin: "https://fixture.test", companyId: "company", leadId: "ceo", state: { task, worker: { id: "morgan" }, interruptedRun, continuationRun, interruptedTask: { ...task, status: "in_progress", executionRunId: "interrupted" }, gateCalls: [{ authorized: true, operation: "read_website_brief" }] }, + tasks: [task], projects: [], goals: [], comments: [], activity: [] as Array<{ entityId: string; agentId: string; action: string }>, documents: [], revisions: [], humanQueries: [], + reply: "Morgan's interrupted attempt [cancelled](/RUN/agents/morgan/runs/interrupted).\nMorgan's continuation [succeeded](/RUN/agents/morgan/runs/continuation).", links: ["/RUN/issues/task", "/RUN/agents/morgan/runs/interrupted", "/RUN/agents/morgan/runs/continuation"], + runs: [interruptedRun, continuationRun], agentRunReadStatuses: [200, 200], currentTask: task }; +} +describe("actual interrupted and continued execution calibration", () => { + it("accepts independently retained cancellation and successful continuation", () => expect(gradeRunHistory(valid()).passed).toBe(true)); + it("rejects exposing a run whose acting-agent detail read is denied", () => { const o = valid(); o.agentRunReadStatuses[0] = 403; expect(gradeRunHistory(o).passed).toBe(false); }); + it("rejects hiding the interrupted attempt", () => { const o = valid(); o.runs = o.runs.slice(1); expect(gradeRunHistory(o).passed).toBe(false); }); + it("rejects the wrong execution owner or task binding", () => { const o = valid(); o.runs[0]!.agentId = "riley"; expect(gradeRunHistory(o).passed).toBe(false); const wrong = valid(); wrong.runs[1]!.contextSnapshot.issueId = "another"; expect(gradeRunHistory(wrong).passed).toBe(false); }); + it("rejects swapped reply states despite both words elsewhere", () => { const o = valid(); o.reply = "Morgan interrupted succeeded.\nMorgan continuation cancelled."; expect(gradeRunHistory(o).passed).toBe(false); }); + it("rejects foreign or missing execution links", () => { const o = valid(); o.links[1] = "https://foreign.test/RUN/agents/morgan/runs/interrupted"; expect(gradeRunHistory(o).passed).toBe(false); }); + it("requires actual held checkout and authorized app call", () => { const o = valid(); o.state.interruptedTask.executionRunId = "other"; expect(gradeRunHistory(o).passed).toBe(false); const call = valid(); call.state.gateCalls = []; expect(gradeRunHistory(call).passed).toBe(false); }); + it("rejects changing task history or ownership while listing", () => { const o = valid(); o.activity.push({ entityId: "task", agentId: "ceo", action: "issue.updated" }); expect(gradeRunHistory(o).passed).toBe(false); }); +}); diff --git a/tests/runner-e2e/chat-catalog-domains/internal-run-history.ts b/tests/runner-e2e/chat-catalog-domains/internal-run-history.ts new file mode 100644 index 00000000000..ac358d1f43f --- /dev/null +++ b/tests/runner-e2e/chat-catalog-domains/internal-run-history.ts @@ -0,0 +1,92 @@ +import { randomUUID } from "node:crypto"; +import { expect } from "@playwright/test"; +import type { ChatCatalogCase } from "../chat-catalog-contract.js"; +import { pollUntil } from "../api.js"; +import { fixtureWorker, makeDefinition } from "./internal-definition.js"; +import { startRunBriefGate } from "./internal-run-brief.js"; +import { checks, entity, hasLink, observation, observeInternal, project, task, type Entity, type InternalObservation } from "./internal-support.js"; +interface OwnedTask extends Entity { checkoutRunId?: string | null; executionRunId?: string | null } +interface RunHistory extends InternalObservation { currentTask: OwnedTask; agentRunReadStatuses: number[] } +function runHref(o: InternalObservation, agentId: string, runId: string) { + return o.links.find(href => { try { const u = new URL(href, o.origin); return u.origin === o.origin && u.pathname.endsWith(`/agents/${agentId}/runs/${runId}`); } catch { return false; } }); +} +export function gradeRunHistory(value: unknown) { + const c = checks(); const o = observation(value) as RunHistory | undefined; + c.add("actual-interrupted-task-history", !!o && !!o.currentTask && Array.isArray(o.state.gateCalls), "Native task-bound runs, actual interrupted checkout evidence and continuation readback are required."); + if (!o || !o.currentTask || !Array.isArray(o.state.gateCalls)) return c.result(); + const worker = o.state.worker as Entity; const t = o.state.task as Entity; const first = o.state.interruptedRun as Entity; const later = o.state.continuationRun as Entity; + const history = o.runs.filter(run => run.agentId === worker?.id && (run.contextSnapshot?.issueId === t?.id || run.contextSnapshot?.taskId === t?.id)); + c.add("both-real-attempts-retained", history.length === 2 && !!first && !!later && first.id !== later.id + && history.some(run => run.id === first.id && run.status === "cancelled") && history.some(run => run.id === later.id && run.status === "succeeded") + && (o.state.gateCalls as Array<{ authorized?: boolean; operation?: string }>).some(call => call.authorized && call.operation === "read_website_brief") + && (o.state.interruptedTask as OwnedTask | undefined)?.executionRunId === first.id, + "A real operator cancellation while Morgan owns the checkout remains alongside the actual successful retry; no synthetic run rows are used."); + c.add("independent-agent-execution-read-authority", Array.isArray(o.agentRunReadStatuses) && o.agentRunReadStatuses.length === 2 && o.agentRunReadStatuses.every(status => status === 200), "The actual acting agent can read both exact run details under its own bearer; task access alone is not assumed to grant run access."); + c.add("exact-task-and-agent-run-links", !!t && hasLink(o, "issues", t) && !!worker && !!first && !!later && !!runHref(o, worker.id, first.id) && !!runHref(o, worker.id, later.id), + "Each relevant execution links to the actual Morgan run detail and the originating website task."); + const runPart = (id: string) => o.reply.split(/\n/).filter(part => part.includes(id)).join(" "); + c.add("interrupted-and-continuation-states-explained", !!first && !!later && /\b(?:cancelled|canceled|interrupted|stopped)\b/i.test(runPart(first.id)) + && /\b(?:succeeded|successful|completed|finished)\b/i.test(runPart(later.id)) && /\bMorgan\b/.test(o.reply), + "Run-attributed reply rows distinguish the cancelled attempt from the successful continuation and identify the actual executing agent."); + c.add("no-history-rewrite-or-extra-owner", o.currentTask.id === t?.id && o.currentTask.assigneeAgentId === worker?.id && o.currentTask.status === "done" + && history.every(run => [first?.id, later?.id].includes(run.id)) + && !o.activity.some(row => row.entityId === t?.id && row.agentId === o.leadId && /issue\.(?:updated|deleted)/.test(row.action ?? "")), + "A history request leaves the completed task, execution ownership and both durable attempts unchanged."); + return c.result(); +} +function findRuns(): ChatCatalogCase { + let gate: Awaited> | undefined; let key: { id: string; token: string } | undefined; + const base = makeDefinition("activity.find-runs", async ctx => { + const worker = await fixtureWorker(ctx); const p = await project(ctx); const root = `/api/companies/${ctx.input.fixtures.company.id}`; + for (const scope of [{ scopeType: "company", scopeId: ctx.input.fixtures.company.id }, { scopeType: "agent", scopeId: worker.id }]) + await ctx.input.api.post(`${root}/budgets/policies`, { ...scope, amount: 500, hardStopEnabled: true, isActive: true, unpricedUsagePolicy: "allow" }); + gate = await startRunBriefGate(ctx, worker.id); + try { + const t = await task(ctx, "Website task", p.id, { status: "todo", assigneeAgentId: worker.id, + description: "Read the authorized Website Brief app's read_website_brief tool before doing any work. The source read may be pending; do not invent its contents or start work without its real response. After it returns, follow that short prerequisite brief and record your note, then finish the task. Do not create other tasks." }); + const first = await pollUntil({ label: "Actual Morgan checkout waiting on authorized brief", deadlineAt: Date.now() + 150_000, + load: async () => ({ task: await ctx.input.api.get(`/api/issues/${t.id}`), runs: await ctx.allRuns() }), + accept: value => gate!.calls.length > 0 && value.runs.length === 1 && value.runs[0]?.agentId === worker.id && value.runs[0]?.status === "running" && value.task.executionRunId === value.runs[0]?.id, + reject: value => value.runs.some(run => ["failed", "cancelled", "succeeded", "timed_out"].includes(run.status)) ? "First prerequisite attempt ended before the real interruption boundary" : undefined }); + const interruptedRun = first.runs[0]!; const gateCalls = structuredClone(gate.calls); + await ctx.input.api.post(`/api/heartbeat-runs/${interruptedRun.id}/cancel`, { cancellationRequestId: randomUUID() }); + const cancelledRun = await pollUntil({ label: "Durable real operator cancellation", deadlineAt: Date.now() + 60_000, + load: () => ctx.input.api.get(`/api/heartbeat-runs/${interruptedRun.id}`), accept: run => run.status === "cancelled" }); + gate.release(); + const receipt = await ctx.input.api.post<{ accepted: boolean; sourceRunId?: string }>(`/api/issues/${t.id}/retry`, { idempotencyKey: randomUUID(), observedRunId: interruptedRun.id }); + if (!receipt.accepted) throw new Error("Public retry did not accept the actual cancelled prerequisite attempt"); + const continuation = await pollUntil({ label: "Actual task continuation completes after the brief is available", deadlineAt: Date.now() + 180_000, + load: async () => ({ task: await ctx.input.api.get(`/api/issues/${t.id}`), runs: await ctx.allRuns() }), + accept: value => value.runs.length === 2 && value.runs.some(run => run.id === interruptedRun.id && run.status === "cancelled") + && value.runs.some(run => run.id !== interruptedRun.id && run.agentId === worker.id && run.status === "succeeded") && value.task.status === "done", + reject: value => value.runs.some(run => run.id !== interruptedRun.id && ["failed", "cancelled", "timed_out"].includes(run.status)) ? "The actual continuation did not finish" : undefined }); + const continuationRun = continuation.runs.find(run => run.id !== interruptedRun.id)!; + if (!ctx.input.privateSecrets) throw new Error("Execution read controls require private credential redaction"); + key = await ctx.input.api.postSensitive(`/api/agents/${ctx.input.fixtures.agent.id}/keys`, { name: `run-history-read-${ctx.input.nonce}` }); ctx.input.privateSecrets.push(key!.token); + const state = { project: p, task: continuation.task, worker, interruptedTask: first.task, interruptedRun: cancelledRun, continuationRun, gateCalls, retryReceipt: receipt }; + await ctx.input.evidence("chat-catalog-real-interrupted-continuation.json", state); return state; + } catch (error) { await gate.close(); gate = undefined; throw error; } + }, gradeRunHistory); + return { ...base, expectedRunCount: 3, minimumRunCount: 3, attemptTimeoutMs: 660_000, + runPolicy: { kind: "terminal-counts", terminalStatusCounts: { cancelled: 1, succeeded: 2 } }, + sourceFiles: [...base.sourceFiles, "chat-catalog-domains/internal-run-history.ts", "chat-catalog-domains/internal-run-brief.ts", "../e2e/fetch-allowed-port.ts"], + async observe(ctx, state): Promise { + if (!key) throw new Error("Run read probe was not created during setup"); const o = await observeInternal(ctx, state); + const agentRunReadStatuses = await Promise.all(["interruptedRun", "continuationRun"].map(async name => { + const url = new URL(`/api/heartbeat-runs/${entity(state, name).id}`, ctx.input.api.baseURL); + const response = await fetch(url, { redirect: "error", signal: AbortSignal.timeout(15_000), headers: { Authorization: `Bearer ${key!.token}` } }); await response.body?.cancel(); return response.status; + })); + return { ...o, agentRunReadStatuses, currentTask: await ctx.input.api.get(`/api/issues/${entity(state, "task").id}`) }; + }, + async verifyResult(ctx, state, value) { + const o = value as RunHistory; await base.verifyResult(ctx, state, value); const source = ctx.input.page.url(); const worker = entity(state, "worker"); + try { + for (const key of ["interruptedRun", "continuationRun"]) { const run = entity(state, key); const href = runHref(o, worker.id, run.id); if (!href) throw new Error("Missing exact native execution link"); + await ctx.input.page.goto(new URL(href, o.origin).href, { waitUntil: "domcontentloaded" }); await expect(ctx.input.page.getByText(new RegExp(run.status === "cancelled" ? "cancelled|canceled" : "succeeded|successful", "i")).first()).toBeVisible(); + await ctx.input.evidence(`chat-catalog-${key}-opened.json`, { runId: run.id, agentId: worker.id, taskId: entity(state, "task").id, url: ctx.input.page.url() }); } + } finally { await ctx.input.page.goto(source, { waitUntil: "domcontentloaded" }); await expect(ctx.input.page.getByTestId("task-chat-composer-input")).toBeVisible(); } + }, + async cleanup(ctx) { try { if (key) { await ctx.input.api.delete(`/api/agents/${ctx.input.fixtures.agent.id}/keys/${key.id}`, { allowNotFound: true }); key = undefined; } } finally { await gate?.close(); gate = undefined; } }, + }; +} +export const runHistoryCases: readonly ChatCatalogCase[] = [findRuns()]; diff --git a/tests/runner-e2e/chat-catalog-domains/internal.test.ts b/tests/runner-e2e/chat-catalog-domains/internal.test.ts index 94af6807652..4225151dbfa 100644 --- a/tests/runner-e2e/chat-catalog-domains/internal.test.ts +++ b/tests/runner-e2e/chat-catalog-domains/internal.test.ts @@ -15,6 +15,14 @@ function event(action: string, entityId: string, details?: Record