FE-1261: Recover prepared execution journals

.changeset/recover-prepared-execution-journals.md

@@ -0,0 +1,5 @@
+---
+"@hashintel/brunch": patch
+---
+
+Recover prepared execution journals across direct lifecycle steps and superseding runs.

src/.pi/extensions/__tests__/registry.test.ts

@@ -1371,6 +1371,11 @@ describe('Brunch explicit Pi extension registry', () => {
       sideEffects: [
         { kind: 'mkdir', path: dirname(newMetadataPath) },
         { kind: 'write_file', path: newMetadataPath, ifExists: 'overwrite' },
+        { kind: 'mkdir', path: join(dirname(newMetadataPath), 'petrinaut') },
+        { kind: 'write_file', path: join(dirname(newMetadataPath), 'petrinaut', 'plan.json') },
+        { kind: 'write_file', path: join(dirname(newMetadataPath), 'petrinaut', 'net.json') },
+        { kind: 'write_file', path: join(dirname(newMetadataPath), 'petrinaut', 'net.sdcpn.json') },
+        { kind: 'write_file', path: join(dirname(newMetadataPath), 'petrinaut', 'events.jsonl') },
       ],
     });
     await expect(readFile(oldMetadataPath, 'utf8')).resolves.not.toContain('supersedesRunId');

src/.pi/extensions/executor/execute-report-init/index.ts

