Land the workflow-run substrate latency fix and remove the trivial agent path

Makefile

@@ -16,7 +16,7 @@ lint: FORCE
 
 test: FORCE
 	bun test packages/ apps/ bin/ tests/agent/ tests/agent-audit-log/ tests/agent-blob-spill/ tests/agent-common/ tests/agent-multi-provider/ tests/agent-quickstart/ tests/agent-resume/ tests/agent-rewind/ tests/agent-rich-tool/ tests/agent-structured-payload/ tests/coding-agent/ tests/hub-agent/lib/ tests/inference-testing/ tests/tool-packaging/ tests/workflow/
-	bun test --timeout 120000 tests/hub-agent/deploy-flow.test.ts tests/workflow-deploy/trivial-roundtrip.test.ts tests/workflow-deploy/multistep-signal.test.ts tests/workflow-deploy/single-step-real-agent.test.ts tests/workflow-deploy/single-step-message-input.test.ts tests/workflow-deploy/single-step-posix-tool.test.ts tests/workflow-deploy/single-step-grants-bridge.test.ts tests/workflow-deploy/single-step-event-threading.test.ts tests/workflow-deploy/single-step-conversation-durability.test.ts tests/workflow-deploy/single-step-full-lifecycle.test.ts tests/workflow-deploy/cross-process-custom-adapter.test.ts tests/workflow-deploy/conversation-state-wal.test.ts tests/workflow-deploy/drain-roundtrip.test.ts tests/workflow-deploy/child-workflow-roundtrip.test.ts tests/workflow-deploy/unresolvable-director.test.ts tests/workflow-deploy/fifo-mail.test.ts tests/workflow-deploy/mail-edge-cases.test.ts tests/inference/ tests/skill-attachment-flow.test.ts tests/hub-api/ tests/db/
+	bun test --timeout 120000 tests/workflow-deploy/multistep-signal.test.ts tests/workflow-deploy/single-step-real-agent.test.ts tests/workflow-deploy/instance-reroute-real-agent.test.ts tests/workflow-deploy/instance-failover-real-agent.test.ts tests/workflow-deploy/multistep-signed-send.test.ts tests/workflow-deploy/single-step-message-input.test.ts tests/workflow-deploy/single-step-posix-tool.test.ts tests/workflow-deploy/single-step-grants-bridge.test.ts tests/workflow-deploy/single-step-event-threading.test.ts tests/workflow-deploy/single-step-conversation-durability.test.ts tests/workflow-deploy/single-step-full-lifecycle.test.ts tests/workflow-deploy/cross-process-custom-adapter.test.ts tests/workflow-deploy/conversation-state-wal.test.ts tests/workflow-deploy/drain-roundtrip.test.ts tests/workflow-deploy/child-workflow-roundtrip.test.ts tests/workflow-deploy/unresolvable-director.test.ts tests/workflow-deploy/fifo-mail.test.ts tests/workflow-deploy/mail-edge-cases.test.ts tests/inference/ tests/skill-attachment-flow.test.ts tests/hub-api/ tests/db/
 
 test-load: FORCE
 	bun test --timeout 300000 tests/workflow-deploy/fifo-mail-load.test.ts

apps/hub/src/index.ts

