diff --git a/apps/agent-orchestrator/src/config.ts b/apps/agent-orchestrator/src/config.ts index e93179f6..54bb8c3f 100644 --- a/apps/agent-orchestrator/src/config.ts +++ b/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", diff --git a/apps/agent-orchestrator/src/engine/temporal-engine.test.ts b/apps/agent-orchestrator/src/engine/temporal-engine.test.ts new file mode 100644 index 00000000..944a8085 --- /dev/null +++ b/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 { + 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; + 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; + 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[] = []; + 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" }); + }); +}); diff --git a/apps/agent-orchestrator/src/engine/temporal-engine.ts b/apps/agent-orchestrator/src/engine/temporal-engine.ts new file mode 100644 index 00000000..c5b3dd92 --- /dev/null +++ b/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 { + 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>>> { + const state = await this.invoke(input); + return { + async *[Symbol.asyncIterator]() { + yield { temporalEngine: state }; + }, + }; + } + + private async start(input: AgentGraphInput): Promise { + 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 { + 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 { + const headers: Record = { "content-type": "application/json" }; + if (this.options.token) headers.authorization = `Bearer ${this.options.token}`; + return headers; + } +} diff --git a/apps/agent-orchestrator/src/index.ts b/apps/agent-orchestrator/src/index.ts index 1e588294..985b2b58 100644 --- a/apps/agent-orchestrator/src/index.ts +++ b/apps/agent-orchestrator/src/index.ts @@ -36,12 +36,13 @@ import { OpenAiResponseComposer } from "./agent/response-composer.js"; import { OpenAiSkillFitChecker } from "./agent/skill-fit-checker.js"; import { OpenAiSkillSelector } from "./agent/skill-selector.js"; import { buildAgentGraph } from "./agent/graph.js"; +import { TemporalEngine } from "./engine/temporal-engine.js"; import { OpenAiTaskCompleter } from "./openai/task-completer.js"; import { InMemorySessionStore } from "./session/in-memory-session-store.js"; import { RedisSessionStore } from "./session/redis-session-store.js"; import type { SessionStore } from "./session/types.js"; import { clearAgentRunAwaitingReply, markAgentRunAwaitingReply } from "./session/inflight-agent-run.js"; -import { InvokeServer } from "./server.js"; +import { InvokeServer, type AgentGraphLike } from "./server.js"; import { retryWithBackoff } from "./retry.js"; import type { ToolDescriptor } from "./tool-descriptor.js"; import type { SkillDescriptor } from "./skills/types.js"; @@ -589,8 +590,33 @@ async function main(): Promise { ); callerToolPruneTimer.unref(); + // Which agent loop runs a turn (docs/adr/0036). `langgraph` is the default, + // so enabling the engine is always an explicit act and this whole block is + // inert until someone sets AGENT_ENGINE=temporal. + // + // Everything else about this process is shared either way: the OpenAI facade, + // /invoke, identity resolution, RBAC, the credential store, the + // authorization pre-flight, and both launchers. Only the loop moves. + let engine: AgentGraphLike = graph; + if (config.agentEngine === "temporal") { + if (!config.temporalEngineUrl) { + throw new Error("AGENT_ENGINE=temporal requires AGENT_TEMPORAL_ENGINE_URL"); + } + engine = new TemporalEngine({ + baseUrl: config.temporalEngineUrl, + ...(config.temporalEngineToken ? { token: config.temporalEngineToken } : {}), + // Reuses the assertion contract rather than trusting a login on an + // internal hop -- see TemporalEngine's own note. + ...(config.senderAssertionSecret ? { senderAssertionSecret: config.senderAssertionSecret } : {}), + timeoutMs: config.agentRunTimeoutSeconds * 1_000, + }); + console.log(`agent engine: temporal (${config.temporalEngineUrl})`); + } else { + console.log("agent engine: langgraph (in-process)"); + } + const invokeServer = new InvokeServer( - graph, + engine, sessionStore, taskCompleter, integrationRouteRegistry, diff --git a/engines/temporal/internal/gateway/invoke.go b/engines/temporal/internal/gateway/invoke.go index 90c6947c..fd7328a3 100644 --- a/engines/temporal/internal/gateway/invoke.go +++ b/engines/temporal/internal/gateway/invoke.go @@ -46,6 +46,18 @@ type invokeRequest struct { Request string `json:"request"` SessionID string `json:"sessionId"` Event map[string]any `json:"event"` + + // ForcedSkillID / ForcedAgentID let a caller that has ALREADY matched a + // route name the target directly, instead of sending the event descriptor + // and having it matched here again. + // + // This is how the TypeScript orchestrator drives this engine: it owns the + // IntegrationRoute registry, so re-matching here would put routing policy in + // two places — the duplication ADR 0024 rejected when it declined to let the + // gateway launch AgentRuns directly. They are still only a routing hint: the + // workflow re-resolves whatever is named under the caller's own roles. + ForcedSkillID string `json:"forcedSkillId,omitempty"` + ForcedAgentID string `json:"forcedAgentId,omitempty"` } type invokeAccepted struct { @@ -114,8 +126,10 @@ func shapeInvokeTurn( senderLogin = strings.TrimSpace(raw) } - var forcedSkillID, forcedAgentID string - if routes != nil && len(req.Event) > 0 { + // An explicitly named target wins: the caller already did the matching, and + // re-deciding it here could only disagree. + forcedSkillID, forcedAgentID := req.ForcedSkillID, req.ForcedAgentID + if forcedSkillID == "" && forcedAgentID == "" && routes != nil && len(req.Event) > 0 { fields := catalog.EventFields(req.Event) if source, event := fields["source"], fields["event"]; source != "" && event != "" { if route, ok := routes.Match(source, event, fields["action"], fields["labelName"]); ok { diff --git a/engines/temporal/internal/gateway/invoke_test.go b/engines/temporal/internal/gateway/invoke_test.go index e05885a2..7973340e 100644 --- a/engines/temporal/internal/gateway/invoke_test.go +++ b/engines/temporal/internal/gateway/invoke_test.go @@ -338,3 +338,35 @@ func TestToolCallsPayloadShape(t *testing.T) { require.Equal(t, 0, payload[0].Index) require.Equal(t, 1, payload[1].Index) } + +// A caller that already matched a route names the target directly, and that +// wins over re-matching here. This is how the TypeScript orchestrator drives +// the engine: it owns the route registry, so re-deciding here could only +// disagree with a decision already made. +func TestShapeInvokeTurnPrefersAnExplicitlyNamedTarget(t *testing.T) { + turn, err := shapeInvokeTurn( + invokeRequest{ + Request: "Triage acme/widgets#7", + ForcedAgentID: "some-other-agent", + // An event that WOULD match the triage route if it were consulted. + Event: issueEvent(), + }, + "", "", triageRoutes(t), testCaller, time.Now(), + ) + require.NoError(t, err) + require.Equal(t, "some-other-agent", turn.ForcedAgentID) + require.Equal(t, "Triage acme/widgets#7", turn.Message, + "the route's promptTemplate must not overwrite a request the caller already shaped") +} + +// With no explicit target, the event descriptor is still matched here — which +// is what a caller driving the engine directly relies on. +func TestShapeInvokeTurnStillMatchesWhenNoTargetIsNamed(t *testing.T) { + turn, err := shapeInvokeTurn( + invokeRequest{Request: "an issue was labeled", Event: issueEvent()}, + "", "", triageRoutes(t), testCaller, time.Now(), + ) + require.NoError(t, err) + require.Equal(t, "claude-code-swe-agent", turn.ForcedAgentID) + require.Equal(t, "Triage acme/widgets#7: Crash on save", turn.Message) +}