AGENT_ENGINE selects which agent loop runs a turn (default: langgraph)

apps/agent-orchestrator/src/config.ts

@@ -31,6 +31,25 @@ export interface AppConfig {
    * prune, so a caller sending a handful pays no embedding or Qdrant cost at all.
    */
   callerToolTopK: number;
+  /**
+   * Which agent loop runs a turn (docs/adr/0036).
+   *
+   * `"langgraph"` is the in-process graph this app has always used and is the
+   * default, so enabling the Temporal engine is always an explicit act.
+   * `"temporal"` forwards the turn to `engines/temporal` over HTTP; everything
+   * else about this process — the facade, identity resolution, RBAC, the
+   * credential store, both launchers — is unchanged either way.
+   *
+   * Process-wide rather than per-request on purpose: the two engines keep
+   * conversation state in different places (a Redis session record vs. workflow
+   * state), so alternating between them mid-conversation would lose whichever
+   * one it left.
+   */
+  agentEngine: "langgraph" | "temporal";
+  /** Base URL of the Temporal engine's gateway Service. Required when agentEngine is "temporal". */
+  temporalEngineUrl: string | undefined;
+  /** Bearer token presented to that gateway. */
+  temporalEngineToken: string | undefined;
   /**
    * How long an unused caller-tool definition survives before being swept
    * (Qdrant has no native TTL, so this is a periodic `prune`). Any turn that
@@ -218,6 +237,9 @@ export const config: AppConfig = {
   agentsQdrantCollection: process.env.AGENT_QDRANT_AGENTS_COLLECTION ?? "agents",
   callerToolsQdrantCollection: process.env.AGENT_QDRANT_CALLER_TOOLS_COLLECTION ?? "caller_tools",
   callerToolTopK: num(process.env.AGENT_CALLER_TOOL_TOP_K, 5),
+  agentEngine: process.env.AGENT_ENGINE === "temporal" ? "temporal" : "langgraph",
+  temporalEngineUrl: process.env.AGENT_TEMPORAL_ENGINE_URL,
+  temporalEngineToken: process.env.AGENT_TEMPORAL_ENGINE_TOKEN,
   callerToolTtlSeconds: num(process.env.AGENT_CALLER_TOOL_TTL_SECONDS, 604_800), // 7 days
   callerToolPruneIntervalSeconds: num(process.env.AGENT_CALLER_TOOL_PRUNE_INTERVAL_SECONDS, 3_600),
   embeddingModel: process.env.AGENT_EMBEDDING_MODEL ?? "text-embedding-3-small",

apps/agent-orchestrator/src/engine/temporal-engine.test.ts

@@ -0,0 +1,147 @@
+import { describe, expect, it, vi } from "vitest";
+import { TemporalEngine } from "./temporal-engine.js";
+import { SENDER_ASSERTION_HEADER } from "../rbac/sender-assertion.js";
+import { verifySenderAssertion } from "../rbac/sender-assertion.js";
+import type { AgentGraphInput } from "../server.js";
+
+const BASE = "http://temporal-engine-gateway:8080";
+
+function input(overrides: Partial<AgentGraphInput> = {}): AgentGraphInput {
+  return { request: "what pods are running?", authToken: "caller-token", ...overrides };
+}
+
+/** Scripts the accept/poll pair: one POST /invoke, then the given GET responses in order. */
+function scriptedFetch(records: unknown[], accepted = { id: "conversation-abc.upd-1", status: "pending" }) {
+  const calls: { url: string; init?: RequestInit }[] = [];
+  let polls = 0;
+  const impl = vi.fn(async (url: string | URL | Request, init?: RequestInit) => {
+    const href = String(url);
+    calls.push({ url: href, ...(init ? { init } : {}) });
+    if (href.endsWith("/invoke")) {
+      return new Response(JSON.stringify(accepted), { status: 202 });
+    }
+    const body = records[Math.min(polls, records.length - 1)];
+    polls++;
+    return new Response(JSON.stringify(body), { status: 200 });
+  });
+  return { impl: impl as unknown as typeof fetch, calls };
+}
+
+describe("TemporalEngine", () => {
+  it("starts a turn and returns the result once the record settles", async () => {
+    const { impl, calls } = scriptedFetch([
+      { id: "x", status: "pending" },
+      { id: "x", status: "succeeded", result: "three pods, all Running" },
+    ]);
+    const engine = new TemporalEngine({ baseUrl: BASE, fetchImpl: impl });
+
+    const state = await engine.invoke(input());
+
+    expect(state.result).toBe("three pods, all Running");
+    expect(state.error).toBeUndefined();
+    expect(calls[0]!.url).toBe(`${BASE}/invoke`);
+    expect(calls[1]!.url).toBe(`${BASE}/invoke/conversation-abc.upd-1`);
+  });
+
+  it("reports a failed turn as an error rather than an empty result", async () => {
+    const { impl } = scriptedFetch([{ id: "x", status: "failed", error: "the planner exploded" }]);
+    const engine = new TemporalEngine({ baseUrl: BASE, fetchImpl: impl });
+
+    const state = await engine.invoke(input());
+    expect(state.error).toBe("the planner exploded");
+    expect(state.result).toBeUndefined();
+  });
+
+  // The second non-error terminal shape (docs/adr/0035) has to survive this hop,
+  // or a caller offering its own tools would get an empty success.
+  it("carries pending caller tool calls through", async () => {
+    const { impl } = scriptedFetch([
+      {
+        id: "x",
+        status: "succeeded",
+        toolCalls: [{ id: "call_1", name: "web_search", arguments: '{"query":"x"}' }],
+      },
+    ]);
+    const engine = new TemporalEngine({ baseUrl: BASE, fetchImpl: impl });
+
+    const state = await engine.invoke(input());
+    expect(state.pendingToolCalls).toEqual([{ id: "call_1", name: "web_search", arguments: '{"query":"x"}' }]);
+    expect(state.result).toBeUndefined();
+  });
+
+  // This process owns the route registry and has already matched it. Re-deriving
+  // the target in the engine would put routing policy in two places.
+  it("names an already-matched route target instead of resending the event", async () => {
+    const { impl, calls } = scriptedFetch([{ id: "x", status: "succeeded", result: "triaged" }]);
+    const engine = new TemporalEngine({ baseUrl: BASE, fetchImpl: impl });
+
+    await engine.invoke(input({ forcedAgentId: "claude-code-swe-agent", sessionId: "github:acme/widgets#7" }));
+
+    const body = JSON.parse(String(calls[0]!.init!.body));
+    expect(body.forcedAgentId).toBe("claude-code-swe-agent");
+    expect(body.sessionId).toBe("github:acme/widgets#7");
+    expect(body.event).toBeUndefined();
+  });
+
+  // The sender login selects which stored credentials a run receives, so an
+  // internal hop is exactly as unsuited to trusting it unsigned as an external
+  // one. Signed with the same contract integration-gateway uses.
+  it("signs the sender login rather than sending it as a body field", async () => {
+    const secret = "shared-with-integration-gateway";
+    const { impl, calls } = scriptedFetch([{ id: "x", status: "succeeded", result: "ok" }]);
+    const engine = new TemporalEngine({ baseUrl: BASE, fetchImpl: impl, senderAssertionSecret: secret });
+
+    await engine.invoke(input({ senderLogin: "imaustink" }));
+
+    const headers = calls[0]!.init!.headers as Record<string, string>;
+    const assertion = headers[SENDER_ASSERTION_HEADER];
+    expect(assertion).toBeDefined();
+    expect(verifySenderAssertion(secret, assertion)).toBe("imaustink");
+
+    const body = JSON.parse(String(calls[0]!.init!.body));
+    expect(body.senderLogin).toBeUndefined();
+  });
+
+  it("omits the assertion when no secret is configured", async () => {
+    const { impl, calls } = scriptedFetch([{ id: "x", status: "succeeded", result: "ok" }]);
+    const engine = new TemporalEngine({ baseUrl: BASE, fetchImpl: impl });
+
+    await engine.invoke(input({ senderLogin: "imaustink" }));
+    const headers = calls[0]!.init!.headers as Record<string, string>;
+    expect(headers[SENDER_ASSERTION_HEADER]).toBeUndefined();
+  });
+
+  // A turn that outlives the poll budget is NOT lost: the record is the
+  // workflow, not this process's memory, so the answer stays collectable. It is
+  // reported as a resumable pause rather than a failure — the same shape ADR
+  // 0033 settled on for an interrupted turn.
+  it("reports a resumable pause rather than an error when it stops waiting", async () => {
+    const { impl } = scriptedFetch([{ id: "x", status: "pending" }]);
+    const engine = new TemporalEngine({ baseUrl: BASE, fetchImpl: impl, timeoutMs: 0 });
+
+    const state = await engine.invoke(input());
+    expect(state.error).toBeUndefined();
+    expect(state.result).toContain("still running");
+  });
+
+  it("throws when the engine is unreachable, so the turn fails honestly", async () => {
+    const impl = vi.fn(async () => new Response("nope", { status: 502 })) as unknown as typeof fetch;
+    const engine = new TemporalEngine({ baseUrl: BASE, fetchImpl: impl });
+
+    await expect(engine.invoke(input())).rejects.toThrow(/502/);
+  });
+
+  // Streaming yields the answer but no per-node narration: those lines describe
+  // LangGraph node transitions, which do not exist on this engine.
+  it("streams a single terminal update", async () => {
+    const { impl } = scriptedFetch([{ id: "x", status: "succeeded", result: "done" }]);
+    const engine = new TemporalEngine({ baseUrl: BASE, fetchImpl: impl });
+
+    const updates: Record<string, unknown>[] = [];
+    for await (const update of await engine.stream(input(), { streamMode: "updates" })) {
+      updates.push(update);
+    }
+    expect(updates).toHaveLength(1);
+    expect(Object.values(updates[0]!)[0]).toMatchObject({ result: "done" });
+  });
+});

apps/agent-orchestrator/src/engine/temporal-engine.ts

@@ -0,0 +1,194 @@
+import { SENDER_ASSERTION_HEADER, mintSenderAssertion } from "../rbac/sender-assertion.js";
+import type { AgentGraphInput, AgentGraphLike } from "../server.js";
+import type { AgentState } from "../agent/graph.js";
+
+/**
+ * Runs a turn on the Temporal engine (`engines/temporal`) instead of the
+ * in-process LangGraph graph — the `AGENT_ENGINE=temporal` half of the switch.
+ *
+ * ## Why this is an HTTP client and not a Temporal client
+ *
+ * The obvious implementation embeds `@temporalio/client` and does
+ * update-with-start from this process. It was rejected for three reasons, in
+ * ascending order of importance:
+ *
+ * 1. It adds a substantial npm dependency to an app that needs one HTTP call.
+ * 2. The engine's Go gateway already implements exactly this contract —
+ *    `POST /invoke` returns an id, `GET /invoke/:id` reports on it — and
+ *    reimplementing the same accept/poll semantics in TypeScript would give two
+ *    definitions of one protocol.
+ * 3. It would put Temporal credentials in the orchestrator pod, which is the pod
+ *    that already holds the Kubernetes identity. `docs/orchestrator.md` reasons
+ *    explicitly about that pod's blast radius; leaving it unchanged is worth an
+ *    extra network hop.
+ *
+ * ## What this engine does NOT return
+ *
+ * The graph's `AgentState` carries the session fields `persistSession` writes:
+ * `selectedSkill`, `agentRunId`, `pendingIdentityLink`, `extractedContinuation`
+ * and the rest. This engine returns none of them, deliberately — the workflow
+ * holds that state itself (docs/adr/0001), which is the entire point of the
+ * change. `persistSession` treats every one as optional and merges rather than
+ * replaces, so an all-undefined outcome is a no-op rather than a clobber.
+ *
+ * Concretely that means a conversation running on this engine keeps its skill
+ * continuity, continuation tokens and pending links inside the workflow, and
+ * the Redis session record simply stays empty for it. Switching a live
+ * conversation between engines mid-flight would lose that state — which is why
+ * the flag is process-wide rather than per-request.
+ */
+export interface TemporalEngineOptions {
+  /** Base URL of the engine's gateway Service, e.g. `http://temporal-engine-gateway:8080`. */
+  baseUrl: string;
+  /** Bearer token presented to the gateway, if it resolves identities by token. */
+  token?: string;
+  /**
+   * Shared secret for the sender assertion. Without it a webhook-driven turn
+   * reaches the engine with no principal, so cross-entry-point credential
+   * sharing degrades — the same documented weaker mode as the gateway hop.
+   */
+  senderAssertionSecret?: string;
+  /** How long to keep polling one turn before giving up. */
+  timeoutMs?: number;
+  /** Injectable for tests; defaults to the global fetch. */
+  fetchImpl?: typeof fetch;
+}
+
+/** Poll interval. The engine's own poll route already waits ~2s server-side, so this is only the gap between waits. */
+const POLL_INTERVAL_MS = 500;
+const DEFAULT_TIMEOUT_MS = 30 * 60_000;
+
+interface InvokeAccepted {
+  id: string;
+  status: string;
+}
+
+interface InvokeRecord {
+  id: string;
+  status: "pending" | "succeeded" | "failed";
+  result?: string;
+  error?: string;
+  toolCalls?: { id: string; name: string; arguments: string }[];
+}
+
+export class TemporalEngine implements AgentGraphLike {
+  private readonly baseUrl: string;
+  private readonly fetchImpl: typeof fetch;
+
+  constructor(private readonly options: TemporalEngineOptions) {
+    this.baseUrl = options.baseUrl.replace(/\/$/, "");
+    this.fetchImpl = options.fetchImpl ?? fetch;
+  }
+
+  async invoke(input: AgentGraphInput): Promise<AgentState> {
+    const id = await this.start(input);
+    const record = await this.poll(id, input);
+
+    if (record.status === "failed") {
+      return { ...input, error: record.error ?? "the turn failed" } as AgentState;
+    }
+    if (record.toolCalls?.length) {
+      // The second non-error terminal shape (docs/adr/0035): the caller's own
+      // client has to run these.
+      return {
+        ...input,
+        pendingToolCalls: record.toolCalls.map((call) => ({
+          id: call.id,
+          name: call.name,
+          arguments: call.arguments,
+        })),
+      } as AgentState;
+    }
+    return { ...input, result: record.result } as AgentState;
+  }
+
+  /**
+   * Yields one terminal update, so a streaming caller still gets its answer.
+   *
+   * No per-node narration: those lines describe LangGraph node transitions,
+   * which do not exist here. The engine narrates its own turns over its own
+   * progress query, and plumbing that through this hop would mean a second
+   * streaming protocol for a status line. A streaming client on this engine
+   * therefore sees the reply rather than the running commentary — a real
+   * difference, recorded rather than papered over.
+   */
+  async stream(
+    input: AgentGraphInput,
+    _options: { streamMode: "updates" },
+  ): Promise<AsyncIterable<Record<string, Partial<AgentState>>>> {
+    const state = await this.invoke(input);
+    return {
+      async *[Symbol.asyncIterator]() {
+        yield { temporalEngine: state };
+      },
+    };
+  }
+
+  private async start(input: AgentGraphInput): Promise<string> {
+    const headers = this.headers();
+
+    // The sender login travels as a SIGNED assertion, not a body field —
+    // reusing the same `x-gateway-user-assertion` contract integration-gateway
+    // already uses (docs/adr/0030 §6), whose Go verifier is byte-compatible
+    // with `mintSenderAssertion`. It selects the caller's principal, and hence
+    // which stored credentials the run receives, so an internal hop is exactly
+    // as unsuited to trusting it unsigned as an external one.
+    if (input.senderLogin && this.options.senderAssertionSecret) {
+      headers[SENDER_ASSERTION_HEADER] = mintSenderAssertion(this.options.senderAssertionSecret, input.senderLogin);
+    }
+
+    const res = await this.fetchImpl(`${this.baseUrl}/invoke`, {
+      method: "POST",
+      headers,
+      body: JSON.stringify({
+        request: input.request,
+        sessionId: input.sessionId,
+        // This process owns the IntegrationRoute registry and has already
+        // matched it, so the target is named rather than re-derived. The engine
+        // still re-resolves it under the caller's own roles.
+        ...(input.forcedSkillId ? { forcedSkillId: input.forcedSkillId } : {}),
+        ...(input.forcedAgentId ? { forcedAgentId: input.forcedAgentId } : {}),
+      }),
+    });
+    if (!res.ok) {
+      throw new Error(`temporal engine /invoke failed: ${res.status}`);
+    }
+    return ((await res.json()) as InvokeAccepted).id;
+  }
+
+  private async poll(id: string, input: AgentGraphInput): Promise<InvokeRecord> {
+    const deadline = Date.now() + (this.options.timeoutMs ?? DEFAULT_TIMEOUT_MS);
+    for (;;) {
+      const res = await this.fetchImpl(`${this.baseUrl}/invoke/${encodeURIComponent(id)}`, {
+        headers: this.headers(),
+      });
+      if (!res.ok) {
+        throw new Error(`temporal engine /invoke/${id} failed: ${res.status}`);
+      }
+      const record = (await res.json()) as InvokeRecord;
+      if (record.status !== "pending") return record;
+
+      if (Date.now() >= deadline) {
+        // Deliberately not an engine error: the turn is still running and its
+        // answer stays collectable, because the record IS the workflow rather
+        // than this process's memory. Reported as a resumable pause, the same
+        // shape docs/adr/0033 settled on for an interrupted turn.
+        return {
+          id,
+          status: "succeeded",
+          result:
+            "That's taking longer than I can wait here — it's still running, " +
+            "and I'll have the answer on your next message.",
+        };
+      }
+      await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS));
+      void input;
+    }
+  }
+
+  private headers(): Record<string, string> {
+    const headers: Record<string, string> = { "content-type": "application/json" };
+    if (this.options.token) headers.authorization = `Bearer ${this.options.token}`;
+    return headers;
+  }
+}