Codemode For AI Agents, Explained For Trading CTOs: What Pi 1.0 Shipped, Where The Trust Boundary Moves, And How To Adopt It Safely

// The model writes this, not your team. An illustrative codemode program for "triage last night's client-services inbox".
// It runs inside the QuickJS sandbox. tools, searchTools, store and load are the only globals it has.
// An earlier call probed five emails to learn their shape; this one processes the whole night.
const { emails } = await tools.list_overnight_emails({
  mailbox: "client-services",
  since: "2026-10-13T17:00:00Z",
  limit: 2000,
});

const results = [];
const failed = [];
let next = 0;
async function worker() {
  while (next < emails.length) {
    const email = emails[next++];
    try {
      const c = await tools.classify({
        taxonomy: "client_email_v1",
        text: `${email.subject}\n\n${email.body}`,
      });
      results.push({ id: email.id, receivedAt: email.receivedAt, label: c.label, confidence: c.confidence, route: c.route });
    } catch (e) {
      failed.push({ id: email.id, error: String(e && e.message ? e.message : e) });
    }
  }
}
await Promise.all([worker(), worker(), worker(), worker()]);
store("triage_2026_10_14", { results, failed });

const counts = {};
for (const r of results) counts[r.label] = (counts[r.label] ?? 0) + 1;
const forPeople = results.filter((r) => r.route === "human");
const complaints = forPeople
  .filter((r) => r.label === "possible_complaint")
  .sort((a, b) => a.receivedAt.localeCompare(b.receivedAt));
return {
  total: emails.length,
  classified: results.length,
  failed: failed.length,
  counts,
  routedToPeople: forPeople.length,
  oldestPossibleComplaint: complaints.length ? complaints[0].id : null,
  detail: "per-email results stored under triage_2026_10_14",
};

// harness/codemode.ts
// The harness side of codemode: runs a program the model wrote inside a QuickJS sandbox.
// Targets quickjs-emscripten 0.32.0, zod 4.6 and Node.js 22.
import { createHash, randomUUID } from "node:crypto";
import { appendFile } from "node:fs/promises";
import {
  getQuickJS,
  shouldInterruptAfterDeadline,
  type QuickJSDeferredPromise,
  type QuickJSHandle,
} from "quickjs-emscripten";
import { z } from "zod";

export interface ReadTool<I extends z.ZodType = z.ZodType> {
  effect: "read"; // the only effect this registry accepts
  description: string;
  input: I;
  maxCallsPerRun: number;
  run(args: z.infer<I>, signal: AbortSignal): Promise<unknown>;
}
export type Registry = Record<string, ReadTool>;

export interface Limits {
  memoryBytes: number; // QuickJS heap ceiling
  stackBytes: number;
  timeoutMs: number; // wall clock for the whole invocation, tool waits included
  maxConcurrency: number; // in-flight tool calls; Pi uses four
  maxToolCalls: number; // across all tools in one invocation
  maxStoreBytes: number; // per store() call
  maxReturnBytes: number; // what goes back into the model's context
}

export const DEFAULT_LIMITS: Limits = {
  memoryBytes: 32 * 1024 * 1024,
  stackBytes: 512 * 1024,
  timeoutMs: 120_000,
  maxConcurrency: 4,
  maxToolCalls: 2_500,
  maxStoreBytes: 2 * 1024 * 1024,
  maxReturnBytes: 12_000,
};

export interface TranscriptEntry {
  kind: "codemode_store";
  runId: string;
  key: string;
  json: string;
  at: string;
}

export interface Session {
  id: string;
  stored: Map<string, string>; // rebuilt from the transcript when a session loads
  appendTranscript(entry: TranscriptEntry): Promise<void>;
}

export interface RunResult {
  runId: string;
  ok: boolean;
  value?: string; // JSON, truncated to maxReturnBytes
  error?: string;
  toolCalls: number;
  ms: number;
}

const AUDIT_FILE = process.env.CODEMODE_AUDIT_FILE ?? "codemode-audit.jsonl";
const sha256 = (s: string) => createHash("sha256").update(s).digest("hex");
const audit = (entry: Record<string, unknown>) =>
  appendFile(AUDIT_FILE, JSON.stringify({ at: new Date().toISOString(), ...entry }) + "\n");

class Semaphore {
  private active = 0;
  private waiting: Array<() => void> = [];
  constructor(private readonly max: number) {}
  async acquire(): Promise<() => void> {
    if (this.active < this.max) this.active++;
    else await new Promise<void>((resolve) => this.waiting.push(resolve)); // slot handed over on release
    return () => {
      const next = this.waiting.shift();
      if (next) next();
      else this.active--;
    };
  }
}

