diff --git a/src/agent/interchange-loop.test.ts b/src/agent/interchange-loop.test.ts new file mode 100644 index 000000000..dad068f69 --- /dev/null +++ b/src/agent/interchange-loop.test.ts @@ -0,0 +1,49 @@ +import { describe, expect, test } from "bun:test"; +import { + INTERCHANGE_LOOP_ENV_VAR, + isInterchangeLoopEnabled, + resolveInterchangeLoopRunner, +} from "./interchange-loop.js"; + +function envWith(value: string | undefined): NodeJS.ProcessEnv { + const env: NodeJS.ProcessEnv = {}; + if (value !== undefined) env[INTERCHANGE_LOOP_ENV_VAR] = value; + return env; +} + +describe("isInterchangeLoopEnabled", () => { + test("is disabled by default", () => { + expect(isInterchangeLoopEnabled(envWith(undefined))).toBe(false); + }); + + test.each(["1", "true", "TRUE", " True "])("is enabled for %j", (value) => { + expect(isInterchangeLoopEnabled(envWith(value))).toBe(true); + }); + + test.each(["0", "false", "yes", ""])("stays disabled for %j", (value) => { + expect(isInterchangeLoopEnabled(envWith(value))).toBe(false); + }); + + test("never reads process.env when an env is passed", () => { + const prior = process.env.CORBITS_INTERCHANGE_LOOP; + process.env.CORBITS_INTERCHANGE_LOOP = "1"; + try { + expect(isInterchangeLoopEnabled(envWith(undefined))).toBe(false); + } finally { + if (prior === undefined) delete process.env.CORBITS_INTERCHANGE_LOOP; + else process.env.CORBITS_INTERCHANGE_LOOP = prior; + } + }); +}); + +describe("resolveInterchangeLoopRunner", () => { + test("never throws and reports unavailable without the workflow package", async () => { + // `@intx/workflow` is not installed in this tree (it arrives with the + // 0.4.0 hub), so the seam must fall back instead of rejecting. + const status = await resolveInterchangeLoopRunner(); + expect(status.status).toBe("unavailable"); + if (status.status === "unavailable") { + expect(status.reason).toContain("@intx/workflow"); + } + }); +}); diff --git a/src/agent/interchange-loop.ts b/src/agent/interchange-loop.ts new file mode 100644 index 000000000..407c2357c --- /dev/null +++ b/src/agent/interchange-loop.ts @@ -0,0 +1,69 @@ +/** + * Optional Interchange-driven core loop (CL-8267). + * + * The reactor (`@intx/agent` + `ChatDirector`) remains the only loop driver. + * This module is the opt-in seam for driving it through Interchange's + * `runWorkflow`/`runLocal` instead, without waiting on the full 0.4.0 hub. + * + * Gating: enabled only when `CORBITS_INTERCHANGE_LOOP` is `1` or `true`. + * Resolution is lazy and total — `@intx/workflow` is not a dependency of this + * package yet, so when it cannot be loaded the caller keeps the reactor path. + * Nothing here changes behavior unless the flag is set. + */ + +export const INTERCHANGE_LOOP_ENV_VAR = "CORBITS_INTERCHANGE_LOOP"; + +const INTERCHANGE_WORKFLOW_SPECIFIER = "@intx/workflow"; + +export type InterchangeLoopStatus = + | { status: "driven"; via: "runWorkflow" | "runLocal" } + | { status: "unavailable"; reason: string }; + +/** True only when the operator explicitly opts into the Interchange loop. */ +export function isInterchangeLoopEnabled( + env: NodeJS.ProcessEnv = process.env, +): boolean { + const raw = env[INTERCHANGE_LOOP_ENV_VAR]?.trim().toLowerCase(); + return raw === "1" || raw === "true"; +} + +type WorkflowModule = Record; + +function pickDriver(loaded: WorkflowModule): InterchangeLoopStatus { + // Prefer the hub-backed entry point; the in-memory runner is the local + // fallback. Both names come from upstream `@intx/workflow`. + if (typeof loaded["runWorkflow"] === "function") + return { status: "driven", via: "runWorkflow" }; + if (typeof loaded["runLocal"] === "function") + return { status: "driven", via: "runLocal" }; + return { + status: "unavailable", + reason: `${INTERCHANGE_WORKFLOW_SPECIFIER} loaded but exports neither runWorkflow nor runLocal`, + }; +} + +/** + * Resolve the Interchange loop driver. Never throws and never imports + * statically — the package is an optional peer, absent until the workflow + * runtime lands. Callers fall through to the reactor on `unavailable`. + */ +export async function resolveInterchangeLoopRunner(): Promise { + let loaded: unknown; + try { + loaded = (await import( + INTERCHANGE_WORKFLOW_SPECIFIER + )) as unknown as WorkflowModule; + } catch (err: unknown) { + const detail = err instanceof Error ? err.message : String(err); + return { + status: "unavailable", + reason: `${INTERCHANGE_WORKFLOW_SPECIFIER} is not installed (${detail})`, + }; + } + if (typeof loaded !== "object" || loaded === null) + return { + status: "unavailable", + reason: `${INTERCHANGE_WORKFLOW_SPECIFIER} did not load as a module`, + }; + return pickDriver(loaded as unknown as WorkflowModule); +} diff --git a/src/exec/runner.ts b/src/exec/runner.ts index 1f4263f00..3316a49a7 100644 --- a/src/exec/runner.ts +++ b/src/exec/runner.ts @@ -20,6 +20,10 @@ import { formatMcpTrustQuestion } from "../trust/project-trust.js"; import { DIRECTOR_REGISTRY } from "../agent/directors/registry.js"; import type { DirectorId, DirectorPackage } from "../agent/directors/types.js"; import { submitOutputDefinition } from "../agent/director.js"; +import { + isInterchangeLoopEnabled, + resolveInterchangeLoopRunner, +} from "../agent/interchange-loop.js"; import { handleChatDirectorEvent } from "../agent/chat-event-subscribers.js"; import { CodexRefreshLockError, @@ -469,6 +473,18 @@ export async function runExec(config: Config): Promise { const sessionId = config.sessionId.length > 0 ? config.sessionId : generateSessionId(); const startedAt = Date.now(); + // CL-8267: optional Interchange loop seam. Off by default; when the + // operator opts in but the workflow runtime is absent, stay on the + // reactor path with a warning instead of failing the run. + if (isInterchangeLoopEnabled()) { + const loop = await resolveInterchangeLoopRunner(); + if (loop.status === "unavailable") { + logger.warn( + "Interchange loop requested but {reason}; continuing on reactor path", + { reason: loop.reason }, + ); + } + } const workdir = sessionContextDir(config.cwd, sessionId); // Setup runs before the main try below: initSessionDir/loadState hit disk // before any session_end coverage exists, so an EACCES/EROFS here would