fix(eve): enforce schema-driven legacy inbox encoding (#2840)

Co-authored-by: vercel-gh-bot-5[bot] <312521305+vercel-gh-bot-5[bot]@users.noreply.github.com>
Co-authored-by: Rui <ruiconti@gmail.com>

.changeset/honest-wire-migrations.md

@@ -0,0 +1,5 @@
+---
+"eve": patch
+---
+
+Fix deliveries to persistent subagent inboxes by projecting current caller metadata through the destination wire schema. Versioned migrations are now pure, immutable data transforms enforced by the wire guard.

packages/eve/src/execution/wire/session-inbox-encoder.test.ts

@@ -0,0 +1,64 @@
+import { describe, expect, it } from "vitest";
+
+import {
+  SessionInboxWireError,
+  type SessionInboxWireTarget,
+} from "#execution/wire/session-inbox-contract.js";
+import { sessionInboxWire } from "#execution/wire/session-inbox-encoder.js";
+
+const legacyTargets = [
+  { variant: "deliver", version: 0 },
+  { variant: "send", version: 0 },
+  { version: 1 },
+] satisfies readonly SessionInboxWireTarget[];
+const activityObserver = {
+  sink: { url: "https://example.com/activity", version: 1 as const },
+};
+
+describe("session inbox encoder", () => {
+  it.each(
+    legacyTargets.flatMap(
+      (target) =>
+        [
+          [target, undefined],
+          [target, activityObserver],
+        ] as const,
+    ),
+  )("projects caller fields for target %o with activity observer %o", (target, observer) => {
+    const caller = {
+      activityObserver: observer,
+      callId: "call-1",
+      futureCallerField: "future-value",
+      replyTo: { kind: "hook" as const, token: "callback-token" },
+      subagentName: "researcher",
+    };
+
+    const wire = sessionInboxWire.encode(
+      { caller, kind: "send", payload: { message: "legacy" } },
+      target,
+    );
+
+    expect(wire).toHaveProperty("caller", {
+      callId: "call-1",
+      replyTo: { kind: "hook", token: "callback-token" },
+      subagentName: "researcher",
+    });
+  });
+
+  it.each(legacyTargets)("wraps malformed caller fields for target %o", (target) => {
+    expect(() =>
+      sessionInboxWire.encode(
+        {
+          caller: {
+            callId: 1 as never,
+            replyTo: { kind: "hook", token: "callback-token" },
+            subagentName: "researcher",
+          },
+          kind: "send",
+          payload: { message: "legacy" },
+        },
+        target,
+      ),
+    ).toThrowError(SessionInboxWireError);
+  });
+});

packages/eve/src/execution/wire/session-inbox-encoder.ts

@@ -28,7 +28,7 @@ type LegacySessionInboxWireTarget = Extract<SessionInboxWireTarget, { readonly v
 type VersionedSessionInboxEncoder = (command: SessionInboxCommand) => unknown;
 
 const versionedEncoders = {
-  1: (command: SessionInboxCommand) => encodeSessionCommandV1(withoutActivityObserver(command)),
+  1: encodeSessionCommandV1,
   2: encodeSessionCommandV2,
 } satisfies Record<SessionInboxWireVersion, VersionedSessionInboxEncoder>;
 
@@ -52,10 +52,7 @@ function encode(
   target: SessionInboxWireTarget,
 ): SessionInboxWireV1 | SessionInboxWireV2 | Record<string, unknown> {
   if (target.version === 0) {
-    return encodeSessionCommandV0(
-      encodeSessionCommandV1(withoutActivityObserver(command)),
-      target.variant,
-    );
+    return encodeSessionCommandV0(encodeSessionCommandV1(command), target.variant);
   }
   if (isSessionInboxWireVersion(target.version)) {
     return versionedEncoders[target.version](command) as SessionInboxWireV1 | SessionInboxWireV2;
@@ -65,11 +62,5 @@ function encode(
   );
 }
 
-function withoutActivityObserver(command: SessionInboxCommand): SessionInboxCommand {
-  if (!("caller" in command) || command.caller?.activityObserver === undefined) return command;
-  const { activityObserver: _activityObserver, ...caller } = command.caller;
-  return { ...command, caller };
-}
-
 /** Server/step-safe producer facade. */
 export const sessionInboxWire = { encode } as const;

packages/eve/src/execution/wire/session-inbox-wire.test.ts

@@ -0,0 +1,30 @@
+import { runInNewContext } from "node:vm";
+
+import { describe, expect, it } from "vitest";
+
+import { sessionInboxWire, SessionInboxWireError } from "#execution/wire/session-inbox-wire.js";
+
+describe("session inbox wire policy", () => {
+  it("rejects a present non-numeric version before normalization", () => {
+    expect(() =>
+      sessionInboxWire.decode({ kind: "deliver", payloads: [], version: undefined }),
+    ).toThrowError(SessionInboxWireError);
+  });
+
+  it("normalizes cross-realm records before running pure migrations", () => {
+    const wire = runInNewContext(`({
+      caller: undefined,
+      kind: "deliver",
+      payloads: [{ message: "legacy", omitted: undefined }],
+      version: 1,
+    })`);
+
+    expect(sessionInboxWire.decode(wire)).toEqual({
+      auth: undefined,
+      caller: undefined,
+      kind: "deliver",
+      payloads: [{ message: "legacy" }],
+      requestId: undefined,
+    });
+  });
+});

packages/eve/src/execution/wire/session-inbox-wire.ts

@@ -14,10 +14,8 @@ import {
 } from "#execution/wire/session-inbox-contract.js";
 import type { SessionInboxWire } from "#execution/wire/session-inbox-encoder.js";
 import { sessionInboxWireV0Migration } from "#execution/wire/session-inbox-wire.v0.js";
-import {
-  normalizeSessionInboxWireV2,
-  sessionInboxWireV1Migration,
-} from "#execution/wire/session-inbox-wire.v2-migration.js";
+import { normalizeSessionInboxWireV2 } from "#execution/wire/session-inbox-wire.v2-migration.js";
+import { sessionInboxWireV1Migration } from "#execution/wire/session-inbox-wire.v2.migration.js";
 
 /**
  * The session inbox wire family: every payload persisted to a session's
@@ -59,6 +57,11 @@ function decode(value: unknown): DecodedSessionInbox {
     typeof value === "object" && value !== null && "version" in value
       ? (value as { readonly version?: unknown }).version
       : undefined;
+  const hasDeclaredVersion = typeof value === "object" && value !== null && "version" in value;
+  if (hasDeclaredVersion && typeof declaredVersion !== "number") {
+    throw new SessionInboxWireError(`${WIRE_LABEL}: value has no numeric "version" field.`);
+  }
+  const normalized = normalizeSessionInboxWireV2(value);
   let migrated: unknown;
   try {
     migrated = runMigrationChain({
@@ -66,7 +69,7 @@ function decode(value: unknown): DecodedSessionInbox {
       label: WIRE_LABEL,
       migrations: sessionInboxMigrations,
       targetVersion: SESSION_INBOX_WIRE_VERSION,
-      value,
+      value: normalized,
     });
   } catch (error) {
     throw new SessionInboxWireError(error instanceof Error ? error.message : String(error));

packages/eve/src/execution/wire/session-inbox-wire.v1.ts

@@ -5,6 +5,7 @@ import type {
   DeliverHookPayload,
   SessionCommand,
   SessionTimeoutHookPayload,
+  TurnCaller,
 } from "#channel/types.js";
 import { coalesceDeliverPayloads } from "#execution/deliver-payloads.js";
 import { SessionInboxWireError } from "#execution/wire/session-inbox-contract.js";
@@ -198,6 +199,7 @@ const callerSchema = z
     taskId: z.string().optional(),
   })
   .strict();
+const callerProjectionSchema = callerSchema.strip();
 const traceContextSchema = z
   .object({ spanId: z.string(), traceFlags: z.number(), traceId: z.string() })
   .strict();
@@ -250,28 +252,29 @@ export type SessionInboxWireV1 = z.infer<typeof sessionInboxWireV1Schema>;
 export function encodeSessionCommandV1(
   command: DeliverHookPayload | SessionCommand | SessionTimeoutHookPayload,
 ): SessionInboxWireV1 {
+  const input = toV1Command(command);
   const wire =
-    command.kind === "send"
+    input.kind === "send"
       ? {
-          auth: command.auth,
-          caller: command.caller,
+          auth: input.auth,
+          caller: input.caller,
           deliveryMetadata:
-            command.delivery === undefined ? undefined : [{ ...command.delivery, payloadIndex: 0 }],
+            input.delivery === undefined ? undefined : [{ ...input.delivery, payloadIndex: 0 }],
           kind: "deliver" as const,
-          payload: command.payload,
-          payloads: [command.payload],
-          requestId: command.requestId,
-          taskDeliveryId: command.taskDeliveryId,
-          turnPolicy: command.turnPolicy,
+          payload: input.payload,
+          payloads: [input.payload],
+          requestId: input.requestId,
+          taskDeliveryId: input.taskDeliveryId,
+          turnPolicy: input.turnPolicy,
           version: VERSION,
         }
-      : command.kind === "deliver"
+      : input.kind === "deliver"
         ? {
-            ...command,
-            payload: coalesceDeliverPayloads(command.payloads),
+            ...input,
+            payload: coalesceDeliverPayloads(input.payloads),
             version: VERSION,
           }
-        : { ...command, version: VERSION };
+        : { ...input, version: VERSION };
   const parsed = sessionInboxWireV1Schema.safeParse(wire);
   if (!parsed.success) {
     throw new SessionInboxWireError(
@@ -280,3 +283,20 @@ export function encodeSessionCommandV1(
   }
   return parsed.data;
 }
+
+function toV1Command(
+  command: DeliverHookPayload | SessionCommand | SessionTimeoutHookPayload,
+): DeliverHookPayload | SessionCommand | SessionTimeoutHookPayload {
+  if (!("caller" in command) || command.caller === undefined) return command;
+  return { ...command, caller: toV1Caller(command.caller) };
+}
+
+function toV1Caller(caller: TurnCaller) {
+  const parsed = callerProjectionSchema.safeParse(caller);
+  if (!parsed.success) {
+    throw new SessionInboxWireError(
+      `Produced a session inbox payload that does not match wire version ${VERSION}: ${formatValidationError(parsed.error)}`,
+    );
+  }
+  return parsed.data;
+}

packages/eve/src/execution/wire/session-inbox-wire.v2-migration.ts

@@ -1,19 +1,5 @@
-import type { VersionMigration } from "#execution/durable-session-migrations/chain.js";
 import { isObject } from "#shared/guards.js";
 
-export const sessionInboxWireV1Migration: VersionMigration = {
-  from: 1,
-  migrate(prior) {
-    const normalized = normalizeSessionInboxWireV2(prior) as Record<string, unknown>;
-    return {
-      ...normalized,
-      ...(normalized.kind === "deliver" && !("payload" in normalized) ? { payload: {} } : {}),
-      version: 2,
-    };
-  },
-  to: 2,
-};
-
 /** Converts Workflow-VM records into this realm before wire consumption. */
 export function normalizeSessionInboxWireV2(value: unknown, arrayFallback = false): unknown {
   if (value === undefined) return arrayFallback ? null : undefined;

packages/eve/src/execution/wire/session-inbox-wire.v2.migration.test.ts

@@ -0,0 +1,43 @@
+import { describe, expect, it } from "vitest";
+
+import { sessionInboxWireV1Migration } from "#execution/wire/session-inbox-wire.v2.migration.js";
+
+describe("session inbox wire v2 migration", () => {
+  it("stamps controls with version 2", () => {
+    expect(sessionInboxWireV1Migration.migrate({ kind: "clear", version: 1 })).toEqual({
+      kind: "clear",
+      version: 2,
+    });
+  });
+
+  it("adds the required payload mirror to v1 deliveries", () => {
+    expect(
+      sessionInboxWireV1Migration.migrate({
+        kind: "deliver",
+        payloads: [{ message: "legacy" }],
+        version: 1,
+      }),
+    ).toEqual({
+      kind: "deliver",
+      payload: {},
+      payloads: [{ message: "legacy" }],
+      version: 2,
+    });
+  });
+
+  it("preserves an existing payload mirror", () => {
+    expect(
+      sessionInboxWireV1Migration.migrate({
+        kind: "deliver",
+        payload: { message: "legacy" },
+        payloads: [{ message: "legacy" }],
+        version: 1,
+      }),
+    ).toEqual({
+      kind: "deliver",
+      payload: { message: "legacy" },
+      payloads: [{ message: "legacy" }],
+      version: 2,
+    });
+  });
+});

packages/eve/src/execution/wire/session-inbox-wire.v2.migration.ts

@@ -0,0 +1,16 @@
+import type { VersionMigration } from "#execution/durable-session-migrations/chain.js";
+
+/** Pure shape migration from the immutable v1 wire contract to v2. */
+export const sessionInboxWireV1Migration: VersionMigration = {
+  from: 1,
+  migrate(prior) {
+    const value = prior as Record<string, unknown>;
+    const migrated: Record<string, unknown> & { readonly version: 2 } = {
+      ...value,
+      version: 2,
+    };
+    if (value.kind === "deliver" && !("payload" in value)) migrated.payload = {};
+    return migrated;
+  },
+  to: 2,
+};

research/session-inbox-wire-schema.md

@@ -1,7 +1,7 @@
 ---
 issue: https://github.com/vercel/eve/issues/1765
 status: proposed
-last_updated: "2026-08-18"
+last_updated: "2026-08-31"
 ---
 
 # Versioned wire schema for the session inbox
@@ -79,6 +79,12 @@ Normative rules:
   `VersionMigration`, freeze the new shape. Historic versions live on as
   executable migrations plus frozen payload fixtures (the turn-workflow
   precedent), not as retained schemas.
+- **Protocol data and migration policy stay separate.** A `*.vN.ts` module
+  owns the immutable schema and version-bound encoder. A
+  `*.vN.migration.ts` module is a pure data transform with no normalization or
+  version-selection dependencies. The encoder and decoder facades own mutable
+  policy: selecting a target, assembling the chain, and normalizing values
+  received from another Workflow VM realm.
 - **The complete transported value is validated once, at encode.** The
   schema owns the envelope and every eve-owned `DeliverPayload` field,
   composing the existing strict `inputResponseSchema` and
@@ -194,6 +200,10 @@ mechanical guard in the existing CI lint job (`pnpm guard:invariants`):
   TypeScript requires an encoder for every registered stamped version. The
   required unit tier then encodes and decodes every registry entry, while each
   version's frozen contract pins its exact shape and backwards migration.
+- Rule 40 also freezes pure `*.vN.migration.ts` modules with their tests and
+  rejects policy imports from those transforms. A one-time historical rewrite
+  is represented by exact old/new Git blob hashes, so the exception expires as
+  soon as the approved rewrite reaches `main`.
 - Exact current-version bytes stay in the unit contract, where the encoded
   object can be asserted without decoding workflow-owned serde. The
   deterministic registry checks cover future stamped-version changes. The

scripts/guard-invariants.mjs

@@ -100,14 +100,15 @@
  *             `pnpm --filter eve build`. Turbo owns workspace dependency
  *             ordering; nested builds race on eve's clean-and-publish dist
  *             directory and let consumers observe a partial package.
- *   rule 40 — Every shipped wire-version module
- *             (`src/execution/wire/*-wire.vN.ts`) must carry a colocated
- *             `*-wire.vN.test.ts`. Version modules, tests, and snapshots already
- *             present on main are immutable. The session-inbox registry must
- *             also be contiguous, name every module, and identify its highest
- *             version as current. Wire versions are append-only protocol
- *             history: change the contract by adding a version and migration,
- *             never by updating a historical schema and its snapshot together.
+ *   rule 40 — Wire schemas and version-bound encoders are immutable protocol
+ *             data. Pure `*.vN.migration.ts` transforms are immutable data too;
+ *             version selection, chain assembly, and realm normalization remain
+ *             editable policy. Every data module must carry a colocated test.
+ *             The session-inbox registry must be contiguous, name every schema
+ *             module, and identify its highest version as current. Wire versions
+ *             are append-only protocol history: change the contract by adding a
+ *             version and migration, never by updating historical data and its
+ *             snapshot together.
  *
  * Baselines for rules with pre-existing violations live in
  * `guard-invariants-baseline.json`. Counts and allowlists in that file
@@ -536,8 +537,21 @@ function importSpecifier(node) {
 const WIRE_FAMILY_DIR = "packages/eve/src/execution/wire";
 const SESSION_INBOX_WIRE_CONTRACT = `${WIRE_FAMILY_DIR}/session-inbox-contract.ts`;
 const VERSIONED_WIRE_HISTORY_RE = new RegExp(
-  `^${WIRE_FAMILY_DIR}/(?:__snapshots__/)?[a-z0-9-]+-wire\\.v\\d+(?:\\.test\\.ts(?:\\.snap)?|\\.ts)$`,
+  `^${WIRE_FAMILY_DIR}/(?:__snapshots__/)?[a-z0-9-]+-wire\\.v\\d+(?:\\.migration)?(?:\\.test\\.ts(?:\\.snap)?|\\.ts)$`,
 );
+const RULE_40_ALLOWED_REWRITES = new Map([
+  [
+    `${WIRE_FAMILY_DIR}/session-inbox-wire.v1.ts`,
+    {
+      from: "5f110be5d7b488216c574a1aef9d2074d670efd2",
+      to: "7f5864f20e6bbb9f430c23918320ca6319c4cb14",
+    },
+  ],
+]);
+const PURE_MIGRATION_IMPORTS = new Map([
+  ["#execution/durable-session-migrations/chain.js", new Map([["VersionMigration", "type"]])],
+  ["#shared/guards.js", new Map([["isObject", "value"]])],
+]);
 
 function gitOutput(args) {
   try {
@@ -554,25 +568,32 @@ function gitOutput(args) {
 function checkRule40ImmutableWireHistory() {
   const hasBase = gitOutput(["rev-parse", "--verify", "origin/main"]) !== undefined;
   const comparisons = [
-    ["diff", "--name-status", "--", WIRE_FAMILY_DIR],
-    ["diff", "--cached", "--name-status", "--", WIRE_FAMILY_DIR],
+    { args: ["diff", "--name-status", "--", WIRE_FAMILY_DIR], state: "worktree" },
+    {
+      args: ["diff", "--cached", "--name-status", "--", WIRE_FAMILY_DIR],
+      state: "index",
+    },
   ];
   if (hasBase)
-    comparisons.push(["diff", "--name-status", "origin/main...HEAD", "--", WIRE_FAMILY_DIR]);
+    comparisons.push({
+      args: ["diff", "--name-status", "origin/main...HEAD", "--", WIRE_FAMILY_DIR],
+      state: "head",
+    });
 
   const changes = new Set();
-  for (const args of comparisons) {
+  for (const { args, state } of comparisons) {
     for (const line of (gitOutput(args) ?? "").trim().split("\n")) {
-      if (line !== "") changes.add(line);
+      if (line !== "") changes.add(`${state}\t${line}`);
     }
   }
 
   /** @type {Violation[]} */
   const violations = [];
   for (const change of changes) {
-    const [status, ...paths] = change.split("\t");
+    const [state, status, ...paths] = change.split("\t");
     const protectedPaths = paths.filter((path) => VERSIONED_WIRE_HISTORY_RE.test(path));
     if (protectedPaths.length === 0 || status === "A") continue;
+    if (protectedPaths.every((path) => isAllowedRule40Rewrite(path, state))) continue;
     if (
       hasBase &&
       protectedPaths.every(
@@ -591,6 +612,48 @@ function checkRule40ImmutableWireHistory() {
   return violations;
 }
 
+function isAllowedRule40Rewrite(path, state) {
+  const rewrite = RULE_40_ALLOWED_REWRITES.get(path);
+  if (rewrite === undefined) return false;
+  const baseHash = gitOutput(["rev-parse", `origin/main:${path}`])?.trim();
+  const currentHash =
+    state === "head"
+      ? gitOutput(["rev-parse", `HEAD:${path}`])?.trim()
+      : state === "index"
+        ? gitOutput(["rev-parse", `:${path}`])?.trim()
+        : gitOutput(["hash-object", path])?.trim();
+  return baseHash === rewrite.from && currentHash === rewrite.to;
+}
+
+function checkRule40MigrationPurity(path, source) {
+  const sourceFile = ts.createSourceFile(
+    path,
+    source,
+    ts.ScriptTarget.Latest,
+    true,
+    ts.ScriptKind.TS,
+  );
+  /** @type {Violation[]} */
+  const violations = [];
+  const visit = (node) => {
+    const specifier = importSpecifier(node);
+    if (specifier !== undefined) {
+      const allowed = PURE_MIGRATION_IMPORTS.get(specifier.text);
+      if (allowed === undefined || !hasOnlyAllowedNamedImports(node, allowed)) {
+        violations.push({
+          rule: 40,
+          file: path,
+          line: sourceFile.getLineAndCharacterOfPosition(specifier.getStart(sourceFile)).line + 1,
+          message: `imports "${specifier.text}". Versioned wire migrations are immutable data transforms, so they may import only the VersionMigration type or dependency-free shared guards. Move normalization and version-selection policy to the wire facade.`,
+        });
+      }
+    }
+    ts.forEachChild(node, visit);
+  };
+  visit(sourceFile);
+  return violations;
+}
+
 async function checkRule40WireContracts() {
   const violations = checkRule40ImmutableWireHistory();
   let entries;
@@ -601,11 +664,11 @@ async function checkRule40WireContracts() {
   }
 
   for (const name of entries) {
-    const match = name.match(/^([a-z0-9-]+)-wire\.v(\d+)\.ts$/);
+    const match = name.match(/^([a-z0-9-]+)-wire\.v(\d+)(\.migration)?\.ts$/);
     if (match === null) continue;
-    const [, family, version] = match;
+    const [, family, version, kind = ""] = match;
 
-    const testName = `${family}-wire.v${version}.test.ts`;
+    const testName = `${family}-wire.v${version}${kind}.test.ts`;
     if (!entries.includes(testName)) {
       violations.push({
         rule: 40,
@@ -614,6 +677,12 @@ async function checkRule40WireContracts() {
         message: `wire family "${family}" version ${version} has no colocated contract test (${testName}). Pin this version's schema/encoder or migration/fixtures before shipping it.`,
       });
     }
+    if (kind === ".migration") {
+      const path = `${WIRE_FAMILY_DIR}/${name}`;
+      violations.push(
+        ...checkRule40MigrationPurity(path, await readFile(join(REPO_ROOT, path), "utf8")),
+      );
+    }
   }
 
   const contractSource = await readFile(join(REPO_ROOT, SESSION_INBOX_WIRE_CONTRACT), "utf8");