// Runs inside the sandbox before the model's code. It captures the host functions,
// deletes them from the global scope and exposes a small, frozen API.
const PRELUDE = `
"use strict";
(() => {
  const call = globalThis.__call, put = globalThis.__store, get = globalThis.__load, search = globalThis.__search;
  const names = JSON.parse(globalThis.__names);
  for (const k of ["__call", "__store", "__load", "__search", "__names"]) delete globalThis[k];
  const tools = {};
  for (const name of names) tools[name] = async (args = {}) => JSON.parse(await call(name, JSON.stringify(args)));
  globalThis.tools = Object.freeze(tools);
  globalThis.searchTools = (query) => JSON.parse(search(String(query)));
  globalThis.store = (key, value) => { put(String(key), JSON.stringify(value)); };
  globalThis.load = (key) => { const s = get(String(key)); return s === undefined ? undefined : JSON.parse(s); };
})();
`;

export async function runCodemode(
  program: string,
  registry: Registry,
  session: Session,
  limits: Limits = DEFAULT_LIMITS,
): Promise<RunResult> {
  for (const [name, tool] of Object.entries(registry)) {
    if (tool.effect !== "read") throw new Error(`refusing to expose ${name}: codemode tools are read-only`);
  }
  const runId = randomUUID();
  const started = Date.now();
  const abort = new AbortController();
  const gate = new Semaphore(limits.maxConcurrency);
  const perTool = new Map<string, number>();
  const pending = new Set<QuickJSDeferredPromise>();
  const writes: Promise<void>[] = [];
  let seq = 0;
  let disposed = false;

  const QuickJS = await getQuickJS();
  const runtime = QuickJS.newRuntime(); // no fetch, no fs, no timers: QuickJS exposes nothing by default
  runtime.setMemoryLimit(limits.memoryBytes);
  runtime.setMaxStackSize(limits.stackBytes);
  runtime.setInterruptHandler(shouldInterruptAfterDeadline(started + limits.timeoutMs));
  const vm = runtime.newContext();

  const drain = () => {
    if (disposed) return;
    const r = runtime.executePendingJobs();
    if (r.error) r.error.dispose();
  };

  // Every inner tool call passes through here: lookup, caps, validation, concurrency, audit.
  async function invoke(name: string, argsJson: string): Promise<string> {
    const callSeq = ++seq;
    const t0 = Date.now();
    let args: unknown = argsJson;
    try {
      const tool = registry[name];
      if (!tool) throw new Error(`unknown tool: ${name}`);
      if (callSeq > limits.maxToolCalls) throw new Error(`budget of ${limits.maxToolCalls} tool calls exhausted`);
      const used = (perTool.get(name) ?? 0) + 1;
      perTool.set(name, used);
      if (used > tool.maxCallsPerRun) throw new Error(`${name} is capped at ${tool.maxCallsPerRun} calls per run`);
      const parsed = tool.input.safeParse(JSON.parse(argsJson));
      if (!parsed.success) throw new Error(`${name}: ${z.prettifyError(parsed.error)}`);
      args = parsed.data;
      const release = await gate.acquire();
      try {
        if (abort.signal.aborted) throw new Error("run cancelled");
        const out = JSON.stringify(await tool.run(parsed.data, abort.signal)) ?? "null";
        await audit({ runId, sessionId: session.id, seq: callSeq, tool: name, args, ok: true,
          ms: Date.now() - t0, resultBytes: out.length, resultSha256: sha256(out) });
        return out;
      } finally {
        release();
      }
    } catch (err) {
      await audit({ runId, sessionId: session.id, seq: callSeq, tool: name, args, ok: false,
        ms: Date.now() - t0, error: String(err instanceof Error ? err.message : err) });
      throw err;
    }
  }

  function settle(d: QuickJSDeferredPromise, make: () => QuickJSHandle, ok: boolean) {
    pending.delete(d);
    if (disposed || !d.alive) return; // the run already ended; drop late results
    const h = make();
    if (ok) d.resolve(h);
    else d.reject(h);
    h.dispose();
    drain();
  }

  const hostFns: Array<[string, QuickJSHandle]> = [
    ["__call", vm.newFunction("__call", (nameH, argsH) => {
      const d = vm.newPromise();
      pending.add(d);
      invoke(vm.getString(nameH), vm.getString(argsH)).then(
        (json) => settle(d, () => vm.newString(json), true),
        (err) => settle(d, () => vm.newError({ name: "ToolError", message: String(err?.message ?? err) }), false),
      );
      return d.handle;
    })],
    ["__store", vm.newFunction("__store", (keyH, jsonH) => {
      const key = vm.getString(keyH);
      const json = vm.getString(jsonH);
      if (json.length > limits.maxStoreBytes) throw new Error(`store("${key}") exceeds ${limits.maxStoreBytes} bytes`);
      session.stored.set(key, json);
      writes.push(session.appendTranscript({ kind: "codemode_store", runId, key, json, at: new Date().toISOString() }));
    })],
    ["__load", vm.newFunction("__load", (keyH) => {
      const json = session.stored.get(vm.getString(keyH));
      return json === undefined ? vm.undefined : vm.newString(json);
    })],
    ["__search", vm.newFunction("__search", (queryH) => {
      const q = vm.getString(queryH).toLowerCase();
      const hits = Object.entries(registry)
        .filter(([n, t]) => n.includes(q) || t.description.toLowerCase().includes(q))
        .map(([name, t]) => ({ name, description: t.description, input: z.toJSONSchema(t.input) }));
      return vm.newString(JSON.stringify(hits));
    })],
  ];
  for (const [name, fn] of hostFns) {
    vm.setProp(vm.global, name, fn);
    fn.dispose();
  }
  vm.newString(JSON.stringify(Object.keys(registry))).consume((h) => vm.setProp(vm.global, "__names", h));
  vm.unwrapResult(vm.evalCode(PRELUDE, "prelude.js")).dispose();

  let timer: NodeJS.Timeout | undefined;
  let promise: QuickJSHandle | undefined;
  let result: RunResult;
  try {
    // The model writes a function body, so top-level await and return both work.
    promise = vm.unwrapResult(vm.evalCode(`(async () => {\n${program}\n})()`, "agent.js"));
    const outcome = vm.resolvePromise(promise);
    drain();
    const timeout = new Promise<"timeout">((r) => { timer = setTimeout(() => r("timeout"), limits.timeoutMs); });
    const settled = await Promise.race([outcome, timeout]);
    if (settled === "timeout") throw new Error(`timed out after ${limits.timeoutMs} ms`);
    if (settled.error) {
      const e = settled.error.consume((h) => vm.dump(h));
      throw new Error(`${e?.name ?? "Error"}: ${e?.message ?? JSON.stringify(e)}`);
    }
    const json = JSON.stringify(settled.value.consume((h) => vm.dump(h))) ?? "null";
    const value = json.length <= limits.maxReturnBytes ? json
      : json.slice(0, limits.maxReturnBytes) + ` [truncated ${json.length - limits.maxReturnBytes} chars: store() the full result and load() it]`;
    result = { runId, ok: true, value, toolCalls: seq, ms: Date.now() - started };
  } catch (err) {
    result = { runId, ok: false, error: String(err instanceof Error ? err.message : err), toolCalls: seq, ms: Date.now() - started };
  } finally {
    clearTimeout(timer);
    abort.abort(); // cancels tool calls still in flight
    disposed = true;
    for (const d of pending) d.dispose();
    promise?.dispose();
    vm.dispose();
    runtime.dispose();
  }
  await Promise.allSettled(writes);
  // The program itself is evidence: log the exact code the model wrote next to its inner calls.
  await audit({ runId, sessionId: session.id, kind: "codemode_run", ok: result.ok, error: result.error,
    toolCalls: result.toolCalls, ms: result.ms, programSha256: sha256(program), program });
  return result;
}

