diff --git a/src/api/agent-api-catalog.ts b/src/api/agent-api-catalog.ts index 76992bc18..a6b29783b 100644 --- a/src/api/agent-api-catalog.ts +++ b/src/api/agent-api-catalog.ts @@ -109,6 +109,25 @@ const FAMILIES: AgentApiFamily[] = [ }, ], }, + { + match: (m, p) => + (m === "GET" && p === "/v1/run-signals/active") || (m === "POST" && /^\/v1\/run-signals\/[^/]+$/.test(p)), + guidance: + "Signals act only on runs the asking person could see in their own UI; a run outside their reach answers 404. Steer requires text.", + routes: [ + { + method: "GET", + path: "/v1/run-signals/active?threadRef=…", + summary: "find the active run in a thread the asking person can see — answers {runId} or {runId:null}", + }, + { + method: "POST", + path: "/v1/run-signals/:runId", + summary: + 'signal a running turn with {kind:"abort"} to stop it or {kind:"steer", text} to redirect it mid-flight — the same stop/steer the web UI\'s buttons send', + }, + ], + }, { match: (m, p) => (m === "GET" && p === "/v1/conversations") || (m === "POST" && /^\/v1\/conversations\/[^/]+$/.test(p)), diff --git a/src/api/routes/turns.ts b/src/api/routes/turns.ts index ce7b4cd26..96f837686 100644 --- a/src/api/routes/turns.ts +++ b/src/api/routes/turns.ts @@ -85,6 +85,37 @@ async function postRunSignal(ctx: ApiCtx): Promise { return sendJson(res, 409, outcome); } +async function postAgentRunSignal(ctx: ApiCtx): Promise { + const { res, app, body, actor, capability } = ctx; + const viewer = actor?.p ?? capability?.actorId; + if (!viewer) + return sendJson(res, 401, { error: "capability_required", message: "this endpoint is for the agent self-API" }); + const id = ctx.params.id!; + const kind = isObj(body) && typeof body.kind === "string" ? body.kind : ""; + if (kind !== "abort" && kind !== "steer") { + return sendJson(res, 400, { error: "bad_request", message: "kind must be abort or steer" }); + } + const text = isObj(body) && typeof body.text === "string" ? body.text : undefined; + const outcome = await app.signalRun(id, { kind, ...(text !== undefined ? { text } : {}) }, viewer); + if (outcome.accepted) return sendJson(res, 200, outcome); + if (outcome.reason === "not_found") + return sendJson(res, 404, { error: "not_found", message: "not a run you can see" }); + if (outcome.reason === "text_required") + return sendJson(res, 400, { error: "bad_request", message: "text required", ...outcome }); + return sendJson(res, 409, outcome); +} + +async function getAgentActiveRunForThread(ctx: ApiCtx): Promise { + const { res, app, url, actor, capability } = ctx; + const viewer = actor?.p ?? capability?.actorId; + if (!viewer) + return sendJson(res, 401, { error: "capability_required", message: "this endpoint is for the agent self-API" }); + const threadRef = url.searchParams.get("threadRef") ?? ""; + if (!threadRef) return sendJson(res, 400, { error: "bad_request", message: "threadRef required" }); + const active = await app.activeRunForThread(threadRef, viewer); + return sendJson(res, 200, { runId: active?.runId ?? null }); +} + async function getRun(ctx: ApiCtx): Promise { const { res, app, actor } = ctx; const id = ctx.params.id!; @@ -157,6 +188,8 @@ export const turnRoutes: ReadonlyArray> = [ { method: "GET", path: "/v1/approvals/:id", auth: "source", handle: getApproval }, { method: "POST", path: "/v1/runs/:id/delivery-state", auth: "source", handle: postRunDeliveryState }, { method: "POST", path: "/v1/runs/:id/signal", auth: "source", handle: postRunSignal }, + { method: "POST", path: "/v1/run-signals/:id", auth: "either", handle: postAgentRunSignal }, + { method: "GET", path: "/v1/run-signals/active", auth: "either", handle: getAgentActiveRunForThread }, { method: "GET", path: "/v1/runs/:id", auth: "source", handle: getRun }, { method: "GET", path: "/v1/runs", auth: "source", handle: getActiveRunForThread }, { method: "GET", path: "/v1/deliveries", auth: "source", handle: listDeliveries }, diff --git a/test/agent-run-signal.test.ts b/test/agent-run-signal.test.ts new file mode 100644 index 000000000..4ecc7b37f --- /dev/null +++ b/test/agent-run-signal.test.ts @@ -0,0 +1,98 @@ +import "./support/auto-fake-sprites.ts"; +import { test, after } from "node:test"; +import assert from "node:assert/strict"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { AddressInfo } from "node:net"; +import { createServer } from "../src/api/server.ts"; +import { buildApp } from "../src/wiring.ts"; +import { mintCapabilityToken, CAPABILITY_TTL_MS, CONTROL_PLANE_AUD } from "../src/auth/capability-token.ts"; +import type { OrchestratorInput } from "../src/core/orchestrator.ts"; +import type { Principal } from "../src/types.ts"; +import { scopeId } from "../src/types.ts"; +import { testConfig } from "./support/test-config.ts"; + +const SECRET = "agent-run-signal-secret".repeat(2); + +const built = buildApp( + testConfig({ dataDir: mkdtempSync(join(tmpdir(), "agent-run-signal-")), signingSecret: SECRET }), +); +const core = createServer(built.app, { signingSecret: SECRET }); +core.listen(0); +const base = `http://localhost:${(core.address() as AddressInfo).port}`; + +after(async () => { + await new Promise((r) => core.close(() => r())); + await built.runtime.stop(); +}); + +const owner: Principal = { id: "internal:U1", type: "internal" }; +function request(text: string, threadRef: string): OrchestratorInput { + return { actor: owner, conversation: { kind: "dm", threadRef, audience: [owner] }, origin: { kind: "direct" }, text }; +} + +const capFor = (actorId: string) => + mintCapabilityToken( + { actorId, scopeId: scopeId("personal", actorId), aud: CONTROL_PLANE_AUD, exp: Date.now() + CAPABILITY_TTL_MS }, + SECRET, + ); + +async function signal( + runId: string, + body: unknown, + token?: string, +): Promise<{ status: number; json: Record }> { + const r = await fetch(`${base}/v1/run-signals/${encodeURIComponent(runId)}`, { + method: "POST", + headers: { "content-type": "application/json", ...(token ? { "x-agent-capability": token } : {}) }, + body: JSON.stringify(body), + }); + return { status: r.status, json: (await r.json()) as Record }; +} + +test("agent route: the owner's capability token can abort and steer their own pending run", async () => { + const { run } = await built.runs.enqueue({ sessionId: "a-accept", request: request("hi", "t-agent-accept") }); + const token = await capFor(owner.id); + for (const body of [{ kind: "steer", text: "go left" }, { kind: "abort" }]) { + const r = await signal(run.id, body, token); + assert.equal(r.status, 200); + assert.equal(r.json.accepted, true); + } +}); + +test("agent route: another principal's token is told the run does not exist", async () => { + const { run } = await built.runs.enqueue({ sessionId: "a-scope", request: request("hi", "t-agent-scope") }); + const r = await signal(run.id, { kind: "abort" }, await capFor("internal:U2")); + assert.equal(r.status, 404); +}); + +test("agent route: no token at all is refused", async () => { + const { run } = await built.runs.enqueue({ sessionId: "a-noauth", request: request("hi", "t-agent-noauth") }); + const r = await signal(run.id, { kind: "abort" }); + assert.equal(r.status, 401); +}); + +test("agent route: bad kind 400, steer without text 400, unknown run 404", async () => { + const { run } = await built.runs.enqueue({ sessionId: "a-bad", request: request("hi", "t-agent-bad") }); + const token = await capFor(owner.id); + assert.equal((await signal(run.id, { kind: "explode" }, token)).status, 400); + assert.equal((await signal(run.id, { kind: "steer" }, token)).status, 400); + assert.equal((await signal("no-such-run", { kind: "abort" }, token)).status, 404); +}); + +test("agent route: active-run discovery is viewer-bound", async () => { + const threadRef = "t-agent-discover"; + const { run } = await built.runs.enqueue({ sessionId: threadRef, request: request("hi", threadRef) }); + const mine = await fetch(`${base}/v1/run-signals/active?threadRef=${encodeURIComponent(threadRef)}`, { + headers: { "x-agent-capability": await capFor(owner.id) }, + }); + assert.equal(mine.status, 200); + assert.equal(((await mine.json()) as { runId?: string | null }).runId, run.id); + + const theirs = await fetch(`${base}/v1/run-signals/active?threadRef=${encodeURIComponent(threadRef)}`, { + headers: { "x-agent-capability": await capFor("internal:U2") }, + }); + assert.equal(theirs.status, 200); + assert.equal(((await theirs.json()) as { runId?: string | null }).runId, null); +});