Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 31 additions & 9 deletions src/cli.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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 <id> --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;
}
Expand All @@ -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;
}
Expand Down
16 changes: 11 additions & 5 deletions src/job-host.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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');
Expand Down Expand Up @@ -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;
}
Expand Down
43 changes: 36 additions & 7 deletions src/job-runtime.ts
Original file line number Diff line number Diff line change
@@ -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';

Expand All @@ -20,6 +20,8 @@ async function run(): Promise<void> {
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;
}
Expand All @@ -28,23 +30,50 @@ async function run(): Promise<void> {
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;
}
Expand Down
27 changes: 20 additions & 7 deletions src/job-writer.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand All @@ -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<typeof Bun.spawnSync<'pipe', 'pipe'>>;
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 } } {
Expand Down
Loading
Loading