diff --git a/src/cli.ts b/src/cli.ts index be02fbf..e65a10d 100755 --- a/src/cli.ts +++ b/src/cli.ts @@ -32,6 +32,7 @@ import { RIG_VERSION_HELP, rigVersionCommand } from './rig-version.ts'; import { DERIVED_HELP, derivedCommand } from './prepared-inputs.ts'; import { SPLIT_CAPTIONS_HELP, splitCaptionsCommand } from './split-captions.ts'; import { VERSION_HELP, versionCommand } from './config-lost.ts'; +import { takePublishedLines } from './job-writer.ts'; import { NEXT_HELP, nextCommand } from './next.ts'; // Input registration is kept in its own CLI block. @@ -293,9 +294,28 @@ export function main(argv: string[], now: number): number { if (positional.length > 1 || action === 'open' && (!options['--reviewer'] || !['owner', 'delegate'].includes(options['--role'] ?? ''))) { refuse('JOB_BINDING_INVALID', 'project.json: job open needs --reviewer --role owner|delegate and one optional directory', USAGE_EXIT); } - const child = Bun.spawnSync([process.execPath, join(import.meta.dir, 'job-runtime.ts'), action!], { - stdin: Buffer.from(JSON.stringify({ root: positional[0] ?? process.cwd(), reviewer: options['--reviewer'], role: options['--role'], draft: options['--draft'], now })), - stdout: 'inherit', stderr: 'inherit', + const stdin = Buffer.from(JSON.stringify({ root: positional[0] ?? process.cwd(), reviewer: options['--reviewer'], role: options['--role'], draft: options['--draft'], now })); + const runtime = join(import.meta.dir, 'job-runtime.ts'); + if (action === 'open') { + const child = Bun.spawn([process.execPath, runtime, action], { + stdin, stdout: 'inherit', stderr: 'inherit', + }); + const forward = (signal: 'SIGINT' | 'SIGTERM') => { + if (child.exitCode === null && child.signalCode === null) child.kill(signal); + }; + const onSigint = () => forward('SIGINT'); + const onSigterm = () => forward('SIGTERM'); + process.on('SIGINT', onSigint); + process.on('SIGTERM', onSigterm); + void child.exited.then(code => { + process.off('SIGINT', onSigint); + process.off('SIGTERM', onSigterm); + process.exitCode = code; + }); + return 0; + } + const child = Bun.spawnSync([process.execPath, runtime, action!], { + stdin, stdout: 'inherit', stderr: 'inherit', }); return child.exitCode; } @@ -311,15 +331,17 @@ export function main(argv: string[], now: number): number { if (command === undefined) { refuse('UNKNOWN_COMMAND', `no command named ${first} (see rig-create --help)`, USAGE_EXIT); } + takePublishedLines(); const result = command(rest, now); - if (typeof result === 'string') { - process.stdout.write(`${result}\n`); - return 0; - } - process.stdout.write(`${result.stdout}\n`); - return result.exit; + const jobLines = takePublishedLines(); + const text = typeof result === 'string' ? result : result.stdout; + const output = !rest.includes('--json') && jobLines.length ? `${text}\n${jobLines.join('\n')}` : text; + process.stdout.write(`${output}\n`); + return typeof result === 'string' ? 0 : result.exit; } catch (error) { + const jobLines = takePublishedLines(); if (error instanceof Refusal) { + if (!rest.includes('--json') && jobLines.length) process.stdout.write(`${jobLines.join('\n')}\n`); process.stderr.write(`${formatRefusal(error)}\n`); return error.exitCode; } diff --git a/src/job-host.ts b/src/job-host.ts index bae805c..33acf6c 100644 --- a/src/job-host.ts +++ b/src/job-host.ts @@ -259,6 +259,16 @@ export function subjectLabel(subject: JobSubject): string { } } +/** Format `asked` and `already` outcome lines for `job open` and bound writers. */ +export function askOutcomeLines(outcomes: readonly AskOutcome[]): string[] { + const lines: string[] = []; + for (const status of ['asked', 'already'] as const) { + const named = outcomes.filter(outcome => outcome.status === status).map(outcome => subjectLabel(outcome.subject)); + if (named.length) lines.push(`job: ${status === 'asked' ? 'asked' : 'already asked'} ${named.join(', ')}`); + } + return lines; +} + /** Name every ask Review refused. The publications themselves are acknowledged. */ export function refuseAskOutcomes(outcomes: readonly AskOutcome[]): void { const refused = outcomes.filter(outcome => outcome.status === 'refused'); @@ -369,11 +379,7 @@ export async function openDraftPublication(root: string, host: JobOwnerHost, now selector = result.selector; published = true; } - const lines = [`job: ${selectorLabel(selector)} ${published ? 'published' : 'already published'}`]; - for (const status of ['asked', 'already'] as const) { - const named = outcomes.filter(outcome => outcome.status === status).map(outcome => subjectLabel(outcome.subject)); - if (named.length) lines.push(`job: ${status === 'asked' ? 'asked' : 'already asked'} ${named.join(', ')}`); - } + const lines = [`job: ${selectorLabel(selector)} ${published ? 'published' : 'already published'}`, ...askOutcomeLines(outcomes)]; refuseAskOutcomes(outcomes); return lines; } diff --git a/src/job-runtime.ts b/src/job-runtime.ts index 488d79a..3a77b49 100644 --- a/src/job-runtime.ts +++ b/src/job-runtime.ts @@ -1,7 +1,7 @@ /** Private stdin transport for synchronous writers. Capabilities never enter argv. */ import { readFileSync } from 'node:fs'; import { resolve } from 'node:path'; -import { assertRequestedDraft, closeCreateJob, openCreateJob, openDraftPublication, publishPending, reconcileCreateJob, refuseAskOutcomes, type AskOutcome } from './job-host.ts'; +import { askOutcomeLines, assertRequestedDraft, closeCreateJob, openCreateJob, openDraftPublication, publishPending, reconcileCreateJob, refuseAskOutcomes, type AskOutcome } from './job-host.ts'; import { failPublication, pendingPublications, reconcilePublication } from './job-outbox.ts'; import { Refusal, refuse } from './refusal.ts'; @@ -20,6 +20,8 @@ async function run(): Promise { failPublication(root, entry, failure); throw new Refusal('JOB_PUBLISH_PENDING', 'project.json: committed publication is pending; run rig-create job reconcile'); } + const lines = askOutcomeLines(outcomes); + if (lines.length) console.log(lines.join('\n')); refuseAskOutcomes(outcomes); break; } @@ -28,23 +30,50 @@ async function run(): Promise { case 'open': { if (!input.reviewer || !input.role) refuse('JOB_BINDING_INVALID', 'project.json: job open needs --reviewer and --role owner|delegate'); assertRequestedDraft(root, input.draft); + let stopping = false; + let stopHost: (() => void) | null = null; + const onSignal = () => { + stopping = true; + stopHost?.(); + }; + process.on('SIGINT', onSignal); + process.on('SIGTERM', onSignal); const host = await openCreateJob(root, { id: input.reviewer, role: input.role }, input.now); + if (stopping) { + process.off('SIGINT', onSignal); + process.off('SIGTERM', onSignal); + host.stop(); + break; + } console.log(host.ownerUrl); // The page shows the record's current draft and its asks. A refused publish or ask is // named here; a foreground host keeps serving so `job reconcile` and `job open` can retry. try { console.log((await openDraftPublication(root, host, input.now, input.draft)).join('\n')); } catch (error) { - if (host.reused) { host.stop(); throw error; } + if (host.reused) { + process.off('SIGINT', onSignal); + process.off('SIGTERM', onSignal); + host.stop(); + throw error; + } console.error(error instanceof Refusal ? `${error.code}: ${error.message}` : 'JOB_PUBLISH_PENDING: project.json: draft publication failed; run rig-create job reconcile'); process.exitCode = 1; } - if (host.reused) host.stop(); - else { - const stop = () => { host.stop(); process.exit(0); }; - process.once('SIGINT', stop); process.once('SIGTERM', stop); + if (host.reused || stopping) { + process.off('SIGINT', onSignal); + process.off('SIGTERM', onSignal); + host.stop(); + } else { const timer = setInterval(() => { - if (host.session.status().status === 'closed') { clearInterval(timer); stop(); } + if (host.session.status().status === 'closed') stopHost?.(); }, 1000); + stopHost = () => { + clearInterval(timer); + process.off('SIGINT', onSignal); + process.off('SIGTERM', onSignal); + host.stop(); + process.exit(process.exitCode ?? 0); + }; } break; } diff --git a/src/job-writer.ts b/src/job-writer.ts index 679491e..b89a330 100644 --- a/src/job-writer.ts +++ b/src/job-writer.ts @@ -22,10 +22,16 @@ export function prepareVersionIntent(root: string, bytes: Buffer, n: number): Pe return enqueuePublication(root, binding, { path: `review-live-v${n}.json`, selector: publicationSelector(project, n) }, bytes); } +const publishedLines: string[] = []; + +export function takePublishedLines(): string[] { + return publishedLines.splice(0); +} + /** Called by the committing writer, before its own locks are released. */ -export function publishCommittedVersion(root: string, n: number | string): void { +export function publishCommittedVersion(root: string, n: number | string): string[] { const binding = readWriterBinding(root); - if (!binding) return; + if (!binding) return []; const project = JSON.parse(readFileSync(join(root, 'project.json'), 'utf8')) as JsonObject; const result = typeof n === 'string' ? draftPublicationResult(n) : { path: `review-live-v${n}.json`, selector: publicationSelector(project, n) }; const entry = enqueuePublication(root, binding, result); @@ -41,20 +47,27 @@ export function publishCommittedVersion(root: string, n: number | string): void } // A separate Bun process runs asynchronous Review while this synchronous writer // keeps its existing call contract and retains both write locks. - let exit: number; + let child: ReturnType>; try { - exit = Bun.spawnSync([process.execPath, fileURLToPath(new URL('./job-runtime.ts', import.meta.url)), 'publish'], { + child = Bun.spawnSync([process.execPath, fileURLToPath(new URL('./job-runtime.ts', import.meta.url)), 'publish'], { stdin: Buffer.from(JSON.stringify({ root, id: entry.id })), stdout: 'pipe', stderr: 'pipe', - }).exitCode; + }); } catch { refuse('JOB_PUBLISH_PENDING', `project.json: record ${entry.recordHash} committed but the publication process could not start; restore the runtime and run rig-create job reconcile`); } - if (exit !== 0) { + const lines = child.stdout.toString('utf8').split('\n').map(line => line.trim()).filter(Boolean); + publishedLines.push(...lines); + if (child.exitCode !== 0) { const path = join(jobPrivateRoot(root), 'outbox', `${entry.id}.json`); - if (!existsSync(path)) return; // The child acknowledged before its abnormal exit. + if (!existsSync(path)) { + const match = /^([A-Z][A-Z0-9_]*): ([\s\S]+)$/.exec(child.stderr.toString('utf8').trim()); + if (match) refuse(match[1]!, match[2]!); + return lines; // The child acknowledged before its abnormal exit. + } const pending = JSON.parse(readFileSync(path, 'utf8')) as PendingPublication; refuse('JOB_PUBLISH_PENDING', `project.json: committed record ${entry.recordHash} awaits publication${pending.failure ? ` (${pending.failure.code}: ${pending.failure.message})` : ''}; run rig-create job reconcile`); } + return lines; } export function draftPublicationResult(id: string): { path: string; selector: { kind: 'draft'; id: string } } { diff --git a/tests/job-signal-ask.test.ts b/tests/job-signal-ask.test.ts new file mode 100644 index 0000000..2c2b9fc --- /dev/null +++ b/tests/job-signal-ask.test.ts @@ -0,0 +1,258 @@ +/** + * `job open` forwards SIGTERM/SIGINT from the CLI wrapper to its spawned runtime so the + * host port stops without leaving an orphan listener, and bound writers surface refused ask + * outcomes (`JOB_ASK_REFUSED`) after acknowledging the publication (#355). + */ +import { Database } from 'bun:sqlite'; +import { afterEach, expect, test } from 'bun:test'; +import { cpSync, mkdtempSync, readFileSync, realpathSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { EXAMPLE_NOW } from '../scripts/example-time.ts'; +import { draftRigCommand } from '../src/drafts.ts'; +import { askOutcomeLines, closeCreateJob } from '../src/job-host.ts'; +import { jobHostStateRoot, jobPrivateRoot, pendingPublications, readAcknowledgedPublication, readWriterBinding } from '../src/job-outbox.ts'; +import { advanceCommand } from '../src/progress-writers.ts'; +import { Refusal } from '../src/refusal.ts'; +import type { JsonObject } from '../src/templates.ts'; + +const CLI = join(import.meta.dir, '../src/cli.ts'); +const roots: string[] = []; +const opened: { proc: Bun.Subprocess; root: string; origin?: string }[] = []; + +async function isOriginListening(origin: string): Promise { + try { + const response = await fetch(origin, { signal: AbortSignal.timeout(300) }); + await response.body?.cancel(); + return true; + } catch { + return false; + } +} + +afterEach(async () => { + for (const { proc, root, origin } of opened.splice(0)) { + if (proc.exitCode === null && proc.signalCode === null) { + proc.kill('SIGTERM'); + } + await proc.exited; + if (readWriterBinding(root) && pendingPublications(root).length === 0) { + try { await closeCreateJob(root); } catch { /* Already closed or unreachable. */ } + } + if (origin) { + const deadline = Date.now() + 2500; + while (Date.now() < deadline && await isOriginListening(origin)) { + await Bun.sleep(100); + } + } + } + for (const root of roots.splice(0)) { + rmSync(jobPrivateRoot(root), { recursive: true, force: true }); + rmSync(jobHostStateRoot(root), { recursive: true, force: true }); + rmSync(root, { recursive: true, force: true }); + } +}); + +const read = (root: string): JsonObject => JSON.parse(readFileSync(join(root, 'project.json'), 'utf8')) as JsonObject; +const write = (root: string, project: JsonObject): void => writeFileSync(join(root, 'project.json'), `${JSON.stringify(project, null, 2)}\n`); + +function copyCheckpoint(checkpoint: string, edit?: (project: JsonObject) => void): string { + const root = realpathSync(mkdtempSync(join(tmpdir(), 'create-job-signal-ask-'))); + roots.push(root); + cpSync(join(import.meta.dir, '../templates/example-chain/checkpoints', checkpoint), root, { recursive: true }); + if (edit) { + const project = read(root); + edit(project); + write(root, project); + } + return root; +} + +async function cli(root: string, args: string[]): Promise<{ stdout: string; stderr: string; exit: number }> { + const child = Bun.spawn([process.execPath, CLI, ...args], { cwd: root, stdout: 'pipe', stderr: 'pipe' }); + const [stdout, stderr, exit] = await Promise.all([new Response(child.stdout).text(), new Response(child.stderr).text(), child.exited]); + return { stdout, stderr, exit }; +} + +async function open(root: string): Promise<{ proc: Bun.Subprocess; url: URL; lines: string[] }> { + const proc = Bun.spawn([process.execPath, CLI, 'job', 'open', root, '--reviewer', 'synthetic-owner', '--role', 'owner'], { + cwd: root, + stdout: 'pipe', + stderr: 'pipe', + }); + const entry: { proc: Bun.Subprocess; root: string; origin?: string } = { proc, root }; + opened.push(entry); + const reader = proc.stdout.getReader(); + const decoder = new TextDecoder(); + let text = ''; + while (!/\njob: .*\n$/.test(text)) { + const chunk = await reader.read(); + if (chunk.done) break; + text += decoder.decode(chunk.value); + } + reader.releaseLock(); + const lines = text.trim().split('\n'); + const url = new URL(lines[0]!); + entry.origin = url.origin; + return { proc, url, lines: lines.slice(1) }; +} + +test.each(['SIGTERM', 'SIGINT'] as const)('job open CLI forwards %s to its runtime and stops its own host port without leaving an orphan listener', async signal => { + const root = copyCheckpoint('1-parts-verdict'); + const { proc, url, lines } = await open(root); + expect(lines).toEqual(['job: draft chain published', 'job: asked parts-verdict chain, checkpoint assemble']); + expect(await isOriginListening(url.origin)).toBe(true); + + proc.kill(signal); + await proc.exited; + + // Give the runtime a brief moment to complete stop() if exiting asynchronously. + const deadline = Date.now() + 1000; + while (Date.now() < deadline && await isOriginListening(url.origin)) { + await Bun.sleep(50); + } + expect(await isOriginListening(url.origin)).toBe(false); +}); + +test('bound writers surface JOB_ASK_REFUSED by named code when Review refuses a declared ask after acknowledging the publication', async () => { + const root = copyCheckpoint('3-rig-review', project => { + project.stage = 'rigging'; + project.advances = (project.advances as JsonObject[]).filter(entry => entry.stage !== 'rig-review'); + }); + const opening = await open(root); + const binding = readWriterBinding(root)!; + + // Plant a one-shot SQLite trigger in the synthetic host state so the published snapshot's + // retained subjects drop `signoff rig-review` after `publish` succeeds, causing Review's + // subsequent `ask` for `signoff rig-review` to refuse with `JOB_ASK_INVALID`. + const sqlitePath = join(binding.identity.stateDir, 'jobs.sqlite'); + const db = new Database(sqlitePath); + try { + db.exec(` + CREATE TRIGGER refuse_signoff_ask AFTER UPDATE OF document ON jobs + WHEN json_extract(NEW.document, '$.head') IS NOT NULL + AND json_extract(OLD.document, '$.head') IS NOT json_extract(NEW.document, '$.head') + BEGIN + UPDATE jobs + SET document = json_remove(NEW.document, '$.publications.' || json_extract(NEW.document, '$.head') || '.retained.subjects[1]') + WHERE job_id = NEW.job_id; + END; + `); + } finally { + db.close(); + } + + let thrown: unknown; + try { + advanceCommand(['rig-review', root, '--json'], EXAMPLE_NOW); + } catch (error) { + thrown = error; + } + expect(thrown).toBeInstanceOf(Refusal); + expect((thrown as Refusal).code).toBe('JOB_ASK_REFUSED'); + expect((thrown as Refusal).message).toContain('signoff rig-review (JOB_ASK_INVALID)'); + + // The record is committed and the publication is acknowledged (never left pending). + expect(read(root).stage).toBe('rig-review'); + expect(pendingPublications(root)).toEqual([]); + const headEvents = await fetch(`${opening.url.origin}/jobs/${binding.jobId}/events`, { + headers: { origin: opening.url.origin, 'X-Review-Capability': opening.url.hash.replace('#cap=', '') }, + signal: AbortSignal.timeout(1000), + }); + const reader = headEvents.body!.getReader(); + const firstChunk = await reader.read(); + await reader.cancel(); + const lines = new TextDecoder().decode(firstChunk.value).trim().split('\n').map(line => JSON.parse(line) as JsonObject); + const published = lines.filter(event => event.type === 'published').at(-1)!.publication as JsonObject; + const acknowledged = readAcknowledgedPublication(root, binding.jobId, String(published.publicationId)); + expect(acknowledged).not.toBeNull(); + expect(acknowledged?.subjects).toEqual([ + { kind: 'review-verdict', stage: 'rig-review' }, + { kind: 'signoff', stage: 'rig-review' }, + ]); + + // Also verify via draftRigCommand and the CLI entrypoint on a bound draft-rig writer that JOB_ASK_REFUSED exits 1 with the named code. + const draftRoot = copyCheckpoint('1-parts-verdict'); + cpSync(join(draftRoot, 'derived/drafts/chain'), join(draftRoot, 'derived/drafts/chain-2'), { recursive: true }); + cpSync(join(draftRoot, 'derived/drafts/chain'), join(draftRoot, 'derived/drafts/chain-3'), { recursive: true }); + await open(draftRoot); + const draftBinding = readWriterBinding(draftRoot)!; + const draftDb = new Database(join(draftBinding.identity.stateDir, 'jobs.sqlite')); + try { + draftDb.exec(` + CREATE TRIGGER refuse_draft_ask AFTER UPDATE OF document ON jobs + WHEN json_extract(NEW.document, '$.head') IS NOT NULL + AND json_extract(OLD.document, '$.head') IS NOT json_extract(NEW.document, '$.head') + BEGIN + UPDATE jobs + SET document = json_remove(NEW.document, '$.publications.' || json_extract(NEW.document, '$.head') || '.retained.subjects[0]') + WHERE job_id = NEW.job_id; + END; + `); + } finally { + draftDb.close(); + } + expect(() => + draftRigCommand(['chain-2', draftRoot, '--config', 'derived/drafts/chain-2/config.json', '--from', 'derived/drafts/chain-2/out', '--parts', 'derived/drafts/chain-2/input'], EXAMPLE_NOW), + ).toThrow(expect.objectContaining({ code: 'JOB_ASK_REFUSED' })); + const cliDraft = await cli(draftRoot, [ + 'draft-rig', 'chain-3', draftRoot, + '--config', 'derived/drafts/chain-3/config.json', + '--from', 'derived/drafts/chain-3/out', + '--parts', 'derived/drafts/chain-3/input', + ]); + expect(cliDraft.exit).toBe(1); + expect(cliDraft.stdout.trim()).toBe('job: asked checkpoint assemble'); + expect(cliDraft.stderr).toContain('JOB_ASK_REFUSED: '); + expect(cliDraft.stderr).toContain('parts-verdict chain-3 (JOB_ASK_INVALID)'); + expect(pendingPublications(draftRoot)).toEqual([]); +}); + +test('bound writers print concise asked/already subject lines in text mode and preserve JSON stdout in --json mode', async () => { + expect(askOutcomeLines([ + { subject: { kind: 'parts-verdict', draftId: 'chain-2' }, status: 'asked' }, + { subject: { kind: 'checkpoint', checkpointId: 'assemble' }, status: 'already' }, + ])).toEqual([ + 'job: asked parts-verdict chain-2', + 'job: already asked checkpoint assemble', + ]); + + const root = copyCheckpoint('3-rig-review', project => { + project.stage = 'rigging'; + project.advances = (project.advances as JsonObject[]).filter(entry => entry.stage !== 'rig-review'); + }); + await open(root); + const advanced = await cli(root, ['advance', 'rig-review', root]); + expect({ exit: advanced.exit, stderr: advanced.stderr }).toEqual({ exit: 0, stderr: '' }); + expect(advanced.stdout.trim().split('\n')).toEqual([ + 'advance: rigging -> rig-review', + 'job: asked review-verdict rig-review, signoff rig-review', + ]); + + const draftRoot = copyCheckpoint('1-parts-verdict'); + cpSync(join(draftRoot, 'derived/drafts/chain'), join(draftRoot, 'derived/drafts/chain-2'), { recursive: true }); + cpSync(join(draftRoot, 'derived/drafts/chain'), join(draftRoot, 'derived/drafts/chain-3'), { recursive: true }); + await open(draftRoot); + + const drafted = await cli(draftRoot, [ + 'draft-rig', 'chain-2', draftRoot, + '--config', 'derived/drafts/chain-2/config.json', + '--from', 'derived/drafts/chain-2/out', + '--parts', 'derived/drafts/chain-2/input', + ]); + expect({ exit: drafted.exit, stderr: drafted.stderr }).toEqual({ exit: 0, stderr: '' }); + expect(drafted.stdout.trim().split('\n')).toEqual([ + 'draft-rig: recorded chain-2; no production version created', + 'job: asked parts-verdict chain-2, checkpoint assemble', + ]); + + const draftedJson = await cli(draftRoot, [ + 'draft-rig', 'chain-3', draftRoot, + '--config', 'derived/drafts/chain-3/config.json', + '--from', 'derived/drafts/chain-3/out', + '--parts', 'derived/drafts/chain-3/input', + '--json', + ]); + expect({ exit: draftedJson.exit, stderr: draftedJson.stderr }).toEqual({ exit: 0, stderr: '' }); + expect(JSON.parse(draftedJson.stdout)).toMatchObject({ id: 'chain-3' }); +});