// harness/tools.ts
// The typed, read-only registry the sandbox can see. No tool here can place, amend or cancel
// an order, move money or send a message: those live behind the approval-gated order gateway.
// Targets zod 4.6, openai 7.33.0 (the Decisions API needs 7.30.0 or later) and Node.js 22.
import { readFileSync } from "node:fs";
import OpenAI from "openai";
import { z } from "zod";
import type { ReadTool, Registry } from "./codemode";

const OPS_API = process.env.OPS_API_URL ?? "";              // read-only replica, separate credentials
const OPS_TOKEN = process.env.OPS_API_READONLY_TOKEN ?? "";
if (!OPS_API || !OPS_TOKEN) throw new Error("OPS_API_URL and OPS_API_READONLY_TOKEN must be set");
const openai = new OpenAI(); // reads OPENAI_API_KEY from the environment

async function opsGet(path: string, params: Record<string, string | number>, signal: AbortSignal) {
  const url = new URL(path, OPS_API);
  for (const [k, v] of Object.entries(params)) url.searchParams.set(k, String(v));
  const res = await fetch(url, { headers: { authorization: `Bearer ${OPS_TOKEN}` }, signal });
  if (!res.ok) throw new Error(`${path} returned HTTP ${res.status}`);
  return res.json();
}