@@ -24,7 +24,7 @@ export function createExecuteReportInitTool() {
     name: BRUNCH_EXECUTE_REPORT_INIT_TOOL,
     label: 'execute_report_init',
     description:
-      'Initialize reports.jsonl for a source-copied cook run. Does not execute slices or create Petri artifacts.',
+      'Initialize reports.jsonl for a source-copied cook run and reconcile an existing Petri observation journal. Does not execute slices or create observer definitions.',
     parameters: toolParameters(ExecuteReportInitParams),
     async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
       const cwd = ctx?.cwd;

src/.pi/extensions/executor/execute-slice-start/index.ts

@@ -25,7 +25,7 @@ export function createExecuteSliceStartTool() {
     name: BRUNCH_EXECUTE_SLICE_START_TOOL,
     label: 'execute_slice_start',
     description:
-      'Append a slice-start marker for a ready cook run. Does not execute agents, tests, or Petri transitions.',
+      'Append a slice-start marker and reconcile its prepared Petri observation. Does not execute agents or tests.',
     parameters: toolParameters(ExecuteSliceStartParams),
     async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
       const cwd = ctx?.cwd;

src/executor/__tests__/orchestrate.test.ts

@@ -1725,7 +1725,12 @@ describe('drive', () => {
           reason: 'petri_input_unreadable',
         });
       } else {
-        await expect(driving).rejects.toThrow();
+        await expect(driving).resolves.toEqual({
+          status: 'halted',
+          step: 'epic_verify',
+          runStatus: 'slice_completed',
+          reason: 'petri_marking_persist_failed',
+        });
       }
       expect(runnerCalls).toBe(0);
       expect((await readRunMetadata(runMetadataPath(cwd, 'run-1')))?.verifiedEpicIds).toBeUndefined();
@@ -3576,7 +3581,7 @@ describe('drive', () => {
     expect(snapshot?.terminalEventKind).toBeUndefined();
   });
 
-  it('does not repeat an effect when its metadata advanced before the transition append failed', async () => {
+  it('repairs a lifecycle journal gap without repeating the completed effect', async () => {
     const cwd = await mkdtemp(join(tmpdir(), 'brunch-drive-journal-gap-restart-'));
     await createRunAtCreated(cwd, ['task-1']);
     await preparePetriObservation({ cwd, runId: 'run-1' });
@@ -3609,14 +3614,129 @@ describe('drive', () => {
     await rename(preservedJournalPath, journalPath);
 
     await expect(drive({ cwd, runId: 'run-1', ports })).resolves.toEqual({
+      status: 'completed',
+      runStatus: 'promotion_prepared',
+    });
+    expect(worktreeCalls).toBe(1);
+    await expect(readRunMetadata(runMetadataPath(cwd, 'run-1'))).resolves.toMatchObject({
+      status: 'promotion_prepared',
+    });
+  });
+
+  it('halts and repairs when lifecycle catch-up cannot persist its marking', async () => {
+    const cwd = await mkdtemp(join(tmpdir(), 'brunch-drive-reconcile-marking-failure-'));
+    await createRunAtCreated(cwd, ['task-1']);
+    await preparePetriObservation({ cwd, runId: 'run-1' });
+    await createWorktree({ cwd, runId: 'run-1', gitWorktree: createFakeGitWorktreePort() });
+    const markingPath = petriMarkingPath(cwd, 'run-1');
+    await mkdir(markingPath);
+
+    await expect(drive({ cwd, runId: 'run-1', ports: fakePorts() })).resolves.toEqual({
       status: 'halted',
       step: 'populate',
       runStatus: 'worktree_created',
-      reason: 'petri_journal_gap',
+      reason: 'petri_marking_persist_failed',
+    });
+    expect(
+      (await readPetriEvents(cwd)).flatMap((event) =>
+        event.kind === 'transition_fired' ? [event.transitionId] : [],
+      ),
+    ).toEqual(['worktree_create']);
+
+    await rm(markingPath, { recursive: true });
+    await expect(drive({ cwd, runId: 'run-1', ports: fakePorts() })).resolves.toEqual({
+      status: 'completed',
+      runStatus: 'promotion_prepared',
+    });
+    await expect(readPetriMarkingSnapshot({ cwd, runId: 'run-1' })).resolves.toMatchObject({
+      lifecycleProvenance: { runStatus: 'promotion_prepared' },
+    });
+  });
+
+  it('does not append when Petri authority becomes unreadable during a reconciling step', async () => {
+    const cwd = await mkdtemp(join(tmpdir(), 'brunch-drive-reconcile-fail-closed-'));
+    await createRunAtCreated(cwd, ['task-1']);
+    const journalPath = petriEventsPath(cwd, 'run-1');
+    const corruptJournal = '{not-json\n';
+    let failureWakeUps = 0;
+    const unsubscribe = subscribePetriJournalFailures({
+      cwd,
+      runId: 'run-1',
+      listener: () => {
+        failureWakeUps += 1;
+      },
+    });
+
+    let outcome: Awaited<ReturnType<typeof drive>>;
+    try {
+      outcome = await drive({
+        cwd,
+        runId: 'run-1',
+        ports: fakePorts(),
+        onStepStart(step) {
+          if (step !== 'report_init') return;
+          writeFileSync(journalPath, corruptJournal, 'utf8');
+        },
+      });
+    } finally {
+      unsubscribe();
+    }
+
+    expect(await readFile(journalPath, 'utf8')).toBe(corruptJournal);
+    expect(failureWakeUps).toBe(1);
+    expect(outcome).toEqual({
+      status: 'halted',
+      step: 'report_init',
+      runStatus: 'source_copied',
+      reason: 'petri_input_unreadable',
     });
-    expect(worktreeCalls).toBe(1);
     await expect(readRunMetadata(runMetadataPath(cwd, 'run-1'))).resolves.toMatchObject({
-      status: 'worktree_created',
+      status: 'source_copied',
+    });
+  });
+
+  it('reconciles direct mechanical lifecycle steps into the prepared Petri journal', async () => {
+    const cwd = await mkdtemp(join(tmpdir(), 'brunch-drive-direct-lifecycle-journal-'));
+    await createRunAtCreated(cwd, ['task-1']);
+    await preparePetriObservation({ cwd, runId: 'run-1' });
+    await createWorktree({ cwd, runId: 'run-1', gitWorktree: createFakeGitWorktreePort() });
+    await populateWorktree({ cwd, runId: 'run-1' });
+    await selectSourcePolicy({ cwd, runId: 'run-1', policy: 'plan_only' });
+    await copyHostSource({ cwd, runId: 'run-1' });
+    await initializeReports({ cwd, runId: 'run-1' });
+
+    expect(
+      (await readPetriEvents(cwd)).flatMap((event) =>
+        event.kind === 'transition_fired' ? [event.transitionId] : [],
+      ),
+    ).toEqual(['worktree_create', 'populate', 'source_policy', 'source_copy', 'report_init']);
+    expect(await readPetriMarkingSnapshot({ cwd, runId: 'run-1' })).toMatchObject({
+      currentMarking: { 'slice:task-1:claim': 1 },
+      firedTransitionCount: 5,
+      lifecycleProvenance: { runStatus: 'reports_initialized' },
+    });
+
+    await expect(
+      drive({ cwd, runId: 'run-1', ports: fakePorts() }, linearScheduler, serialFiringPolicy, {
+        maxFirings: 1,
+      }),
+    ).resolves.toEqual({ status: 'completed', runStatus: 'slice_started' });
+    expect(
+      (await readPetriEvents(cwd)).flatMap((event) =>
+        event.kind === 'transition_fired' ? [event.transitionId] : [],
+      ),
+    ).toEqual([
+      'worktree_create',
+      'populate',
+      'source_policy',
+      'source_copy',
+      'report_init',
+      'slice_start:task-1',
+    ]);
+    expect(await readPetriMarkingSnapshot({ cwd, runId: 'run-1' })).toMatchObject({
+      currentMarking: { 'slice:task-1:started': 1 },
+      firedTransitionCount: 6,
+      lifecycleProvenance: { runStatus: 'slice_started', activeSliceId: 'task-1' },
     });
   });
 
@@ -6227,7 +6347,7 @@ describe('petriScheduler', () => {
     });
   });
 
-  it('refuses standalone start after one parallel claim journals before batch marking', async () => {
+  it('halts before a parallel claim when equal marking repair cannot persist', async () => {
     const cwd = await mkdtemp(join(tmpdir(), 'brunch-petri-single-journal-claim-'));
     await createRunAtCreated(cwd, ['task-1', 'task-2']);
     await expect(
@@ -6241,19 +6361,13 @@ describe('petriScheduler', () => {
     await expect(
       drive({ cwd, runId: 'run-1', ports: fakePorts() }, petriScheduler, frontierFiringPolicy),
     ).resolves.toMatchObject({ status: 'halted', reason: 'petri_marking_persist_failed' });
-    await expect(readRunDetail(cwd, 'run-1')).resolves.toMatchObject({
-      petriReadySteps: [],
-      petriBlockedSteps: [
-        { kind: 'authority_unreadable', blockers: [{ kind: 'parallel_authority_unreadable' }] },
-        {
-          kind: 'slice_start',
-          sliceId: 'task-1',
-          blockers: [{ kind: 'parallel_authority_unreadable' }],
-        },
-      ],
-    });
+    expect(
+      (await readPetriEvents(cwd)).filter(
+        (event) => event.kind === 'transition_fired' && event.transitionId.startsWith('slice_start:'),
+      ),
+    ).toEqual([]);
     await expect(startSlice({ cwd, runId: 'run-1', sliceId: 'task-2' })).resolves.toMatchObject({
-      status: 'parallel_batch_active',
+      status: 'petri_marking_persist_failed',
       sideEffects: [],
     });
     await expect(readRunMetadata(runMetadataPath(cwd, 'run-1'))).resolves.toMatchObject({
@@ -6277,7 +6391,7 @@ describe('petriScheduler', () => {
     });
 
     await expect(startSlice({ cwd, runId: 'run-1', sliceId: 'task-2' })).resolves.toEqual({
-      status: 'parallel_batch_active',
+      status: 'petri_journal_gap',
       runStatus: 'slice_completed',
       runId: 'run-1',
       metadataPath: runMetadataPath(cwd, 'run-1'),

src/executor/__tests__/report.test.ts

@@ -2,12 +2,15 @@ import { access, mkdir, mkdtemp, readFile, writeFile } from 'node:fs/promises';
 import { tmpdir } from 'node:os';
 import { join } from 'node:path';
 
-import { describe, expect, it } from 'vitest';
+import { describe, expect, it, vi } from 'vitest';
 
+import { appendPetriEvent, petriEventsPath } from '../petri-events.js';
+import * as petriLifecycleReconciliation from '../petri-lifecycle-reconciliation.js';
+import { preparePetriObservation } from '../petri.js';
 import { planFilePath } from '../plan-file.js';
 import { populateWorktree } from '../populate.js';
 import { initializeReports, reportsPath } from '../report.js';
-import { runDirPath, runMetadataPath, createRun } from '../run.js';
+import { runDirPath, runMetadataPath, readRunMetadata, createRun } from '../run.js';
 import { copyHostSource } from '../source-copy.js';
 import { selectSourcePolicy } from '../source-policy.js';
 import { createWorktree } from '../worktree.js';
@@ -102,4 +105,64 @@ describe('initializeReports', () => {
     });
     expect(await pathExists(join(runDirPath(cwd, 'run-1'), 'petrinaut'))).toBe(false);
   });
+
+  it('reports a blocked prepared-journal reconciliation after persisting report initialization', async () => {
+    const cwd = await mkdtemp(join(tmpdir(), 'brunch-report-blocked-journal-'));
+    await createSourceCopiedRun(cwd);
+    const reconcile = vi
+      .spyOn(petriLifecycleReconciliation, 'reconcilePreparedLifecycleJournal')
+      .mockResolvedValueOnce({ status: 'synchronized' })
+      .mockResolvedValueOnce({ status: 'blocked', reason: 'petri_input_unreadable' });
+
+    try {
+      await expect(initializeReports({ cwd, runId: 'run-1' })).resolves.toEqual({
+        status: 'petri_input_unreadable',
+        runStatus: 'reports_initialized',
+        runId: 'run-1',
+        metadataPath: runMetadataPath(cwd, 'run-1'),
+        reportsPath: reportsPath(cwd, 'run-1'),
+        sideEffects: [
+          { kind: 'write_file', path: reportsPath(cwd, 'run-1'), ifExists: 'overwrite' },
+          { kind: 'write_file', path: runMetadataPath(cwd, 'run-1'), ifExists: 'overwrite' },
+        ],
+      });
+    } finally {
+      reconcile.mockRestore();
+    }
+    await expect(readRunMetadata(runMetadataPath(cwd, 'run-1'))).resolves.toMatchObject({
+      status: 'reports_initialized',
+    });
+  });
+
+  it('does not initialize reports after the prepared journal records a terminal', async () => {
+    const cwd = await mkdtemp(join(tmpdir(), 'brunch-report-terminal-journal-'));
+    await createSourceCopiedRun(cwd);
+    await preparePetriObservation({ cwd, runId: 'run-1' });
+    await appendPetriEvent({
+      cwd,
+      runId: 'run-1',
+      event: {
+        kind: 'net_halted',
+        runId: 'run-1',
+        runStatus: 'source_copied',
+        step: 'report_init',
+        reason: 'operator_halt',
+        failedSliceIds: [],
+      },
+    });
+    const journalBefore = await readFile(petriEventsPath(cwd, 'run-1'), 'utf8');
+
+    await expect(initializeReports({ cwd, runId: 'run-1' })).resolves.toEqual({
+      status: 'petri_terminal_recorded',
+      runStatus: 'source_copied',
+      runId: 'run-1',
+      metadataPath: runMetadataPath(cwd, 'run-1'),
+      sideEffects: [],
+    });
+    await expect(readRunMetadata(runMetadataPath(cwd, 'run-1'))).resolves.toMatchObject({
+      status: 'source_copied',
+    });
+    await expect(pathExists(reportsPath(cwd, 'run-1'))).resolves.toBe(false);
+    await expect(readFile(petriEventsPath(cwd, 'run-1'), 'utf8')).resolves.toBe(journalBefore);
+  });
 });