feat: add dormant Boss adapter contracts

.gitignore

@@ -4,5 +4,4 @@ dist/*
 !dist/build-info.json
 *.log
 .DS_Store
-package-lock.json
 progress.md

README.md

@@ -56,6 +56,23 @@ appear in the already-open terminal instead of existing only in the saved
 transcript. This refresh happens after the turn is idle so Codex does not reopen
 in a phantom `Working` state. Retryable app-server stream errors are reported as reconnecting without terminating the sidecar, allowing Codex's own retry to complete. Orchestrated `fresh: true` launches remove the saved `coi` thread state before registration.
 
+The additive Stage-B `boss-run-v1` adapter contracts are present but dormant.
+Ordinary local and remote-access communication continues to use the existing
+protocol-v3 behavior when Boss metadata is omitted. The legacy broker does not
+advertise or bind Boss participants until a protected provider supplies all of
+the required broker identity, credential-registry, authority-transition, and
+participant-health predicates. Boss-scoped discovery and routing fail closed
+across ordinary sessions and other Boss runs. The `boss_participant` and
+`boss_reviewer` launch-profile markers make production Codex launches fail
+with `PROVIDER_AUTHORITY_UNAVAILABLE`: this adapter has no broker-owned,
+artifact-attested provider executable and therefore never resolves protected
+launches through caller `PATH`. Their non-spawning parser validators still
+reject disabled approvals and `danger-full-access`, and reviewers remain
+read-only. The markers do not install or advertise a restricted operation client.
+Restricted Boss operation clients are not advertised or exported yet. They
+remain unavailable until the protected broker can provision both the binding
+and transport through a non-caller-forgeable factory.
+
 ## Install
 
 For normal use, install the package so the command-line entry points are on

boss-control-outbox.test.ts

@@ -0,0 +1,74 @@
+import assert from "node:assert/strict";
+import { mkdtempSync, readdirSync, rmSync, writeFileSync, mkdirSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { createHash } from "node:crypto";
+import test from "node:test";
+import { BOSS_CONTROL_ENVELOPE_VERSION } from "@dataforxyz/agent-intercom-core/boss";
+import { PersistentBossControlOutbox } from "./boss-control-outbox.ts";
+
+const envelope = {
+  type: "boss.worker.health" as const,
+  version: BOSS_CONTROL_ENVELOPE_VERSION,
+  messageId: "message-1",
+  bossRunId: "run-1",
+  participantId: "worker-1",
+  bindingEpoch: 1,
+  idempotencyKey: "operation-1",
+  payload: { state: "working" },
+};
+
+test("stable idempotency with a new messageId keeps one canonical request and the new caller correlation", () => {
+  const dir = mkdtempSync(join(tmpdir(), "boss-control-outbox-"));
+  try {
+    const outbox = new PersistentBossControlOutbox("session-1", dir);
+    assert.equal(outbox.enqueue("target-session", envelope), "added");
+    assert.equal(outbox.enqueue("target-session", {
+      payload: { state: "working" },
+      idempotencyKey: "operation-1",
+      bindingEpoch: 1,
+      participantId: "worker-1",
+      bossRunId: "run-1",
+      messageId: "message-2",
+      version: BOSS_CONTROL_ENVELOPE_VERSION,
+      type: "boss.worker.health",
+    }), "existing");
+    assert.equal(outbox.list().length, 1);
+    assert.equal(outbox.list()[0]?.envelope.messageId, "message-2");
+    assert.throws(() => outbox.enqueue("other-session", envelope), /different canonical request/);
+  } finally {
+    rmSync(dir, { recursive: true, force: true });
+  }
+});
+
+test("result-order probes require durable ACK and exact deliveryId before outbox removal", () => {
+  const dir = mkdtempSync(join(tmpdir(), "boss-control-outbox-order-"));
+  try {
+    const outbox = new PersistentBossControlOutbox("session-1", dir);
+    outbox.enqueue("target-session", envelope);
+    assert.throws(() => outbox.removeCorrelated("operation-1", "message-1", "delivery-1"), /before the matching durable acknowledgement/);
+    assert.equal(outbox.markAccepted("operation-1", "message-1", "delivery-1"), "accepted");
+    assert.equal(new PersistentBossControlOutbox("session-1", dir).find("operation-1")?.state, "accepted");
+    assert.equal(outbox.markAccepted("operation-1", "message-1", "delivery-1"), "already-accepted");
+    assert.throws(() => outbox.removeCorrelated("operation-1", "message-1", "delivery-2"), /before the matching durable acknowledgement/);
+    assert.throws(() => outbox.removeCorrelated("operation-1", "message-1"), /omitted the durable deliveryId/);
+    outbox.removeCorrelated("operation-1", "message-1", "delivery-1");
+    assert.deepEqual(new PersistentBossControlOutbox("session-1", dir).list(), []);
+  } finally {
+    rmSync(dir, { recursive: true, force: true });
+  }
+});
+
+test("corrupt Boss outbox is quarantined and fails closed", () => {
+  const dir = mkdtempSync(join(tmpdir(), "boss-control-outbox-corrupt-"));
+  try {
+    const outboxDir = join(dir, "boss-control-outbox");
+    mkdirSync(outboxDir, { recursive: true });
+    const file = `${createHash("sha256").update("session-1").digest("hex")}.json`;
+    writeFileSync(join(outboxDir, file), "{not-json");
+    assert.throws(() => new PersistentBossControlOutbox("session-1", dir), /corrupt and quarantined/);
+    assert.match(readdirSync(outboxDir)[0] ?? "", new RegExp(`^${file}\\.corrupt-`));
+  } finally {
+    rmSync(dir, { recursive: true, force: true });
+  }
+});

boss-control-outbox.ts

@@ -0,0 +1,195 @@
+import { createHash } from "node:crypto";
+import { chmodSync, existsSync, mkdirSync, readFileSync, renameSync } from "node:fs";
+import { join } from "node:path";
+import { canonicalHash } from "@dataforxyz/agent-intercom-core/canonical";
+import { parseBossControlEnvelope, type BossControlEnvelope } from "@dataforxyz/agent-intercom-core/boss";
+import { assertBossCanonicalData } from "./broker/boss-adapter.ts";
+import { ensureIntercomRuntimeDir, getIntercomDirPath, INTERCOM_DIR_MODE, restrictIntercomRuntimeFile } from "./broker/paths.ts";
+import { writeDurableJson } from "./durable-json.ts";
+
+const BOSS_CONTROL_OUTBOX_VERSION = 2;
+const MAX_BOSS_CONTROL_OUTBOX_ENTRIES = 256;
+
+export interface StoredBossControl {
+  to: string;
+  envelope: BossControlEnvelope;
+  scope: string;
+  fingerprint: string;
+  queuedAt: number;
+  state: "queued" | "accepted";
+  deliveryId?: string;
+}
+
+interface BossControlOutboxState {
+  version: typeof BOSS_CONTROL_OUTBOX_VERSION;
+  entries: StoredBossControl[];
+}
+
+function scope(envelope: BossControlEnvelope): string {
+  return canonicalHash("agent-intercom-codex/boss-control/outbox-scope/v1", {
+    bossRunId: envelope.bossRunId,
+    participantId: envelope.participantId,
+    bindingEpoch: Number(envelope.bindingEpoch),
+    idempotencyKey: envelope.idempotencyKey,
+  });
+}
+
+function fingerprint(to: string, envelope: BossControlEnvelope): string {
+  const { messageId: _transportMessageId, ...stableEnvelope } = envelope;
+  return canonicalHash("agent-intercom-codex/boss-control/outbox-request/v1", { to, envelope: stableEnvelope });
+}
+
+function exactKeys(value: Record<string, unknown>, required: readonly string[], optional: readonly string[] = []): boolean {
+  const permitted = new Set([...required, ...optional]);
+  const keys = Reflect.ownKeys(value);
+  return required.every((key) => Object.hasOwn(value, key))
+    && keys.every((key) => typeof key === "string" && permitted.has(key));
+}
+
+function parseEntry(value: unknown): StoredBossControl {
+  assertBossCanonicalData(value, "$.bossControlOutbox.entries[]");
+  if (typeof value !== "object" || value === null || Array.isArray(value)) throw new Error("Invalid Boss outbox entry");
+  const entry = value as Record<string, unknown>;
+  if (!exactKeys(entry, ["to", "envelope", "scope", "fingerprint", "queuedAt", "state"], ["deliveryId"])) {
+    throw new Error("Invalid Boss outbox entry fields");
+  }
+  if (
+    typeof entry.to !== "string"
+    || entry.to.length === 0
+    || typeof entry.scope !== "string"
+    || !/^[a-f0-9]{64}$/.test(entry.scope)
+    || typeof entry.fingerprint !== "string"
+    || !/^[a-f0-9]{64}$/.test(entry.fingerprint)
+    || typeof entry.queuedAt !== "number"
+    || !Number.isSafeInteger(entry.queuedAt)
+    || (entry.state !== "queued" && entry.state !== "accepted")
+    || (entry.state === "queued" && Object.hasOwn(entry, "deliveryId"))
+    || (entry.state === "accepted" && (!Object.hasOwn(entry, "deliveryId") || typeof entry.deliveryId !== "string" || entry.deliveryId.length === 0))
+  ) throw new Error("Invalid Boss outbox entry binding");
+  const envelope = parseBossControlEnvelope(entry.envelope);
+  if (entry.scope !== scope(envelope) || entry.fingerprint !== fingerprint(entry.to, envelope)) {
+    throw new Error("Boss outbox entry canonical binding mismatch");
+  }
+  return {
+    to: entry.to,
+    envelope,
+    scope: entry.scope,
+    fingerprint: entry.fingerprint,
+    queuedAt: entry.queuedAt,
+    state: entry.state,
+    ...(entry.deliveryId === undefined ? {} : { deliveryId: entry.deliveryId as string }),
+  };
+}
+
+function fileName(sessionId: string): string {
+  return `${createHash("sha256").update(sessionId).digest("hex")}.json`;
+}
+
+export class PersistentBossControlOutbox {
+  private readonly path: string;
+  private state: BossControlOutboxState;
+
+  constructor(sessionId: string, intercomDir: string = getIntercomDirPath()) {
+    ensureIntercomRuntimeDir(intercomDir);
+    const directory = join(intercomDir, "boss-control-outbox");
+    mkdirSync(directory, { recursive: true, mode: INTERCOM_DIR_MODE });
+    if (process.platform !== "win32") chmodSync(directory, INTERCOM_DIR_MODE);
+    this.path = join(directory, fileName(sessionId));
+    this.state = this.load();
+  }
+
+  list(): StoredBossControl[] {
+    return structuredClone(this.state.entries);
+  }
+
+  find(idempotencyKey: string): StoredBossControl | undefined {
+    const entry = this.state.entries.find((candidate) => candidate.envelope.idempotencyKey === idempotencyKey);
+    return entry === undefined ? undefined : structuredClone(entry);
+  }
+
+  enqueue(to: string, envelopeValue: unknown): "added" | "existing" {
+    if (typeof to !== "string" || to.length === 0) throw new Error("Boss target session ID is required");
+    assertBossCanonicalData(envelopeValue, "$.envelope");
+    const envelope = parseBossControlEnvelope(envelopeValue);
+    const candidateScope = scope(envelope);
+    const candidateFingerprint = fingerprint(to, envelope);
+    const existing = this.state.entries.find((entry) => entry.scope === candidateScope);
+    if (existing) {
+      if (existing.fingerprint !== candidateFingerprint) {
+        throw new Error(`Boss idempotency key ${envelope.idempotencyKey} is queued with a different canonical request`);
+      }
+      if (existing.envelope.messageId !== envelope.messageId) {
+        existing.envelope = envelope;
+        existing.queuedAt = Date.now();
+        this.persist();
+      }
+      return "existing";
+    }
+    if (this.state.entries.some((entry) => entry.envelope.messageId === envelope.messageId)) {
+      throw new Error(`Boss message ID ${envelope.messageId} is queued with a different idempotency scope`);
+    }
+    if (this.state.entries.length >= MAX_BOSS_CONTROL_OUTBOX_ENTRIES) throw new Error("Durable Boss control outbox is full");
+    this.state.entries.push({
+      to,
+      envelope,
+      scope: candidateScope,
+      fingerprint: candidateFingerprint,
+      queuedAt: Date.now(),
+      state: "queued",
+    });
+    this.persist();
+    return "added";
+  }
+
+  markAccepted(idempotencyKey: string, messageId: string, deliveryId: string): "accepted" | "already-accepted" {
+    const entry = this.state.entries.find((candidate) => candidate.envelope.idempotencyKey === idempotencyKey);
+    if (!entry || entry.envelope.messageId !== messageId || !deliveryId) {
+      throw new Error("Boss acknowledgement does not match the durable outbox binding");
+    }
+    if (entry.state === "accepted") {
+      if (entry.deliveryId !== deliveryId) throw new Error("Boss acknowledgement changed the durable deliveryId");
+      return "already-accepted";
+    }
+    entry.state = "accepted";
+    entry.deliveryId = deliveryId;
+    this.persist();
+    return "accepted";
+  }
+
+  removeCorrelated(idempotencyKey: string, messageId: string, deliveryId?: string): void {
+    const index = this.state.entries.findIndex((candidate) => candidate.envelope.idempotencyKey === idempotencyKey);
+    if (index < 0) throw new Error("Boss terminal result has no durable outbox binding");
+    const entry = this.state.entries[index];
+    if (entry.envelope.messageId !== messageId) throw new Error("Boss terminal result messageId does not match the durable caller");
+    if (deliveryId === undefined) {
+      if (entry.state !== "queued") throw new Error("Boss post-acceptance failure omitted the durable deliveryId");
+    } else if (entry.state !== "accepted" || entry.deliveryId !== deliveryId) {
+      throw new Error("Boss terminal result arrived before the matching durable acknowledgement");
+    }
+    this.state.entries.splice(index, 1);
+    this.persist();
+  }
+
+  private load(): BossControlOutboxState {
+    if (!existsSync(this.path)) return { version: BOSS_CONTROL_OUTBOX_VERSION, entries: [] };
+    try {
+      const parsed: unknown = JSON.parse(readFileSync(this.path, "utf8"));
+      assertBossCanonicalData(parsed, "$.bossControlOutbox");
+      if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) throw new Error("expected object");
+      const state = parsed as Record<string, unknown>;
+      if (!exactKeys(state, ["version", "entries"]) || state.version !== BOSS_CONTROL_OUTBOX_VERSION || !Array.isArray(state.entries)) {
+        throw new Error("invalid Boss outbox state");
+      }
+      return { version: BOSS_CONTROL_OUTBOX_VERSION, entries: state.entries.map(parseEntry) };
+    } catch (error) {
+      const corruptPath = `${this.path}.corrupt-${Date.now()}`;
+      renameSync(this.path, corruptPath);
+      restrictIntercomRuntimeFile(corruptPath);
+      throw new Error(`Boss control outbox was corrupt and quarantined at ${corruptPath}`, { cause: error });
+    }
+  }
+
+  private persist(): void {
+    writeDurableJson(this.path, this.state);
+  }
+}

broker/authorization.test.ts

@@ -2,6 +2,7 @@ import assert from "node:assert/strict";
 import test from "node:test";
 import { authorizeSessionAction, visibleSessions } from "./authorization.ts";
 import type { SessionInfo } from "../types.ts";
+import { BOSS_PARTICIPANT_BINDING_VERSION, BOSS_RUN_FEATURE_CONTRACT, type BossParticipantRole } from "@dataforxyz/agent-intercom-core/boss";
 
 function local(id: string): SessionInfo {
   return { id, name: id, cwd: "/tmp", model: "test", pid: 1, startedAt: 1, lastActivity: 1, origin: "local" };
@@ -32,6 +33,37 @@ const sessions = [
   remote("child-b", "manager"),
 ];
 
+function boss(
+  id: string,
+  bossRunId: string,
+  participantId: string,
+  role: BossParticipantRole,
+  options: { manager?: string; assigned?: string[] } = {},
+): SessionInfo {
+  return {
+    ...local(id),
+    boss: {
+      featureContract: BOSS_RUN_FEATURE_CONTRACT,
+      binding: {
+        version: BOSS_PARTICIPANT_BINDING_VERSION,
+        bossRunId,
+        participantId,
+        role,
+        communicationProfile: role,
+        bindingEpoch: 1,
+        sessionId: id,
+        brokerGeneration: 1,
+        brokerBootInstance: "boot-1",
+        state: "active",
+        ...(options.manager === undefined ? {} : { assignedManagerParticipantId: options.manager }),
+        authorityTransitionId: "transition-1",
+      },
+      brokerIdentityVerified: true,
+      ...(options.assigned === undefined ? {} : { assignedParticipantIds: options.assigned }),
+    },
+  };
+}
+
 test("phase one discovery and communication use the same ancestor-chain policy", () => {
   assert.equal(authorizeSessionAction(sessions, "root", "send", "manager").allowed, true);
   assert.equal(authorizeSessionAction(sessions, "manager", "ask", "root").allowed, true);
@@ -47,3 +79,37 @@ test("visibility hides unauthorized sessions rather than revealing denial detail
   assert.deepEqual(visibleSessions(sessions, "root").map((session) => session.id).sort(), ["child-a", "child-b", "manager", "root", "unrelated"]);
   assert.deepEqual(visibleSessions(sessions, "unrelated").map((session) => session.id).sort(), ["root", "unrelated"]);
 });
+
+test("Boss registrations are isolated from ordinary sessions and other runs", () => {
+  const bossSessions = [
+    local("ordinary"),
+    boss("manager", "run-1", "manager-1", "manager", { assigned: ["worker-1"] }),
+    boss("worker", "run-1", "worker-1", "worker", { manager: "manager-1" }),
+    boss("other-run", "run-2", "worker-2", "worker", { manager: "manager-2" }),
+  ];
+  assert.equal(authorizeSessionAction(bossSessions, "manager", "send", "worker").allowed, true);
+  assert.equal(authorizeSessionAction(bossSessions, "worker", "discover", "other-run").allowed, false);
+  assert.equal(authorizeSessionAction(bossSessions, "ordinary", "discover", "worker").allowed, false);
+  assert.deepEqual(visibleSessions(bossSessions, "worker").map((session) => session.id).sort(), ["manager", "worker"]);
+});
+
+test("Boss structured control uses the directional Core policy matrix", () => {
+  const bossSessions = [
+    boss("manager", "run-1", "manager-1", "manager", { assigned: ["worker-1"] }),
+    boss("worker", "run-1", "worker-1", "worker", { manager: "manager-1" }),
+  ];
+  assert.equal(authorizeSessionAction(
+    bossSessions,
+    "manager",
+    "control",
+    "worker",
+    { controlKind: "assignment_request", correlated: true },
+  ).allowed, true);
+  assert.equal(authorizeSessionAction(
+    bossSessions,
+    "worker",
+    "control",
+    "manager",
+    { controlKind: "assignment_request", correlated: true },
+  ).allowed, false);
+});