// Label sets belong to the harness and carry a version, so every calibration run maps to one taxonomy.
const TAXONOMIES = {
  client_email_v1: {
    instructions: "What is the client asking the firm to do? Judge the request, not the tone.",
    choices: [
      { value: "possible_complaint", description: "Dissatisfaction with the firm, a fill, a charge or the service" },
      { value: "payments", description: "Deposits, withdrawals, payment status, card or bank details" },
      { value: "account_admin", description: "Login, documents, verification, address or personal details" },
      { value: "platform_question", description: "How the platform, charts, order types or margin figures work" },
      { value: "other", description: "Anything that fits none of the categories above" },
    ],
  },
} as const;

// Written by calibration/calibrate.py. autoRouteAt is null until the classifier earns it.
const CALIBRATION: { autoRouteAt: number | null } = JSON.parse(
  readFileSync(process.env.CALIBRATION_FILE ?? "calibration/client_email_v1.json", "utf8"),
);
const ALWAYS_HUMAN = new Set(["possible_complaint"]); // whatever the confidence says

const tool = <I extends z.ZodType>(t: ReadTool<I>): ReadTool<I> => t;

export const registry: Registry = {
  list_overnight_emails: tool({
    effect: "read",
    description: "Client emails received by a shared mailbox since a time. Returns { emails: [{ id, receivedAt, subject, body }] }; bodies are cut to 4,000 characters and account numbers are masked upstream.",
    input: z.object({
      mailbox: z.enum(["client-services", "dealing-desk"]),
      since: z.iso.datetime(),
      limit: z.number().int().min(1).max(2000),
    }),
    maxCallsPerRun: 5,
    run: (a, signal) => opsGet("/v1/mailboxes/emails", a, signal),
  }),
  get_news: tool({
    effect: "read",
    description: "Headlines and summaries from the licensed news feed since a time. Returns { items: [{ id, publishedAt, headline, summary, instruments }] }.",
    input: z.object({ since: z.iso.datetime(), limit: z.number().int().min(1).max(2000) }),
    maxCallsPerRun: 5,
    run: (a, signal) => opsGet("/v1/news", a, signal),
  }),
  get_positions: tool({
    effect: "read",
    description: "End-of-day positions for one account: { positions: [{ instrument, quantity, avgPrice, currency }] }.",
    input: z.object({ account: z.string().regex(/^[A-Z0-9-]{4,32}$/), asOf: z.iso.date() }),
    maxCallsPerRun: 500,
    run: (a, signal) => opsGet(`/v1/accounts/${a.account}/positions`, { asOf: a.asOf }, signal),
  }),
  classify: tool({
    effect: "read",
    description: "Classify text against a harness-owned taxonomy. Returns { label, confidence, route }, where route is 'auto' or 'human' and is decided by the harness, not the caller.",
    input: z.object({ taxonomy: z.enum(["client_email_v1"]), text: z.string().min(1).max(8000) }),
    maxCallsPerRun: 2000,
    run: async ({ taxonomy, text }, signal) => {
      const t = TAXONOMIES[taxonomy];
      const decision = await openai.decisions.create(
        {
          model: "gpt-6-luna",
          input: text,
          questions: [{ type: "choice", name: "label", instructions: t.instructions, choices: [...t.choices] }],
        },
        { signal },
      );
      const a = decision.answers[0];
      if (!a || a.type !== "choice") return { taxonomy, label: null, confidence: 0, route: "human" };
      const label = String(a.choice);
      const cut = CALIBRATION.autoRouteAt ?? Number.POSITIVE_INFINITY;
      const route = ALWAYS_HUMAN.has(label) || a.confidence < cut ? "human" : "auto";
      return { taxonomy, label, confidence: a.confidence, route };
    },
  }),
};

"""calibration/calibrate.py

Reliability bins, expected calibration error (ECE) and an abstain-to-human
threshold for the classifier the harness calls inside codemode.
Python 3.11 or later, standard library only.

Input is a labelled sample exported from the audit log and reviewed by people:
    id,predicted,confidence,true_label
Output is the JSON file harness/tools.ts reads. autoRouteAt stays null, and
every item goes to a person, until the classifier earns a threshold.

    python calibration/calibrate.py labelled/client_email_v1.csv \
        --taxonomy client_email_v1 --target 0.98 --max-ece 0.05 \
        --out calibration/client_email_v1.json
"""
from __future__ import annotations

import argparse
import csv
import json
import math
from dataclasses import asdict, dataclass
from datetime import date

