Agent-callable run signals: abort/steer + active-run discovery

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)),

src/api/routes/turns.ts

@@ -85,6 +85,37 @@ async function postRunSignal(ctx: ApiCtx): Promise<void> {
   return sendJson(res, 409, outcome);
 }
 
+async function postAgentRunSignal(ctx: ApiCtx): Promise<void> {
+  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<void> {
+  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<void> {
   const { res, app, actor } = ctx;
   const id = ctx.params.id!;
@@ -157,6 +188,8 @@ export const turnRoutes: ReadonlyArray<Route<ApiCtx>> = [
   { 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 },

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<void>((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<string, unknown> }> {
+  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<string, unknown> };
+}
+
+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);
+});