feat(mosaic): add correlated lease promotion CLI

2026-08-08 16:05:04 -05:00
parent a5e88a8d97
commit 9d0822570c
7 changed files with 798 additions and 6 deletions
@@ -10,15 +10,18 @@ import secrets
 import stat
 import subprocess
 import sys
+import time
 from collections.abc import Callable, Mapping
 from pathlib import Path
 from typing import Final, NamedTuple, TextIO

 MAX_FRAME: Final = 64 * 1024
 PROMOTER_TIMEOUT_SECONDS: Final = 10.0
+LEASE_TTL_SECONDS: Final = 60 * 60
 PROMOTER: Final = Path(__file__).resolve().with_name("lease_promote.py")
 PENDING_DIRECTORY: Final = "mosaic-lease"
 LOCK_FILE: Final = "promotion.lock"
+RESULT_FILE: Final = "last-result.json"
 TERMINAL_FAILURE_CODES: Final = frozenset(
    {
        "RECEIPT_REPLAY",
@@ -145,6 +148,54 @@ def read_pending(directory_descriptor: int, name: str) -> PendingChallenge | Non
    return PendingChallenge(challenge, metadata.st_dev, metadata.st_ino)


+def write_result(
+    directory_descriptor: int,
+    attempt_id: str,
+    verified: bool,
+    reason: str | None,
+    session_id: str,
+    wall_clock: float,
+) -> None:
+    result = {
+        "attempt_id": attempt_id,
+        "expires_at_wallclock": wall_clock + LEASE_TTL_SECONDS if verified else None,
+        "reason": reason,
+        "session_id": session_id,
+        "ts": wall_clock,
+        "verified": verified,
+    }
+    temporary = f".{RESULT_FILE}.tmp-{secrets.token_hex(8)}"
+    flags = (
+        os.O_WRONLY
+        | os.O_CREAT
+        | os.O_EXCL
+        | getattr(os, "O_CLOEXEC", 0)
+        | getattr(os, "O_NOFOLLOW", 0)
+    )
+    descriptor = os.open(temporary, flags, 0o600, dir_fd=directory_descriptor)
+    try:
+        os.fchmod(descriptor, 0o600)
+        with os.fdopen(descriptor, "w", encoding="utf-8", closefd=False) as stream:
+            json.dump(result, stream, separators=(",", ":"), sort_keys=True)
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(
+            temporary,
+            RESULT_FILE,
+            src_dir_fd=directory_descriptor,
+            dst_dir_fd=directory_descriptor,
+        )
+        os.fsync(directory_descriptor)
+    except Exception:
+        try:
+            os.unlink(temporary, dir_fd=directory_descriptor)
+        except FileNotFoundError:
+            pass
+        raise
+    finally:
+        os.close(descriptor)
+
+
 def delete_pending_if_unchanged(
    directory_descriptor: int,
    name: str,
@@ -233,6 +284,7 @@ def main(
    environ: Mapping[str, str] | None = None,
    stderr: TextIO | None = None,
    run: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
+    now: Callable[[], float] = time.time,
 ) -> int:
    source_environment = os.environ if environ is None else environ
    error_stream = sys.stderr if stderr is None else stderr
@@ -241,6 +293,7 @@ def main(

    try:
        runtime_dir, pending_name = session_pending_name(source_environment)
+        session_id = source_environment["MOSAIC_LEASE_SESSION_ID"]
        directory_descriptor = open_pending_directory(runtime_dir)
        if directory_descriptor is None:
            return 0
@@ -270,6 +323,7 @@ def main(
        )
        reply = parse_reply(completed)
        if reply is not None and reply.get("ok") is True:
+            write_result(directory_descriptor, pending.value, True, None, session_id, now())
            delete_pending_if_unchanged(
                directory_descriptor,
                pending_name,
@@ -283,6 +337,7 @@ def main(
            code = str(reply["code"])
            print(f"Mosaic promotion incomplete: {code}.", file=error_stream)
            if code in TERMINAL_FAILURE_CODES:
+                write_result(directory_descriptor, pending.value, False, code, session_id, now())
                delete_pending_if_unchanged(
                    directory_descriptor,
                    pending_name,
@@ -15,6 +15,7 @@ import { registerAgentCommand } from './commands/agent.js';
 import { registerInteractionCommand } from './commands/interaction.js';
 import { registerConfigCommand } from './commands/config.js';
 import { registerFleetCommand } from './commands/fleet.js';
+import { registerPromoteCommand } from './commands/promote.js';
 import { registerMissionCommand } from './commands/mission.js';
 import { registerUninstallCommand } from './commands/uninstall.js';
 import { registerRestoreCommand } from './commands/restore.js';
@@ -370,6 +371,7 @@ registerInteractionCommand(program);
 // ─── fleet ─────────────────────────────────────────────────────────────

 registerFleetCommand(program);
+registerPromoteCommand(program);

 // ─── config ────────────────────────────────────────────────────────────

@@ -0,0 +1,219 @@
+import { Command } from 'commander';
+import { describe, expect, it, vi } from 'vitest';
+import {
+  promoteSeat,
+  registerPromoteCommand,
+  type PromotionBreadcrumbStore,
+  type PromotionTransport,
+} from './promote.js';
+
+const attemptId = 'a'.repeat(64);
+const target = {
+  bundle: 'local',
+  seat: 'claude-seat',
+  sessionId: 'b'.repeat(64),
+};
+
+function transport(): PromotionTransport {
+  return {
+    resolve: vi.fn(async () => target),
+    sendPromotion: vi.fn(async () => {}),
+  };
+}
+
+describe('mosaic promote', () => {
+  it('accepts only a fresh result correlated to this attempt', async () => {
+    const promotionTransport = transport();
+    const store: PromotionBreadcrumbStore = {
+      readAttemptId: vi.fn().mockResolvedValueOnce(null).mockResolvedValue(attemptId),
+      readResult: vi
+        .fn()
+        .mockResolvedValueOnce({
+          attempt_id: 'c'.repeat(64),
+          expires_at_wallclock: 4_600,
+          reason: null,
+          session_id: target.sessionId,
+          ts: 1_001,
+          verified: true,
+        })
+        .mockResolvedValueOnce({
+          attempt_id: attemptId,
+          expires_at_wallclock: 4_600,
+          reason: null,
+          session_id: target.sessionId,
+          ts: 1_001,
+          verified: true,
+        }),
+    };
+
+    const result = await promoteSeat('claude-seat', {
+      clock: () => 1_000,
+      sleep: async () => {},
+      store,
+      timeoutMs: 1,
+      transport: promotionTransport,
+    });
+
+    expect(result).toEqual({
+      bundle: 'local',
+      expiresAtWallclock: 4_600,
+      reason: null,
+      seat: 'claude-seat',
+      sessionId: 'b'.repeat(64),
+      status: 'VERIFIED',
+    });
+    expect(promotionTransport.sendPromotion).toHaveBeenCalledWith(target);
+    expect(store.readResult).toHaveBeenCalledTimes(2);
+  });
+
+  it('accepts a fresh result for this session when completion consumed the pending nonce', async () => {
+    const promotionTransport = transport();
+    let now = 1_000;
+    const store: PromotionBreadcrumbStore = {
+      readAttemptId: vi.fn(async () => null),
+      readResult: vi.fn(async () => ({
+        attempt_id: attemptId,
+        expires_at_wallclock: 4_600,
+        reason: null,
+        session_id: target.sessionId,
+        ts: 1_001,
+        verified: true,
+      })),
+    };
+
+    const result = await promoteSeat('claude-seat', {
+      clock: () => now,
+      sleep: async () => {
+        now += 10;
+      },
+      store,
+      timeoutMs: 10,
+      transport: promotionTransport,
+    });
+
+    expect(result.status).toBe('VERIFIED');
+  });
+
+  it('prints VERIFIED with the resolved seat, session, bundle, and wall-clock expiry', async () => {
+    const promotionTransport = transport();
+    const store: PromotionBreadcrumbStore = {
+      readAttemptId: vi.fn().mockResolvedValueOnce(null).mockResolvedValue(attemptId),
+      readResult: vi.fn(async () => ({
+        attempt_id: attemptId,
+        expires_at_wallclock: 4_600,
+        reason: null,
+        session_id: target.sessionId,
+        ts: Number.MAX_SAFE_INTEGER,
+        verified: true,
+      })),
+    };
+    const output = vi.spyOn(console, 'log').mockImplementation(() => {});
+    const program = new Command().exitOverride();
+    registerPromoteCommand(program, { store, transport: promotionTransport });
+
+    try {
+      await program.parseAsync(['node', 'mosaic', 'promote', 'claude-seat']);
+      expect(output).toHaveBeenCalledWith(
+        `VERIFIED seat=claude-seat session=${target.sessionId} bundle=local expiry=1970-01-01T01:16:40.000Z`,
+      );
+      expect(process.exitCode).not.toBe(1);
+    } finally {
+      output.mockRestore();
+      process.exitCode = undefined;
+    }
+  });
+
+  it('rejects a stale result even when its nonce matches', async () => {
+    const promotionTransport = transport();
+    let now = 1_000;
+    const store: PromotionBreadcrumbStore = {
+      readAttemptId: vi.fn().mockResolvedValueOnce(null).mockResolvedValue(attemptId),
+      readResult: vi.fn(async () => ({
+        attempt_id: attemptId,
+        expires_at_wallclock: 4_600,
+        reason: null,
+        session_id: target.sessionId,
+        ts: 1_000,
+        verified: true,
+      })),
+    };
+
+    const result = await promoteSeat('claude-seat', {
+      clock: () => now,
+      sleep: async () => {
+        now += 10;
+      },
+      store,
+      timeoutMs: 10,
+      transport: promotionTransport,
+    });
+
+    expect(result).toEqual({
+      bundle: 'local',
+      expiresAtWallclock: null,
+      reason: 'PROMOTION_TIMEOUT',
+      seat: 'claude-seat',
+      sessionId: 'b'.repeat(64),
+      status: 'UNVERIFIED',
+    });
+  });
+
+  it('does not accept a result for a pending attempt that existed before send', async () => {
+    const promotionTransport = transport();
+    let now = 1_000;
+    const store: PromotionBreadcrumbStore = {
+      readAttemptId: vi.fn(async () => attemptId),
+      readResult: vi.fn(async () => ({
+        attempt_id: attemptId,
+        expires_at_wallclock: 4_600,
+        reason: null,
+        session_id: target.sessionId,
+        ts: 1_001,
+        verified: true,
+      })),
+    };
+
+    const result = await promoteSeat('claude-seat', {
+      clock: () => now,
+      sleep: async () => {
+        now += 10;
+      },
+      store,
+      timeoutMs: 10,
+      transport: promotionTransport,
+    });
+
+    expect(result.status).toBe('UNVERIFIED');
+    expect(store.readResult).not.toHaveBeenCalled();
+  });
+
+  it('returns UNVERIFIED after a bounded timeout without reading stdin', async () => {
+    const promotionTransport = transport();
+    let now = 1_000;
+    const store: PromotionBreadcrumbStore = {
+      readAttemptId: vi.fn(async () => null),
+      readResult: vi.fn(async () => null),
+    };
+
+    const result = await promoteSeat('claude-seat', {
+      clock: () => now,
+      sleep: async () => {
+        now += 10;
+      },
+      store,
+      timeoutMs: 10,
+      transport: promotionTransport,
+    });
+
+    expect(result).toEqual({
+      bundle: 'local',
+      expiresAtWallclock: null,
+      reason: 'PROMOTION_TIMEOUT',
+      seat: 'claude-seat',
+      sessionId: 'b'.repeat(64),
+      status: 'UNVERIFIED',
+    });
+    expect(promotionTransport.sendPromotion).toHaveBeenCalledOnce();
+    expect(store.readResult).toHaveBeenCalledTimes(2);
+  });
+});
@@ -0,0 +1,284 @@
+import { spawn } from 'node:child_process';
+import { constants } from 'node:fs';
+import { open } from 'node:fs/promises';
+import { join } from 'node:path';
+import type { Command } from 'commander';
+import {
+  TmuxPromotionTransport,
+  type PromotionTarget,
+  type PromotionTransport,
+} from '../fleet/promotion-transport.js';
+
+export type { PromotionTransport } from '../fleet/promotion-transport.js';
+import { resolveFleetPaths, type CommandRunner } from './fleet.js';
+
+const ATTEMPT_ID_PATTERN = /^[a-f0-9]{64}$/;
+const DEFAULT_POLL_INTERVAL_MS = 250;
+const DEFAULT_TIMEOUT_MS = 30_000;
+const PENDING_DIRECTORY = 'mosaic-lease';
+const RESULT_FILE = 'last-result.json';
+
+export interface PromotionBreadcrumb {
+  attempt_id: string;
+  expires_at_wallclock: number | null;
+  reason: string | null;
+  session_id: string;
+  ts: number;
+  verified: boolean;
+}
+
+export interface PromotionBreadcrumbStore {
+  readAttemptId(sessionId: string): Promise<string | null>;
+  readResult(): Promise<PromotionBreadcrumb | null>;
+}
+
+export interface PromotionResult {
+  bundle: string;
+  expiresAtWallclock: number | null;
+  reason: string | null;
+  seat: string;
+  sessionId: string;
+  status: 'VERIFIED' | 'UNVERIFIED';
+}
+
+export interface PromoteSeatOptions {
+  clock?: () => number;
+  pollIntervalMs?: number;
+  sleep?: (milliseconds: number) => Promise<void>;
+  store: PromotionBreadcrumbStore;
+  timeoutMs?: number;
+  transport: PromotionTransport;
+}
+
+export interface PromoteCommandDeps {
+  mosaicHome?: string;
+  runner?: CommandRunner;
+  store?: PromotionBreadcrumbStore;
+  transport?: PromotionTransport;
+}
+
+/** Private, local result store shared with the in-seat completion hook. */
+export class FilePromotionBreadcrumbStore implements PromotionBreadcrumbStore {
+  constructor(private readonly runtimeDirectory = defaultRuntimeDirectory()) {}
+
+  async readAttemptId(sessionId: string): Promise<string | null> {
+    if (!ATTEMPT_ID_PATTERN.test(sessionId)) return null;
+    const content = await readPrivateFile(
+      join(this.runtimeDirectory, PENDING_DIRECTORY, `pending-${sessionId}`),
+    );
+    const attemptId = content?.trim();
+    return attemptId !== undefined && ATTEMPT_ID_PATTERN.test(attemptId) ? attemptId : null;
+  }
+
+  async readResult(): Promise<PromotionBreadcrumb | null> {
+    const content = await readPrivateFile(
+      join(this.runtimeDirectory, PENDING_DIRECTORY, RESULT_FILE),
+    );
+    if (content === null) return null;
+    try {
+      return parseBreadcrumb(JSON.parse(content) as unknown);
+    } catch {
+      return null;
+    }
+  }
+}
+
+/** Drives one bounded, non-interactive in-seat promotion attempt. */
+export async function promoteSeat(
+  seat: string,
+  options: PromoteSeatOptions,
+): Promise<PromotionResult> {
+  const clock = options.clock ?? wallClockSeconds;
+  const sleep = options.sleep ?? defaultSleep;
+  const timeoutMs = normalizeTimeout(options.timeoutMs);
+  const pollIntervalMs = normalizePollInterval(options.pollIntervalMs);
+  const target = await options.transport.resolve(seat);
+  const previousAttemptId = await options.store.readAttemptId(target.sessionId);
+  const preSendTimestamp = clock();
+  try {
+    await options.transport.sendPromotion(target);
+  } catch (error: unknown) {
+    return unverified(target, `DELIVERY_FAILED: ${errorMessage(error)}`);
+  }
+
+  const deadline = preSendTimestamp + timeoutMs / 1_000;
+  let attemptId: string | null = null;
+  while (true) {
+    const currentAttemptId = await options.store.readAttemptId(target.sessionId);
+    if (currentAttemptId !== null && currentAttemptId !== previousAttemptId) {
+      attemptId = currentAttemptId;
+    }
+    if (attemptId !== null || previousAttemptId === null) {
+      const breadcrumb = await options.store.readResult();
+      if (
+        breadcrumb !== null &&
+        breadcrumb.session_id === target.sessionId &&
+        (attemptId === null || breadcrumb.attempt_id === attemptId) &&
+        breadcrumb.ts > preSendTimestamp
+      ) {
+        return {
+          bundle: target.bundle,
+          expiresAtWallclock: breadcrumb.expires_at_wallclock,
+          reason: breadcrumb.reason,
+          seat: target.seat,
+          sessionId: target.sessionId,
+          status: breadcrumb.verified ? 'VERIFIED' : 'UNVERIFIED',
+        };
+      }
+    }
+    if (clock() >= deadline) return unverified(target, 'PROMOTION_TIMEOUT');
+    await sleep(Math.min(pollIntervalMs, Math.max(0, deadline - clock()) * 1_000));
+  }
+}
+
+export function registerPromoteCommand(program: Command, deps: PromoteCommandDeps = {}): void {
+  const mosaicHome = deps.mosaicHome ?? resolveFleetPaths().mosaicHome;
+  const transport =
+    deps.transport ?? new TmuxPromotionTransport({ mosaicHome, runner: deps.runner ?? runCommand });
+  const store = deps.store ?? new FilePromotionBreadcrumbStore();
+
+  program
+    .command('promote <seat>')
+    .description('Promote a Claude fleet seat and report the correlated lease result')
+    .option(
+      '--timeout <ms>',
+      `Bounded result wait in milliseconds (default: ${DEFAULT_TIMEOUT_MS})`,
+    )
+    .action(async (seat: string, opts: { timeout?: string }) => {
+      const result = await promoteSeat(seat, {
+        store,
+        timeoutMs: parseOptionTimeout(opts.timeout),
+        transport,
+      });
+      const expiry =
+        result.expiresAtWallclock === null
+          ? 'none'
+          : new Date(result.expiresAtWallclock * 1_000).toISOString();
+      const reason = result.reason === null ? '' : ` reason=${result.reason}`;
+      console.log(
+        `${result.status} seat=${result.seat} session=${result.sessionId} bundle=${result.bundle} expiry=${expiry}${reason}`,
+      );
+      if (result.status === 'UNVERIFIED') process.exitCode = 1;
+    });
+}
+
+function defaultRuntimeDirectory(): string {
+  const configured = process.env['XDG_RUNTIME_DIR'];
+  if (configured) return configured;
+  const uid = typeof process.getuid === 'function' ? process.getuid() : 0;
+  return `/run/user/${uid}`;
+}
+
+async function readPrivateFile(path: string): Promise<string | null> {
+  let handle: Awaited<ReturnType<typeof open>>;
+  try {
+    handle = await open(path, constants.O_RDONLY | constants.O_NOFOLLOW);
+  } catch {
+    return null;
+  }
+  try {
+    const metadata = await handle.stat();
+    if (
+      !metadata.isFile() ||
+      metadata.uid !== (typeof process.getuid === 'function' ? process.getuid() : 0) ||
+      (metadata.mode & 0o077) !== 0
+    ) {
+      return null;
+    }
+    return handle.readFile({ encoding: 'utf8' });
+  } catch {
+    return null;
+  } finally {
+    await handle.close();
+  }
+}
+
+function parseBreadcrumb(value: unknown): PromotionBreadcrumb | null {
+  if (!isRecord(value) || Object.keys(value).length !== 6) return null;
+  const { attempt_id, expires_at_wallclock, reason, session_id, ts, verified } = value;
+  if (
+    typeof attempt_id !== 'string' ||
+    !ATTEMPT_ID_PATTERN.test(attempt_id) ||
+    typeof verified !== 'boolean' ||
+    typeof session_id !== 'string' ||
+    !ATTEMPT_ID_PATTERN.test(session_id) ||
+    typeof ts !== 'number' ||
+    !Number.isFinite(ts) ||
+    (expires_at_wallclock !== null &&
+      (typeof expires_at_wallclock !== 'number' || !Number.isFinite(expires_at_wallclock))) ||
+    (reason !== null && typeof reason !== 'string')
+  ) {
+    return null;
+  }
+  return { attempt_id, expires_at_wallclock, reason, session_id, ts, verified };
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === 'object' && value !== null && !Array.isArray(value);
+}
+
+function normalizeTimeout(value: number | undefined): number {
+  return value !== undefined && Number.isFinite(value) ? Math.max(0, value) : DEFAULT_TIMEOUT_MS;
+}
+
+function normalizePollInterval(value: number | undefined): number {
+  return value !== undefined && Number.isFinite(value)
+    ? Math.max(1, value)
+    : DEFAULT_POLL_INTERVAL_MS;
+}
+
+function parseOptionTimeout(value: string | undefined): number | undefined {
+  if (value === undefined) return undefined;
+  const parsed = Number.parseInt(value, 10);
+  return Number.isFinite(parsed) ? parsed : undefined;
+}
+
+function unverified(target: PromotionTarget, reason: string): PromotionResult {
+  return {
+    bundle: target.bundle,
+    expiresAtWallclock: null,
+    reason,
+    seat: target.seat,
+    sessionId: target.sessionId,
+    status: 'UNVERIFIED',
+  };
+}
+
+function errorMessage(error: unknown): string {
+  return error instanceof Error ? error.message : String(error);
+}
+
+function wallClockSeconds(): number {
+  return Date.now() / 1_000;
+}
+
+function defaultSleep(milliseconds: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, milliseconds));
+}
+
+function runCommand(
+  command: string,
+  args: string[],
+): Promise<{
+  exitCode: number;
+  stderr: string;
+  stdout: string;
+}> {
+  return new Promise((resolve) => {
+    const child = spawn(command, args, { stdio: ['ignore', 'pipe', 'pipe'] });
+    let stdout = '';
+    let stderr = '';
+    child.stdout.on('data', (chunk: Buffer) => {
+      stdout += chunk.toString('utf8');
+    });
+    child.stderr.on('data', (chunk: Buffer) => {
+      stderr += chunk.toString('utf8');
+    });
+    child.on('error', (error: Error) => {
+      resolve({ exitCode: 127, stderr: error.message, stdout });
+    });
+    child.on('close', (code: number | null) => {
+      resolve({ exitCode: code ?? 1, stderr, stdout });
+    });
+  });
+}
@@ -0,0 +1,53 @@
+import { describe, expect, it, vi } from 'vitest';
+import type { CommandResult, CommandRunner, FleetRoster } from '../commands/fleet.js';
+import { TmuxPromotionTransport } from './promotion-transport.js';
+
+const sessionId = 'a'.repeat(64);
+const roster: FleetRoster = {
+  agents: [{ className: 'worker', name: 'claude-seat', runtime: 'claude' }],
+  defaults: { workingDirectory: '~/src' },
+  runtimes: {},
+  tmux: { holderSession: '_holder', socketName: 'mosaic-fleet' },
+  transport: 'tmux',
+  version: 1,
+};
+
+function result(stdout = '', exitCode = 0, stderr = ''): CommandResult {
+  return { exitCode, stderr, stdout };
+}
+
+describe('TmuxPromotionTransport', () => {
+  it('resolves the exact roster seat and sends through the maintained tmux sender', async () => {
+    const runner = vi
+      .fn<CommandRunner>()
+      .mockResolvedValueOnce(result('1234 claude 0 0 0 0\n'))
+      .mockResolvedValueOnce(result());
+    const environmentReader = vi.fn(async () => `MOSAIC_LEASE_SESSION_ID=${sessionId}\0`);
+    const transport = new TmuxPromotionTransport({
+      environmentReader,
+      mosaicHome: '/mosaic',
+      rosterLoader: async () => roster,
+      runner,
+    });
+
+    const target = await transport.resolve('claude-seat');
+    await transport.sendPromotion(target);
+
+    expect(target).toEqual({
+      bundle: 'mosaic-fleet',
+      seat: 'claude-seat',
+      sessionId,
+    });
+    expect(environmentReader).toHaveBeenCalledWith(1234);
+    expect(runner).toHaveBeenNthCalledWith(2, '/mosaic/tools/tmux/agent-send.sh', [
+      '-L',
+      'mosaic-fleet',
+      '-S',
+      expect.stringMatching(/:operator$/),
+      '-s',
+      'claude-seat',
+      '-m',
+      '/mosaic-promote',
+    ]);
+  });
+});
@@ -0,0 +1,116 @@
+import { readFile } from 'node:fs/promises';
+import {
+  buildAgentSendCommand,
+  buildTmuxListPanesCommand,
+  getDefaultOperatorSourceLabel,
+  getRosterAgent,
+  parseTmuxListPanes,
+  resolveFleetPaths,
+  type CommandResult,
+  type CommandRunner,
+  type FleetRoster,
+  RUNTIME_ACCEPTABLE_COMMANDS,
+} from '../commands/fleet.js';
+import { loadFleetRoster } from './fleet-roster-v1.js';
+
+const PROMOTION_COMMAND = '/mosaic-promote';
+const SESSION_ID_PATTERN = /^[a-f0-9]{64}$/;
+
+export interface PromotionTarget {
+  bundle: string;
+  seat: string;
+  sessionId: string;
+}
+
+export interface PromotionTransport {
+  resolve(seat: string): Promise<PromotionTarget>;
+  sendPromotion(target: PromotionTarget): Promise<void>;
+}
+
+export interface TmuxPromotionTransportOptions {
+  environmentReader?: (pid: number) => Promise<string>;
+  mosaicHome: string;
+  rosterLoader?: () => Promise<FleetRoster>;
+  runner: CommandRunner;
+}
+
+/** Local, roster-bound transport for the in-seat promotion command. */
+export class TmuxPromotionTransport implements PromotionTransport {
+  private readonly environmentReader: (pid: number) => Promise<string>;
+  private readonly rosterLoader: () => Promise<FleetRoster>;
+
+  constructor(private readonly options: TmuxPromotionTransportOptions) {
+    this.environmentReader = options.environmentReader ?? readPaneEnvironment;
+    this.rosterLoader =
+      options.rosterLoader ??
+      (() => loadFleetRoster(resolveFleetPaths(options.mosaicHome).rosterPath));
+  }
+
+  async resolve(seat: string): Promise<PromotionTarget> {
+    const roster = await this.rosterLoader();
+    const agent = getRosterAgent(roster, seat);
+    if (agent.runtime !== 'claude') {
+      throw new Error(`Lease promotion is currently available only for Claude seats: ${seat}.`);
+    }
+    const paneResult = await this.run(
+      buildTmuxListPanesCommand(agent.name, roster.tmux.socketName),
+    );
+    if (paneResult.exitCode !== 0) {
+      throw new Error(`Promotion seat is unavailable: ${seat}.`);
+    }
+    const pane = parseTmuxListPanes(paneResult.stdout);
+    const allowedCommands = RUNTIME_ACCEPTABLE_COMMANDS.claude;
+    if (
+      pane.dead ||
+      pane.pid === null ||
+      pane.command === null ||
+      allowedCommands === undefined ||
+      !allowedCommands.includes(pane.command)
+    ) {
+      throw new Error(`Promotion seat runtime identity mismatch: ${seat}.`);
+    }
+    const sessionId = parseLeaseSessionId(await this.environmentReader(pane.pid));
+    if (sessionId === null) {
+      throw new Error(`Promotion seat has no readable lease session: ${seat}.`);
+    }
+    return {
+      bundle: roster.tmux.socketName || 'default',
+      seat: agent.name,
+      sessionId,
+    };
+  }
+
+  async sendPromotion(target: PromotionTarget): Promise<void> {
+    const command = buildAgentSendCommand(
+      resolveFleetPaths(this.options.mosaicHome),
+      target.seat,
+      PROMOTION_COMMAND,
+      target.bundle === 'default' ? '' : target.bundle,
+      getDefaultOperatorSourceLabel(),
+    );
+    const result = await this.run(command);
+    if (result.exitCode !== 0) {
+      throw new Error(`Promotion command delivery failed: ${target.seat}.`);
+    }
+  }
+
+  private async run(command: string[]): Promise<CommandResult> {
+    const [executable, ...args] = command;
+    if (executable === undefined) {
+      throw new Error('Promotion transport command is empty.');
+    }
+    return this.options.runner(executable, args);
+  }
+}
+
+async function readPaneEnvironment(pid: number): Promise<string> {
+  return readFile(`/proc/${pid}/environ`, 'utf8');
+}
+
+function parseLeaseSessionId(environment: string): string | null {
+  const value = environment
+    .split('\0')
+    .find((entry) => entry.startsWith('MOSAIC_LEASE_SESSION_ID='))
+    ?.slice('MOSAIC_LEASE_SESSION_ID='.length);
+  return value !== undefined && SESSION_ID_PATTERN.test(value) ? value : null;
+}
@@ -112,13 +112,20 @@ class PromotionHookFixture(unittest.TestCase):
        self.pending_file.write_text(challenge, encoding="utf-8")
        self.pending_file.chmod(0o600)

-    def run_complete(self, runner: mock.Mock) -> tuple[int, str]:
+    def run_complete(
+        self,
+        runner: mock.Mock,
+        now: float | None = None,
+    ) -> tuple[int, str]:
        stderr = io.StringIO()
-        result = self.complete.main(
-            environ=self.environment,
-            stderr=stderr,
-            run=runner,
-        )
+        options: dict[str, object] = {
+            "environ": self.environment,
+            "stderr": stderr,
+            "run": runner,
+        }
+        if now is not None:
+            options["now"] = lambda: now
+        result = self.complete.main(**options)
        return result, stderr.getvalue()


@@ -292,6 +299,62 @@ class PromotionCompleteTest(PromotionHookFixture):
        self.assertFalse(self.pending_file.exists())
        self.assertEqual(runner.call_args.args[0][-2:], ["--complete", CHALLENGE])

+    def test_success_atomically_writes_a_private_correlated_result_with_wall_clock_expiry(self) -> None:
+        self.write_pending()
+        runner = mock.Mock(
+            return_value=self.completed(
+                {"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
+            )
+        )
+
+        with mock.patch.object(self.complete.os, "replace", wraps=os.replace) as replace:
+            result, _stderr = self.run_complete(runner, now=12_345.0)
+
+        result_file = self.pending_dir / "last-result.json"
+        self.assertEqual(result, 0)
+        self.assertEqual(stat.S_IMODE(result_file.stat().st_mode), 0o600)
+        self.assertEqual(
+            json.loads(result_file.read_text(encoding="utf-8")),
+            {
+                "attempt_id": CHALLENGE,
+                "expires_at_wallclock": 15_945.0,
+                "reason": None,
+                "session_id": SESSION_ID,
+                "ts": 12_345.0,
+                "verified": True,
+            },
+        )
+        temporary, destination = replace.call_args.args
+        self.assertRegex(temporary, r"^\.last-result\.json\.tmp-[0-9a-f]+$")
+        self.assertEqual(destination, "last-result.json")
+        self.assertFalse(any(path.name.startswith(".last-result.json.tmp-") for path in self.pending_dir.iterdir()))
+
+    def test_terminal_failure_writes_a_private_correlated_unverified_result(self) -> None:
+        self.write_pending()
+        runner = mock.Mock(
+            return_value=self.completed(
+                {"stage": "observe_receipt", "ok": False, "code": "RECEIPT_MISMATCH"}
+            )
+        )
+
+        result, _stderr = self.run_complete(runner, now=12_345.0)
+
+        result_file = self.pending_dir / "last-result.json"
+        self.assertEqual(result, 0)
+        self.assertFalse(self.pending_file.exists())
+        self.assertEqual(stat.S_IMODE(result_file.stat().st_mode), 0o600)
+        self.assertEqual(
+            json.loads(result_file.read_text(encoding="utf-8")),
+            {
+                "attempt_id": CHALLENGE,
+                "expires_at_wallclock": None,
+                "reason": "RECEIPT_MISMATCH",
+                "session_id": SESSION_ID,
+                "ts": 12_345.0,
+                "verified": False,
+            },
+        )
+
    def test_each_terminal_failure_deletes_pending_file(self) -> None:
        terminal_codes = (
            "RECEIPT_REPLAY",