diff --git a/packages/mosaic/framework/tools/lease-broker/promote-complete.py b/packages/mosaic/framework/tools/lease-broker/promote-complete.py index 680d798a..64003988 100644 --- a/packages/mosaic/framework/tools/lease-broker/promote-complete.py +++ b/packages/mosaic/framework/tools/lease-broker/promote-complete.py @@ -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, diff --git a/packages/mosaic/src/cli.ts b/packages/mosaic/src/cli.ts index 958cb3d0..f38e64a7 100644 --- a/packages/mosaic/src/cli.ts +++ b/packages/mosaic/src/cli.ts @@ -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 ──────────────────────────────────────────────────────────── diff --git a/packages/mosaic/src/commands/promote.spec.ts b/packages/mosaic/src/commands/promote.spec.ts new file mode 100644 index 00000000..7147e94e --- /dev/null +++ b/packages/mosaic/src/commands/promote.spec.ts @@ -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); + }); +}); diff --git a/packages/mosaic/src/commands/promote.ts b/packages/mosaic/src/commands/promote.ts new file mode 100644 index 00000000..33e2afea --- /dev/null +++ b/packages/mosaic/src/commands/promote.ts @@ -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; + readResult(): Promise; +} + +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; + 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 { + 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 { + 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 { + 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 ') + .description('Promote a Claude fleet seat and report the correlated lease result') + .option( + '--timeout ', + `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 { + let handle: Awaited>; + 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 { + 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 { + 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 }); + }); + }); +} diff --git a/packages/mosaic/src/fleet/promotion-transport.spec.ts b/packages/mosaic/src/fleet/promotion-transport.spec.ts new file mode 100644 index 00000000..35f256cc --- /dev/null +++ b/packages/mosaic/src/fleet/promotion-transport.spec.ts @@ -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() + .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', + ]); + }); +}); diff --git a/packages/mosaic/src/fleet/promotion-transport.ts b/packages/mosaic/src/fleet/promotion-transport.ts new file mode 100644 index 00000000..e2008a61 --- /dev/null +++ b/packages/mosaic/src/fleet/promotion-transport.ts @@ -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; + sendPromotion(target: PromotionTarget): Promise; +} + +export interface TmuxPromotionTransportOptions { + environmentReader?: (pid: number) => Promise; + mosaicHome: string; + rosterLoader?: () => Promise; + runner: CommandRunner; +} + +/** Local, roster-bound transport for the in-seat promotion command. */ +export class TmuxPromotionTransport implements PromotionTransport { + private readonly environmentReader: (pid: number) => Promise; + private readonly rosterLoader: () => Promise; + + constructor(private readonly options: TmuxPromotionTransportOptions) { + this.environmentReader = options.environmentReader ?? readPaneEnvironment; + this.rosterLoader = + options.rosterLoader ?? + (() => loadFleetRoster(resolveFleetPaths(options.mosaicHome).rosterPath)); + } + + async resolve(seat: string): Promise { + 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 { + 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 { + 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 { + 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; +} diff --git a/packages/mosaic/src/lease-broker/promotion_trigger_unittest.py b/packages/mosaic/src/lease-broker/promotion_trigger_unittest.py index dea87f37..3d954c65 100644 --- a/packages/mosaic/src/lease-broker/promotion_trigger_unittest.py +++ b/packages/mosaic/src/lease-broker/promotion_trigger_unittest.py @@ -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",