MIN_SAMPLE = 400      # below this, no threshold is worth trusting
MIN_AUTO_SLICE = 150  # the auto-routed slice must be big enough to measure


@dataclass(frozen=True)
class Item:
    confidence: float
    correct: bool


@dataclass(frozen=True)
class Bin:
    lower: float
    upper: float
    n: int
    mean_confidence: float
    accuracy: float


def load(path: str) -> list[Item]:
    items: list[Item] = []
    with open(path, newline="", encoding="utf-8") as f:
        for row in csv.DictReader(f):
            c = float(row["confidence"])
            if not 0.0 <= c <= 1.0:
                raise ValueError(f"row {row['id']}: confidence {c} is outside [0, 1]")
            items.append(Item(c, row["predicted"].strip() == row["true_label"].strip()))
    if len(items) < MIN_SAMPLE:
        raise SystemExit(f"{len(items)} labelled items; label at least {MIN_SAMPLE} first")
    return items


def reliability(items: list[Item], n_bins: int = 10) -> list[Bin]:
    bins = []
    for b in range(n_bins):
        lo, hi = b / n_bins, (b + 1) / n_bins
        members = [i for i in items if lo <= i.confidence < hi or (hi == 1.0 and i.confidence == 1.0)]
        if members:
            bins.append(Bin(lo, hi, len(members),
                            sum(i.confidence for i in members) / len(members),
                            sum(i.correct for i in members) / len(members)))
    return bins


def expected_calibration_error(bins: list[Bin], total: int) -> float:
    return sum(b.n / total * abs(b.accuracy - b.mean_confidence) for b in bins)


def wilson_lower(successes: int, n: int, z: float = 1.96) -> float:
    """95% lower bound on a proportion; honest about small slices."""
    p = successes / n
    centre = p + z * z / (2 * n)
    margin = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n))
    return (centre - margin) / (1 + z * z / n)


def abstain_threshold(items: list[Item], target: float) -> tuple[float, float, float, float] | None:
    """Lowest cut-off whose auto-routed slice is at least `target` accurate at the 95% lower bound."""
    for t in sorted({round(i.confidence, 3) for i in items}):
        auto = [i for i in items if i.confidence >= t]
        if len(auto) < MIN_AUTO_SLICE:
            return None
        right = sum(i.correct for i in auto)
        lower = wilson_lower(right, len(auto))
        if lower >= target:
            return t, len(auto) / len(items), right / len(auto), lower
    return None


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("csv")
    ap.add_argument("--taxonomy", required=True)
    ap.add_argument("--target", type=float, default=0.98, help="required accuracy of auto-routed items")
    ap.add_argument("--max-ece", type=float, default=0.05)
    ap.add_argument("--out", required=True)
    args = ap.parse_args()

    items = load(args.csv)
    bins = reliability(items)
    ece = expected_calibration_error(bins, len(items))
    print(f"{len(items)} items, accuracy {sum(i.correct for i in items) / len(items):.3f}, ECE {ece:.3f}")
    for b in bins:
        print(f"  [{b.lower:.1f}, {b.upper:.1f})  n={b.n:5d}  says {b.mean_confidence:.3f}  "
              f"is right {b.accuracy:.3f}  gap {b.accuracy - b.mean_confidence:+.3f}")
    sure = [i for i in items if i.confidence >= 0.99]
    if sure:
        print(f"  claims 99% or more on {len(sure)} items; right on {sum(i.correct for i in sure) / len(sure):.1%}")

    auto_route_at = None
    if ece > args.max_ece:
        print(f"ECE {ece:.3f} is above {args.max_ece}: every item goes to a person")
    elif (found := abstain_threshold(items, args.target)) is None:
        print(f"no cut-off reaches {args.target:.0%} at the 95% lower bound: every item goes to a person")
    else:
        auto_route_at, coverage, accuracy, lower = found
        print(f"auto-route at confidence >= {auto_route_at}: {coverage:.1%} of items, "
              f"accuracy {accuracy:.3f}, 95% lower bound {lower:.3f}; the rest go to a person")

    with open(args.out, "w", encoding="utf-8") as f:
        json.dump({"taxonomy": args.taxonomy, "autoRouteAt": auto_route_at, "ece": round(ece, 4),
                   "n": len(items), "target": args.target, "calibratedOn": date.today().isoformat(),
                   "bins": [asdict(b) for b in bins]}, f, indent=2)
    return 0 if auto_route_at is not None else 1  # non-zero fails the CI job that refreshes thresholds


if __name__ == "__main__":
    raise SystemExit(main())