@@ -1,4 +1,4 @@
-import { createDB, createGrantStore } from "@intx/db";
+import { createDB } from "@intx/db";
 import { createApp, createAuth } from "@intx/hub-api";
 import {
   createAgentRepoStore,
@@ -35,7 +35,6 @@ const { db } = createDB({
 });
 
 const auth = createAuth(db);
-const grantStore = createGrantStore(db);
 
 const hubDataDir = process.env["HUB_DATA_DIR"];
 if (!hubDataDir) {
@@ -141,7 +140,6 @@ createHubSessionOrchestrator({
   router: sidecarRouter,
   db,
   eventCollectors,
-  grantStore,
   agentRepoStore,
 });
 

apps/sidecar/src/agent-key-registration-lifecycle.test.ts

@@ -29,7 +29,13 @@ import {
 } from "@intx/workflow-host";
 import type { AgentDeployFrame } from "@intx/types/sidecar";
 
-import { createSidecarDeployRouter } from "./workflow-host-wiring";
+import { type } from "arktype";
+
+import {
+  createSidecarDeployRouter,
+  deriveDeploymentId,
+} from "./workflow-host-wiring";
+import { WorkflowDeploymentRecord } from "./workflow-deployment-record";
 import {
   createMultistepDrainRouter,
   createMultistepMailRouter,
@@ -219,34 +225,33 @@ describe("agent signing-key registration lifecycle on the host transport", () =>
     const repoStore = createSpawnTestRepoStore(tempBase);
 
     const router = createSidecarDeployRouter({
-      // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- multi-step branch never invokes sessions
+      // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- the single-step branch invokes only initRepo (head deploy-tree repo); provisionAgent/persistHubPublicKey stay unused (the supervised child mints its own key and persists no hub-agent config)
       sessions: {
         provisionAgent: async () => {
           throw new Error("must not invoke provisionAgent");
         },
         persistHubPublicKey: async () => {
           throw new Error("must not invoke persistHubPublicKey");
         },
+        initRepo: async () => undefined,
       } as unknown as Parameters<
         typeof createSidecarDeployRouter
       >[0]["sessions"],
-      // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- only loadOrGenerateKey is exercised (the single-step branch registers the agent's signing key on the transport before spawn)
+      // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- the single-step branch registers the agent's signing key (loadOrGenerateKey) and records the hub key (recordHubKey) at the head before spawn
       keyStore: {
-        recordHubKey: () => {
-          throw new Error("must not invoke recordHubKey");
-        },
+        recordHubKey: () => undefined,
         loadOrGenerateKey: async () => ({
           keyPair: await generateKeyPair(),
           isNew: false,
         }),
       } as unknown as Parameters<
         typeof createSidecarDeployRouter
       >[0]["keyStore"],
-      onAgentEvent: () => () => undefined,
       transport,
       repoStore,
       signingKeySeed: keyPair.privateKey,
       createAgentCrypto: createEd25519Crypto,
+      assertSourceBuildable: () => undefined,
       registerDeployment: () => undefined,
       unregisterDeployment: () => undefined,
       multistepSubprocessSpawner: spawner,
@@ -271,13 +276,15 @@ describe("agent signing-key registration lifecycle on the host transport", () =>
           steps: { "step-1": { kind: "step" } },
         },
         sources: {
-          "step-1": {
-            id: "step-1",
-            provider: "anthropic",
-            baseURL: "https://api.anthropic.com",
-            apiKey: "sk-step-1",
-            model: "claude-3-5",
-          },
+          "step-1": [
+            {
+              id: "step-1",
+              provider: "anthropic",
+              baseURL: "https://api.anthropic.com",
+              apiKey: "sk-step-1",
+              model: "claude-3-5",
+            },
+          ],
         },
       },
     };
@@ -316,6 +323,26 @@ describe("agent signing-key registration lifecycle on the host transport", () =>
     // (b) Registered after deploy -- the supervisor can now sign agent mail.
     expect(isRegistered(transport)).toBe(true);
 
+    // (b') The deploy persisted a schema-valid restore record for the
+    // deployment, carrying the head address so a boot-time restore can
+    // re-establish it.
+    const deploymentId = deriveDeploymentId(AGENT_ADDRESS);
+    const recordFile = path.join(
+      dataDir,
+      "workflow-runs",
+      deploymentId,
+      "deployment.json",
+    );
+    const parsedRecord = WorkflowDeploymentRecord(
+      JSON.parse(await fs.readFile(recordFile, "utf8")),
+    );
+    if (parsedRecord instanceof type.errors) {
+      throw new Error(
+        `deployment record failed validation: ${parsedRecord.summary}`,
+      );
+    }
+    expect(parsedRecord.agentAddress).toBe(AGENT_ADDRESS);
+
     const undeploy = router.undeploy;
     if (undeploy === undefined) throw new Error("router.undeploy undefined");
     await undeploy({
@@ -327,6 +354,15 @@ describe("agent signing-key registration lifecycle on the host transport", () =>
     // (c) Unregistered after undeploy -- no leaked key for a torn-down agent.
     expect(isRegistered(transport)).toBe(false);
 
+    // (c') Undeploy dropped the restore record so a boot-time restore will
+    // not re-spawn the torn-down deployment.
+    expect(
+      await fs.access(recordFile).then(
+        () => true,
+        () => false,
+      ),
+    ).toBe(false);
+
     await fs.rm(tempBase, { recursive: true, force: true });
     await fs.rm(dataDir, { recursive: true, force: true });
   });

apps/sidecar/src/atomic-write.test.ts

@@ -0,0 +1,102 @@
+import { describe, test, expect } from "bun:test";
+import fs from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+
+import { writeFileAtomicDurable } from "./atomic-write";
+
+async function makeDir(): Promise<string> {
+  return fs.mkdtemp(path.join(os.tmpdir(), "atomic-write-"));
+}
+
+async function listNames(dir: string): Promise<string[]> {
+  return (await fs.readdir(dir)).sort();
+}
+
+describe("writeFileAtomicDurable", () => {
+  test("round-trips contents to the target path", async () => {
+    const dir = await makeDir();
+    const file = path.join(dir, "record.json");
+
+    await writeFileAtomicDurable(file, '{"a":1}', { mode: 0o600 });
+    expect(await fs.readFile(file, "utf8")).toBe('{"a":1}');
+
+    await fs.rm(dir, { recursive: true, force: true });
+  });
+
+  test("applies mode on the created file", async () => {
+    const dir = await makeDir();
+    const file = path.join(dir, "record.json");
+
+    await writeFileAtomicDurable(file, "x", { mode: 0o600 });
+    const stat = await fs.stat(file);
+    expect(stat.mode & 0o777).toBe(0o600);
+
+    await fs.rm(dir, { recursive: true, force: true });
+  });
+
+  test("re-applies mode when overwriting an existing file", async () => {
+    const dir = await makeDir();
+    const file = path.join(dir, "record.json");
+
+    // Seed a pre-existing file with a laxer mode. A plain in-place
+    // overwrite keeps the original mode; the temp+rename path creates a
+    // fresh file every write, so the requested mode actually lands.
+    await fs.writeFile(file, "old", { mode: 0o644 });
+    await writeFileAtomicDurable(file, "new", { mode: 0o600 });
+
+    expect(await fs.readFile(file, "utf8")).toBe("new");
+    const stat = await fs.stat(file);
+    expect(stat.mode & 0o777).toBe(0o600);
+
+    await fs.rm(dir, { recursive: true, force: true });
+  });
+
+  test("leaves no temp orphan after a successful write", async () => {
+    const dir = await makeDir();
+    const file = path.join(dir, "record.json");
+
+    await writeFileAtomicDurable(file, "x", { mode: 0o600 });
+    expect(await listNames(dir)).toEqual(["record.json"]);
+
+    await fs.rm(dir, { recursive: true, force: true });
+  });
+
+  test("a write that fails before staging leaves the prior file intact", async () => {
+    const dir = await makeDir();
+    const file = path.join(dir, "record.json");
+    await fs.writeFile(file, "old", { mode: 0o600 });
+
+    // Deny writes to the directory so the temp-file creation fails
+    // outright. The prior complete record must survive untouched -- an
+    // interrupted rotation must never corrupt the sole restore source.
+    await fs.chmod(dir, 0o500);
+    try {
+      await expect(
+        writeFileAtomicDurable(file, "new", { mode: 0o600 }),
+      ).rejects.toThrow();
+      expect(await fs.readFile(file, "utf8")).toBe("old");
+      expect(await listNames(dir)).toEqual(["record.json"]);
+    } finally {
+      await fs.chmod(dir, 0o700);
+      await fs.rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("a write that fails at the rename unlinks its staged temp", async () => {
+    const dir = await makeDir();
+    const file = path.join(dir, "record.json");
+
+    // Occupy the target path with a directory so the rename fails after
+    // the temp file has already been staged and fsynced. This exercises
+    // the error-path unlink: the staged temp must be removed rather than
+    // stranded as an orphan.
+    await fs.mkdir(file);
+    await expect(
+      writeFileAtomicDurable(file, "new", { mode: 0o600 }),
+    ).rejects.toThrow();
+    expect(await listNames(dir)).toEqual(["record.json"]);
+
+    await fs.rm(dir, { recursive: true, force: true });
+  });
+});

apps/sidecar/src/atomic-write.ts

@@ -0,0 +1,79 @@
+// Atomic, durable file replacement for the sidecar's non-rebuildable
+// on-disk records. Distinct from the cache's rebuildable temp+rename
+// (no fsync, a lost write just forces a re-fetch) and from
+// `fsyncWriteFile`'s in-place fsync write (no atomicity, a torn write
+// leaves a half-file): this is the tier for a sole restore source that
+// must survive both a process kill and a power loss without ever
+// exposing a torn record.
+
+import { open, rename, unlink } from "node:fs/promises";
+import { dirname } from "node:path";
+
+import { getLogger } from "@intx/log";
+import { hexEncode } from "@intx/types";
+
+const logger = getLogger(["interchange", "sidecar", "atomic-write"]);
+
+export interface AtomicWriteOptions {
+  /** Permission mode applied when the temp file is created. */
+  mode: number;
+}
+
+/**
+ * Replace `path` with `contents` atomically and durably. The bytes land
+ * in a fresh per-write temp file that is fsynced and then `rename`d over
+ * `path`; because rename is atomic within a directory, a reader only
+ * ever observes the prior complete file or the new complete file, never
+ * a torn one. The fsync before the rename is what extends that
+ * guarantee past process death to OS crash / power loss: without it, the
+ * ext4 delayed-allocation window can surface the renamed path as a
+ * zero-length file after a power loss.
+ *
+ * The parent directory is fsynced after the rename so the new link is
+ * itself durable, but a filesystem that rejects directory fsync
+ * (FAT/exFAT, some network mounts) only degrades durability -- the file
+ * is already renamed and fsynced -- so that failure is logged, not
+ * thrown.
+ *
+ * `mode` is applied on the temp file's creation, so it takes effect on
+ * every write. A plain in-place overwrite of an existing file would
+ * silently keep the original file's mode instead.
+ *
+ * The temp file follows `createTarballCache`'s `.tmp.<pid>.<rand>`
+ * naming convention for consistency across the sidecar's staged writes.
+ */
+export async function writeFileAtomicDurable(
+  path: string,
+  contents: string,
+  options: AtomicWriteOptions,
+): Promise<void> {
+  const tmp = `${path}.tmp.${String(process.pid)}.${hexEncode(crypto.getRandomValues(new Uint8Array(8)))}`;
+  try {
+    const handle = await open(tmp, "w", options.mode);
+    try {
+      await handle.writeFile(contents);
+      await handle.sync();
+    } finally {
+      await handle.close();
+    }
+    await rename(tmp, path);
+  } catch (cause) {
+    // The write failed and is about to rethrow; unlink the temp so a
+    // failed write leaves no orphan. Best-effort: the temp may never
+    // have been created, and a second failure here must not mask the
+    // original cause.
+    await unlink(tmp).catch(() => undefined);
+    throw cause;
+  }
+
+  try {
+    const dirHandle = await open(dirname(path), "r");
+    try {
+      await dirHandle.sync();
+    } finally {
+      await dirHandle.close();
+    }
+  } catch (err) {
+    logger.warn`parent-dir fsync failed for ${path}; durability is degraded but the file is renamed and fsynced — ${err instanceof Error ? err.message : String(err)}`;
+  }
+}

apps/sidecar/src/config.ts

@@ -9,7 +9,6 @@
 // re-reading env, so the boundary stays at the boot edge.
 
 import { AdapterManifest } from "@intx/inference";
-import type { GCPolicy, RetentionPolicy } from "@intx/storage-isogit";
 
 const DEFAULT_CACHE_MAX_BYTES = 10 * 1024 * 1024 * 1024;
 
@@ -69,55 +68,3 @@ export function readAdapterManifest(): AdapterManifest {
   }
   return AdapterManifest.assert(parsed);
 }
-
-// Write-path GC for the sidecar's deployed-agent repos. The reactor
-// commits loose objects every cycle, so these repos accumulate loose
-// objects far faster than packs; the loose threshold is the dominant
-// trigger here. Retention defaults to tip-only: the sidecar treats the
-// repo as the agent's current state, not a long-term archive, so dropping
-// commit history keeps the repo small for reactor read/commit latency. An
-// operator that needs the history preserved sets
-// SIDECAR_AGENT_GC_RETENTION=keep-history.
-const DEFAULT_SIDECAR_AGENT_GC_PACK_THRESHOLD = 16;
-const DEFAULT_SIDECAR_AGENT_GC_LOOSE_THRESHOLD = 512;
-const DEFAULT_SIDECAR_AGENT_GC_WARN_BYTES = 128 * 1024 * 1024;
-
-function readPositiveIntEnv(name: string, fallback: number): number {
-  const raw = process.env[name];
-  if (raw === undefined || raw.trim() === "") return fallback;
-  const n = Number(raw);
-  if (!Number.isInteger(n) || n <= 0) {
-    throw new Error(`${name} must be a positive integer; got ${raw}`);
-  }
-  return n;
-}
-
-function readRetentionEnv(
-  name: string,
-  fallback: RetentionPolicy,
-): RetentionPolicy {
-  const raw = process.env[name];
-  if (raw === undefined || raw.trim() === "") return fallback;
-  if (raw === "tip-only" || raw === "keep-history") return raw;
-  throw new Error(
-    `${name} must be "tip-only" or "keep-history"; got ${JSON.stringify(raw)}`,
-  );
-}
-
-export function readAgentGCPolicy(): GCPolicy {
-  return {
-    packThreshold: readPositiveIntEnv(
-      "SIDECAR_AGENT_GC_PACK_THRESHOLD",
-      DEFAULT_SIDECAR_AGENT_GC_PACK_THRESHOLD,
-    ),
-    looseThreshold: readPositiveIntEnv(
-      "SIDECAR_AGENT_GC_LOOSE_THRESHOLD",
-      DEFAULT_SIDECAR_AGENT_GC_LOOSE_THRESHOLD,
-    ),
-    warnBytes: readPositiveIntEnv(
-      "SIDECAR_AGENT_GC_WARN_BYTES",
-      DEFAULT_SIDECAR_AGENT_GC_WARN_BYTES,
-    ),
-    retention: readRetentionEnv("SIDECAR_AGENT_GC_RETENTION", "tip-only"),
-  };
-}