fix(mosaic): bound promotion transport delivery

This commit is contained in:
Jason Woltje
2026-08-08 16:05:04 -05:00
parent 9d0822570c
commit a57933a4c4
5 changed files with 197 additions and 20 deletions
@@ -1,9 +1,12 @@
import { Command } from 'commander'; import { Command } from 'commander';
import { describe, expect, it, vi } from 'vitest'; import { describe, expect, it, vi } from 'vitest';
import type { FleetRoster } from './fleet.js';
import { TmuxPromotionTransport } from '../fleet/promotion-transport.js';
import { import {
promoteSeat, promoteSeat,
registerPromoteCommand, registerPromoteCommand,
type PromotionBreadcrumbStore, type PromotionBreadcrumbStore,
type PromotionResult,
type PromotionTransport, type PromotionTransport,
} from './promote.js'; } from './promote.js';
@@ -94,6 +97,46 @@ describe('mosaic promote', () => {
expect(result.status).toBe('VERIFIED'); 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 () => { it('prints VERIFIED with the resolved seat, session, bundle, and wall-clock expiry', async () => {
const promotionTransport = transport(); const promotionTransport = transport();
const store: PromotionBreadcrumbStore = { 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 () => { it('rejects a stale result even when its nonce matches', async () => {
const promotionTransport = transport(); const promotionTransport = transport();
let now = 1_000; let now = 1_000;
+38 -3
View File
@@ -15,6 +15,7 @@ import { resolveFleetPaths, type CommandRunner } from './fleet.js';
const ATTEMPT_ID_PATTERN = /^[a-f0-9]{64}$/; const ATTEMPT_ID_PATTERN = /^[a-f0-9]{64}$/;
const DEFAULT_POLL_INTERVAL_MS = 250; const DEFAULT_POLL_INTERVAL_MS = 250;
const DEFAULT_TIMEOUT_MS = 30_000; const DEFAULT_TIMEOUT_MS = 30_000;
const SUBPROCESS_TIMEOUT_MS = 4_500;
const PENDING_DIRECTORY = 'mosaic-lease'; const PENDING_DIRECTORY = 'mosaic-lease';
const RESULT_FILE = 'last-result.json'; const RESULT_FILE = 'last-result.json';
@@ -92,7 +93,12 @@ export async function promoteSeat(
const sleep = options.sleep ?? defaultSleep; const sleep = options.sleep ?? defaultSleep;
const timeoutMs = normalizeTimeout(options.timeoutMs); const timeoutMs = normalizeTimeout(options.timeoutMs);
const pollIntervalMs = normalizePollInterval(options.pollIntervalMs); 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 previousAttemptId = await options.store.readAttemptId(target.sessionId);
const preSendTimestamp = clock(); const preSendTimestamp = clock();
try { try {
@@ -109,6 +115,9 @@ export async function promoteSeat(
attemptId = currentAttemptId; attemptId = currentAttemptId;
} }
if (attemptId !== null || previousAttemptId === null) { 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(); const breadcrumb = await options.store.readResult();
if ( if (
breadcrumb !== null && breadcrumb !== null &&
@@ -233,6 +242,17 @@ function parseOptionTimeout(value: string | undefined): number | undefined {
return Number.isFinite(parsed) ? parsed : 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 { function unverified(target: PromotionTarget, reason: string): PromotionResult {
return { return {
bundle: target.bundle, bundle: target.bundle,
@@ -268,6 +288,21 @@ function runCommand(
const child = spawn(command, args, { stdio: ['ignore', 'pipe', 'pipe'] }); const child = spawn(command, args, { stdio: ['ignore', 'pipe', 'pipe'] });
let stdout = ''; let stdout = '';
let stderr = ''; 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) => { child.stdout.on('data', (chunk: Buffer) => {
stdout += chunk.toString('utf8'); stdout += chunk.toString('utf8');
}); });
@@ -275,10 +310,10 @@ function runCommand(
stderr += chunk.toString('utf8'); stderr += chunk.toString('utf8');
}); });
child.on('error', (error: Error) => { 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) => { child.on('close', (code: number | null) => {
resolve({ exitCode: code ?? 1, stderr, stdout }); finish({ exitCode: code ?? 1, stderr, stdout });
}); });
}); });
} }
@@ -17,10 +17,11 @@ function result(stdout = '', exitCode = 0, stderr = ''): CommandResult {
} }
describe('TmuxPromotionTransport', () => { 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 const runner = vi
.fn<CommandRunner>() .fn<CommandRunner>()
.mockResolvedValueOnce(result('1234 claude 0 0 0 0\n')) .mockResolvedValueOnce(result('1234 claude 0 0 0 0\n'))
.mockResolvedValueOnce(result())
.mockResolvedValueOnce(result()); .mockResolvedValueOnce(result());
const environmentReader = vi.fn(async () => `MOSAIC_LEASE_SESSION_ID=${sessionId}\0`); const environmentReader = vi.fn(async () => `MOSAIC_LEASE_SESSION_ID=${sessionId}\0`);
const transport = new TmuxPromotionTransport({ const transport = new TmuxPromotionTransport({
@@ -39,15 +40,22 @@ describe('TmuxPromotionTransport', () => {
sessionId, sessionId,
}); });
expect(environmentReader).toHaveBeenCalledWith(1234); expect(environmentReader).toHaveBeenCalledWith(1234);
expect(runner).toHaveBeenNthCalledWith(2, '/mosaic/tools/tmux/agent-send.sh', [ expect(runner).toHaveBeenNthCalledWith(2, 'tmux', [
'-L', '-L',
'mosaic-fleet', 'mosaic-fleet',
'-S', 'send-keys',
expect.stringMatching(/:operator$/), '-t',
'-s', '=claude-seat:0.0',
'claude-seat', '-l',
'-m',
'/mosaic-promote', '/mosaic-promote',
]); ]);
expect(runner).toHaveBeenNthCalledWith(3, 'tmux', [
'-L',
'mosaic-fleet',
'send-keys',
'-t',
'=claude-seat:0.0',
'Enter',
]);
}); });
}); });
@@ -1,8 +1,6 @@
import { readFile } from 'node:fs/promises'; import { readFile } from 'node:fs/promises';
import { import {
buildAgentSendCommand,
buildTmuxListPanesCommand, buildTmuxListPanesCommand,
getDefaultOperatorSourceLabel,
getRosterAgent, getRosterAgent,
parseTmuxListPanes, parseTmuxListPanes,
resolveFleetPaths, resolveFleetPaths,
@@ -10,11 +8,13 @@ import {
type CommandRunner, type CommandRunner,
type FleetRoster, type FleetRoster,
RUNTIME_ACCEPTABLE_COMMANDS, RUNTIME_ACCEPTABLE_COMMANDS,
socketArgs,
} from '../commands/fleet.js'; } from '../commands/fleet.js';
import { loadFleetRoster } from './fleet-roster-v1.js'; import { loadFleetRoster } from './fleet-roster-v1.js';
const PROMOTION_COMMAND = '/mosaic-promote'; const PROMOTION_COMMAND = '/mosaic-promote';
const SESSION_ID_PATTERN = /^[a-f0-9]{64}$/; const SESSION_ID_PATTERN = /^[a-f0-9]{64}$/;
const TRANSPORT_COMMAND_TIMEOUT_MS = 5_000;
export interface PromotionTarget { export interface PromotionTarget {
bundle: string; bundle: string;
@@ -81,16 +81,34 @@ export class TmuxPromotionTransport implements PromotionTransport {
} }
async sendPromotion(target: PromotionTarget): Promise<void> { async sendPromotion(target: PromotionTarget): Promise<void> {
const command = buildAgentSendCommand( const targetPane = `=${target.seat}:0.0`;
resolveFleetPaths(this.options.mosaicHome), const socketName = target.bundle === 'default' ? '' : target.bundle;
target.seat, // 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, 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<void> {
const result = await this.run(command); const result = await this.run(command);
if (result.exitCode !== 0) { 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) { if (executable === undefined) {
throw new Error('Promotion transport command is empty.'); 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<T>(operation: Promise<T>, timeoutMs: number): Promise<T> {
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<string> { async function readPaneEnvironment(pid: number): Promise<string> {
return readFile(`/proc/${pid}/environ`, 'utf8'); return readFile(`/proc/${pid}/environ`, 'utf8');
} }
@@ -410,6 +410,33 @@ class PromotionCompleteTest(PromotionHookFixture):
self.assertTrue(self.pending_file.exists()) self.assertTrue(self.pending_file.exists())
self.assertIn("ConnectionRefusedError", stderr) 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: def test_insecure_runtime_directory_mode_preserves_pending(self) -> None:
self.write_pending(CHALLENGE) self.write_pending(CHALLENGE)
self.runtime_dir.chmod(0o755) self.runtime_dir.chmod(0o755)