Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 49 additions & 0 deletions src/agent/interchange-loop.test.ts
Original file line number Diff line number Diff line change
@@ -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");
}
});
});
69 changes: 69 additions & 0 deletions src/agent/interchange-loop.ts
Original file line number Diff line number Diff line change
@@ -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<string, unknown>;

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<InterchangeLoopStatus> {
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);
}
16 changes: 16 additions & 0 deletions src/exec/runner.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -469,6 +473,18 @@ export async function runExec(config: Config): Promise<ExecResult> {
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
Expand Down
Loading