From a57933a4c4f5738b5d299a4568d8ae76aeee7d99 Mon Sep 17 00:00:00 2001 From: Jason Woltje Date: Sat, 8 Aug 2026 11:59:12 -0500 Subject: [PATCH] fix(mosaic): bound promotion transport delivery --- packages/mosaic/src/commands/promote.spec.ts | 71 +++++++++++++++++++ packages/mosaic/src/commands/promote.ts | 41 ++++++++++- .../src/fleet/promotion-transport.spec.ts | 22 ++++-- .../mosaic/src/fleet/promotion-transport.ts | 56 ++++++++++++--- .../promotion_trigger_unittest.py | 27 +++++++ 5 files changed, 197 insertions(+), 20 deletions(-) diff --git a/packages/mosaic/src/commands/promote.spec.ts b/packages/mosaic/src/commands/promote.spec.ts index 7147e94e..76f754f1 100644 --- a/packages/mosaic/src/commands/promote.spec.ts +++ b/packages/mosaic/src/commands/promote.spec.ts @@ -1,9 +1,12 @@ import { Command } from 'commander'; import { describe, expect, it, vi } from 'vitest'; +import type { FleetRoster } from './fleet.js'; +import { TmuxPromotionTransport } from '../fleet/promotion-transport.js'; import { promoteSeat, registerPromoteCommand, type PromotionBreadcrumbStore, + type PromotionResult, type PromotionTransport, } from './promote.js'; @@ -94,6 +97,46 @@ describe('mosaic promote', () => { expect(result.status).toBe('VERIFIED'); }); + it('returns UNVERIFIED within the command bound when a tmux runner wedges', async () => { + vi.useFakeTimers(); + 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, + }; + const promotionTransport = new TmuxPromotionTransport({ + mosaicHome: '/mosaic', + rosterLoader: async () => roster, + runner: async () => new Promise(() => {}), + }); + const store: PromotionBreadcrumbStore = { + readAttemptId: vi.fn(async () => null), + readResult: vi.fn(async () => null), + }; + + try { + let observedResult: PromotionResult | undefined; + void promoteSeat('claude-seat', { + store, + transport: promotionTransport, + }).then((result) => { + observedResult = result; + }); + await vi.advanceTimersByTimeAsync(5_000); + + expect(observedResult).toMatchObject({ + reason: 'RESOLVE_FAILED: Promotion transport command timed out after 5000ms.', + seat: 'claude-seat', + status: 'UNVERIFIED', + }); + } finally { + vi.useRealTimers(); + } + }); + it('prints VERIFIED with the resolved seat, session, bundle, and wall-clock expiry', async () => { const promotionTransport = transport(); const store: PromotionBreadcrumbStore = { @@ -123,6 +166,34 @@ describe('mosaic promote', () => { } }); + it('prints UNVERIFIED and exits 1 when delivery fails', async () => { + const promotionTransport: PromotionTransport = { + resolve: vi.fn(async () => target), + sendPromotion: vi.fn(async () => { + throw new Error('tmux unavailable'); + }), + }; + const store: PromotionBreadcrumbStore = { + readAttemptId: vi.fn(async () => null), + readResult: vi.fn(async () => null), + }; + const output = vi.spyOn(console, 'log').mockImplementation(() => {}); + const program = new Command().exitOverride(); + registerPromoteCommand(program, { store, transport: promotionTransport }); + + try { + process.exitCode = undefined; + await program.parseAsync(['node', 'mosaic', 'promote', 'claude-seat']); + expect(output).toHaveBeenCalledWith( + `UNVERIFIED seat=claude-seat session=${target.sessionId} bundle=local expiry=none reason=DELIVERY_FAILED: tmux unavailable`, + ); + expect(process.exitCode).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; diff --git a/packages/mosaic/src/commands/promote.ts b/packages/mosaic/src/commands/promote.ts index 33e2afea..304dea48 100644 --- a/packages/mosaic/src/commands/promote.ts +++ b/packages/mosaic/src/commands/promote.ts @@ -15,6 +15,7 @@ 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 SUBPROCESS_TIMEOUT_MS = 4_500; const PENDING_DIRECTORY = 'mosaic-lease'; const RESULT_FILE = 'last-result.json'; @@ -92,7 +93,12 @@ export async function promoteSeat( const sleep = options.sleep ?? defaultSleep; const timeoutMs = normalizeTimeout(options.timeoutMs); const pollIntervalMs = normalizePollInterval(options.pollIntervalMs); - const target = await options.transport.resolve(seat); + let target: PromotionTarget; + try { + target = await options.transport.resolve(seat); + } catch (error: unknown) { + return unverifiedUnresolvedSeat(seat, `RESOLVE_FAILED: ${errorMessage(error)}`); + } const previousAttemptId = await options.store.readAttemptId(target.sessionId); const preSendTimestamp = clock(); try { @@ -109,6 +115,9 @@ export async function promoteSeat( attemptId = currentAttemptId; } if (attemptId !== null || previousAttemptId === null) { + // Completion can consume a first attempt's nonce before this poll observes it. + // In that branch, correlation degrades to session_id + fresh timestamp, which + // is acceptable for this 0600, same-UID local trust boundary. const breadcrumb = await options.store.readResult(); if ( breadcrumb !== null && @@ -233,6 +242,17 @@ function parseOptionTimeout(value: string | undefined): number | undefined { return Number.isFinite(parsed) ? parsed : undefined; } +function unverifiedUnresolvedSeat(seat: string, reason: string): PromotionResult { + return { + bundle: 'unresolved', + expiresAtWallclock: null, + reason, + seat, + sessionId: 'unresolved', + status: 'UNVERIFIED', + }; +} + function unverified(target: PromotionTarget, reason: string): PromotionResult { return { bundle: target.bundle, @@ -268,6 +288,21 @@ function runCommand( const child = spawn(command, args, { stdio: ['ignore', 'pipe', 'pipe'] }); let stdout = ''; let stderr = ''; + let settled = false; + const finish = (result: { exitCode: number; stderr: string; stdout: string }): void => { + if (settled) return; + settled = true; + clearTimeout(timeout); + resolve(result); + }; + const timeout = setTimeout(() => { + child.kill('SIGKILL'); + finish({ + exitCode: 124, + stderr: `Promotion transport subprocess timed out after ${SUBPROCESS_TIMEOUT_MS}ms.`, + stdout, + }); + }, SUBPROCESS_TIMEOUT_MS); child.stdout.on('data', (chunk: Buffer) => { stdout += chunk.toString('utf8'); }); @@ -275,10 +310,10 @@ function runCommand( stderr += chunk.toString('utf8'); }); child.on('error', (error: Error) => { - resolve({ exitCode: 127, stderr: error.message, stdout }); + finish({ exitCode: 127, stderr: error.message, stdout }); }); child.on('close', (code: number | null) => { - resolve({ exitCode: code ?? 1, stderr, stdout }); + finish({ 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 index 35f256cc..c37dcb9c 100644 --- a/packages/mosaic/src/fleet/promotion-transport.spec.ts +++ b/packages/mosaic/src/fleet/promotion-transport.spec.ts @@ -17,10 +17,11 @@ function result(stdout = '', exitCode = 0, stderr = ''): CommandResult { } describe('TmuxPromotionTransport', () => { - it('resolves the exact roster seat and sends through the maintained tmux sender', async () => { + it('resolves the exact roster seat and sends the registered command literally', async () => { const runner = vi .fn() .mockResolvedValueOnce(result('1234 claude 0 0 0 0\n')) + .mockResolvedValueOnce(result()) .mockResolvedValueOnce(result()); const environmentReader = vi.fn(async () => `MOSAIC_LEASE_SESSION_ID=${sessionId}\0`); const transport = new TmuxPromotionTransport({ @@ -39,15 +40,22 @@ describe('TmuxPromotionTransport', () => { sessionId, }); expect(environmentReader).toHaveBeenCalledWith(1234); - expect(runner).toHaveBeenNthCalledWith(2, '/mosaic/tools/tmux/agent-send.sh', [ + expect(runner).toHaveBeenNthCalledWith(2, 'tmux', [ '-L', 'mosaic-fleet', - '-S', - expect.stringMatching(/:operator$/), - '-s', - 'claude-seat', - '-m', + 'send-keys', + '-t', + '=claude-seat:0.0', + '-l', '/mosaic-promote', ]); + expect(runner).toHaveBeenNthCalledWith(3, 'tmux', [ + '-L', + 'mosaic-fleet', + 'send-keys', + '-t', + '=claude-seat:0.0', + 'Enter', + ]); }); }); diff --git a/packages/mosaic/src/fleet/promotion-transport.ts b/packages/mosaic/src/fleet/promotion-transport.ts index e2008a61..4a3f3015 100644 --- a/packages/mosaic/src/fleet/promotion-transport.ts +++ b/packages/mosaic/src/fleet/promotion-transport.ts @@ -1,8 +1,6 @@ import { readFile } from 'node:fs/promises'; import { - buildAgentSendCommand, buildTmuxListPanesCommand, - getDefaultOperatorSourceLabel, getRosterAgent, parseTmuxListPanes, resolveFleetPaths, @@ -10,11 +8,13 @@ import { type CommandRunner, type FleetRoster, RUNTIME_ACCEPTABLE_COMMANDS, + socketArgs, } from '../commands/fleet.js'; import { loadFleetRoster } from './fleet-roster-v1.js'; const PROMOTION_COMMAND = '/mosaic-promote'; const SESSION_ID_PATTERN = /^[a-f0-9]{64}$/; +const TRANSPORT_COMMAND_TIMEOUT_MS = 5_000; export interface PromotionTarget { bundle: string; @@ -81,16 +81,34 @@ export class TmuxPromotionTransport implements PromotionTransport { } async sendPromotion(target: PromotionTarget): Promise { - const command = buildAgentSendCommand( - resolveFleetPaths(this.options.mosaicHome), - target.seat, + const targetPane = `=${target.seat}:0.0`; + const socketName = target.bundle === 'default' ? '' : target.bundle; + // Registered Claude commands must arrive as their exact literal text; the + // fleet agent sender prepends an identity envelope, so it cannot carry this + // command without preventing the UserPromptSubmit matcher from recognizing it. + await this.runPromotionCommand([ + 'tmux', + ...socketArgs(socketName), + 'send-keys', + '-t', + targetPane, + '-l', PROMOTION_COMMAND, - target.bundle === 'default' ? '' : target.bundle, - getDefaultOperatorSourceLabel(), - ); + ]); + await this.runPromotionCommand([ + 'tmux', + ...socketArgs(socketName), + 'send-keys', + '-t', + targetPane, + 'Enter', + ]); + } + + private async runPromotionCommand(command: string[]): Promise { const result = await this.run(command); if (result.exitCode !== 0) { - throw new Error(`Promotion command delivery failed: ${target.seat}.`); + throw new Error('Promotion command delivery failed.'); } } @@ -99,10 +117,28 @@ export class TmuxPromotionTransport implements PromotionTransport { if (executable === undefined) { throw new Error('Promotion transport command is empty.'); } - return this.options.runner(executable, args); + return await withTimeout(this.options.runner(executable, args), TRANSPORT_COMMAND_TIMEOUT_MS); } } +function withTimeout(operation: Promise, timeoutMs: number): Promise { + return new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + reject(new Error(`Promotion transport command timed out after ${timeoutMs}ms.`)); + }, timeoutMs); + void operation.then( + (value) => { + clearTimeout(timeout); + resolve(value); + }, + (error: unknown) => { + clearTimeout(timeout); + reject(error); + }, + ); + }); +} + async function readPaneEnvironment(pid: number): Promise { return readFile(`/proc/${pid}/environ`, 'utf8'); } diff --git a/packages/mosaic/src/lease-broker/promotion_trigger_unittest.py b/packages/mosaic/src/lease-broker/promotion_trigger_unittest.py index 3d954c65..2bc2b9e0 100644 --- a/packages/mosaic/src/lease-broker/promotion_trigger_unittest.py +++ b/packages/mosaic/src/lease-broker/promotion_trigger_unittest.py @@ -410,6 +410,33 @@ class PromotionCompleteTest(PromotionHookFixture): self.assertTrue(self.pending_file.exists()) self.assertIn("ConnectionRefusedError", stderr) + def test_result_write_failure_exits_zero(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, "write_result", side_effect=OSError("disk full")): + result, stderr = self.run_complete(runner) + + self.assertEqual(result, 0) + self.assertIn("OSError", stderr) + + def test_missing_lease_session_id_exits_zero(self) -> None: + self.write_pending() + environment = dict(self.environment) + del environment["MOSAIC_LEASE_SESSION_ID"] + runner = mock.Mock() + stderr = io.StringIO() + + result = self.complete.main(environ=environment, stderr=stderr, run=runner) + + self.assertEqual(result, 0) + runner.assert_not_called() + self.assertIn("KeyError", stderr.getvalue()) + def test_insecure_runtime_directory_mode_preserves_pending(self) -> None: self.write_pending(CHALLENGE) self.runtime_dir.chmod(0o755)