From 9a6f221aad6ed96d43cb1af421c04f4e1e5da3cf Mon Sep 17 00:00:00 2001 From: trvon Date: Sun, 30 Aug 2026 11:01:03 -0600 Subject: [PATCH 1/3] test(workflow): probe blocker reconciliation --- .../blocker-reconciliation-experiment.test.ts | 293 ++++++++++++++++++ .../blocker-reconciliation-experiment.ts | 187 +++++++++++ 2 files changed, 480 insertions(+) create mode 100644 test/blocker-reconciliation-experiment.test.ts create mode 100644 test/experiments/blocker-reconciliation-experiment.ts diff --git a/test/blocker-reconciliation-experiment.test.ts b/test/blocker-reconciliation-experiment.test.ts new file mode 100644 index 0000000..6ac3597 --- /dev/null +++ b/test/blocker-reconciliation-experiment.test.ts @@ -0,0 +1,293 @@ +import { mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { LoopStore } from "../src/store.js"; +import { TaskStore } from "../src/task-store.js"; +import { + attemptClaimedTransition, + contextFor, + type ExperimentClaim, + type ExperimentObservation, + reconcileClaim, +} from "./experiments/blocker-reconciliation-experiment.js"; + +const NOW = 1_800_000_000_000; +const directories: string[] = []; + +function setup() { + const directory = mkdtempSync(join(tmpdir(), "pi-loop-reconciliation-experiment-")); + directories.push(directory); + const loopPath = join(directory, "loops.json"); + const taskPath = join(directory, "tasks.json"); + const loopStore = new LoopStore(loopPath); + const taskStore = new TaskStore(taskPath); + const workflow = loopStore.create({ type: "dynamic" }, "Assess a claimed blocker", { + recurring: true, + workflow: { + version: 1, + initialState: "assess", + states: { + assess: { + prompt: "Assess the blocker claim.", + on: { blocked: "blocked", continue: "review" }, + }, + review: { + prompt: "Continue reviewing.", + on: { blocked: "blocked" }, + }, + blocked: { prompt: "Wait for resolution.", terminal: "paused" }, + }, + }, + }); + taskStore.create("Sentinel task", "This standalone task must remain byte-identical during workflow reconciliation."); + return { directory, loopPath, taskPath, loopStore, taskStore, workflow }; +} + +function environmentalClaim( + context: ReturnType, + fact: string, + expected: string | number | boolean | null, +): ExperimentClaim { + return { class: "environmental", fact, expected, context }; +} + +function observation( + context: ReturnType, + fact: string, + actual: string | number | boolean | null, + overrides: Partial = {}, +): ExperimentObservation { + return { + fact, + actual, + sourceClass: "deterministic", + provider: "fixture", + providerVersion: "1", + observedAt: NOW - 10, + expiresAt: NOW + 1_000, + context, + status: "observed", + ...overrides, + }; +} + +afterEach(() => { + for (const directory of directories.splice(0)) rmSync(directory, { recursive: true, force: true }); +}); + +describe("test-only blocker reconciliation experiment", () => { + it("E1/E4 resolves normalized environmental facts without provider-specific core logic", () => { + const { workflow } = setup(); + const context = contextFor(workflow, "workspace-A"); + + expect(reconcileClaim( + environmentalClaim(context, "repository_dirty", true), + [observation(context, "repository_dirty", false)], + NOW, + )).toMatchObject({ decision: "contradicted" }); + expect(reconcileClaim( + environmentalClaim(context, "artifact_present", true), + [observation(context, "artifact_present", true, { provider: "repository-fact" })], + NOW, + )).toMatchObject({ decision: "confirmed" }); + }); + + it("E2/E5 never derives user authority from machine evidence or another workflow scope", () => { + const { workflow } = setup(); + const context = contextFor(workflow, "workspace-A"); + const claim: ExperimentClaim = { + class: "user_authority", + fact: "destructive_change_approved", + expected: true, + context, + }; + + expect(reconcileClaim(claim, [observation(context, claim.fact, true)], NOW)).toMatchObject({ + decision: "requires_user_authority", + }); + expect(reconcileClaim(claim, [observation( + { ...context, workflowId: "other-workflow" }, + claim.fact, + true, + { sourceClass: "user_authority", provider: "explicit-user-decision" }, + )], NOW)).toMatchObject({ decision: "requires_user_authority" }); + expect(reconcileClaim(claim, [observation( + context, + claim.fact, + true, + { sourceClass: "user_authority", provider: "explicit-user-decision" }, + )], NOW)).toMatchObject({ decision: "confirmed" }); + }); + + it("E3 treats monitor status as a narrow fact and conflicts as unresolved", () => { + const { workflow } = setup(); + const context = contextFor(workflow, "workspace-A"); + const claim = environmentalClaim(context, "validation_process_failed", true); + + expect(reconcileClaim(claim, [observation( + context, + claim.fact, + false, + { provider: "monitor-terminal-status" }, + )], NOW)).toMatchObject({ decision: "contradicted" }); + expect(reconcileClaim(claim, [ + observation(context, claim.fact, false, { provider: "monitor-A" }), + observation(context, claim.fact, true, { provider: "monitor-B" }), + ], NOW)).toMatchObject({ decision: "unresolved", reason: "conflicting_observations" }); + }); + + it("E6 leaves absent, abstained, and provider-error evidence unresolved", () => { + const { workflow } = setup(); + const context = contextFor(workflow, "workspace-A"); + const claim = environmentalClaim(context, "service_available", true); + + expect(reconcileClaim(claim, [], NOW)).toMatchObject({ decision: "unresolved" }); + expect(reconcileClaim(claim, [observation(context, claim.fact, true, { status: "abstained" })], NOW)) + .toMatchObject({ decision: "unresolved" }); + expect(reconcileClaim(claim, [observation(context, claim.fact, true, { status: "error" })], NOW)) + .toMatchObject({ decision: "unresolved" }); + }); + + it("E7 rejects expired and context-mismatched observations", () => { + const { workflow } = setup(); + const context = contextFor(workflow, "workspace-A"); + const claim = environmentalClaim(context, "repository_dirty", true); + + expect(reconcileClaim(claim, [observation(context, claim.fact, true, { expiresAt: NOW - 1 })], NOW)) + .toMatchObject({ decision: "unresolved", reason: "no_current_observation" }); + expect(reconcileClaim(claim, [observation( + { ...context, transitionSeq: context.transitionSeq + 1 }, + claim.fact, + true, + )], NOW)).toMatchObject({ decision: "unresolved", reason: "no_current_observation" }); + }); + + it("E8 admits one confirmed claim through the actual LoopStore CAS path without touching TaskStore", () => { + const { loopPath, taskPath, loopStore, workflow } = setup(); + const context = contextFor(workflow, "workspace-A"); + const taskBytes = readFileSync(taskPath); + const result = attemptClaimedTransition({ + store: loopStore, + workflowId: workflow.id, + outcome: "blocked", + claim: environmentalClaim(context, "repository_dirty", true), + observations: [observation(context, "repository_dirty", true)], + now: NOW, + }); + + expect(result).toMatchObject({ decision: { decision: "confirmed" }, transition: { applied: true, terminal: "paused" } }); + expect(loopStore.get(workflow.id)).toMatchObject({ status: "paused", workflow: { currentState: "blocked", transitionSeq: 1 } }); + expect(readFileSync(loopPath)).not.toHaveLength(0); + expect(readFileSync(taskPath)).toEqual(taskBytes); + }); + + it.each([ + ["contradicted", false, "observed"], + ["unresolved", true, "abstained"], + ["provider error", true, "error"], + ] as const)("E8 preserves exact store bytes when a claim is %s", (_label, actual, status) => { + const { loopPath, taskPath, loopStore, workflow } = setup(); + const context = contextFor(workflow, "workspace-A"); + const loopBytes = readFileSync(loopPath); + const taskBytes = readFileSync(taskPath); + + const result = attemptClaimedTransition({ + store: loopStore, + workflowId: workflow.id, + outcome: "blocked", + claim: environmentalClaim(context, "repository_dirty", true), + observations: [observation(context, "repository_dirty", actual, { status })], + now: NOW, + }); + + expect(result.transition).toBeUndefined(); + expect(readFileSync(loopPath)).toEqual(loopBytes); + expect(readFileSync(taskPath)).toEqual(taskBytes); + }); + + it("E7/E8 preserves exact bytes for expired, stale-scope, unauthorized, and malformed evidence", () => { + const cases: Array<{ + label: string; + claimClass?: ExperimentClaim["class"]; + mutate: (item: ExperimentObservation) => ExperimentObservation; + }> = [ + { label: "expired", mutate: (item) => ({ ...item, expiresAt: NOW - 1 }) }, + { + label: "stale scope", + mutate: (item) => ({ + ...item, + context: { ...item.context, definitionRevision: item.context.definitionRevision + 1 }, + }), + }, + { label: "unauthorized", claimClass: "user_authority", mutate: (item) => item }, + { label: "malformed provider", mutate: (item) => ({ ...item, provider: "" }) }, + ]; + + for (const testCase of cases) { + const { loopPath, taskPath, loopStore, workflow } = setup(); + const context = contextFor(workflow, "workspace-A"); + const loopBytes = readFileSync(loopPath); + const taskBytes = readFileSync(taskPath); + const claim: ExperimentClaim = { + class: testCase.claimClass ?? "environmental", + fact: "repository_dirty", + expected: true, + context, + }; + const result = attemptClaimedTransition({ + store: loopStore, + workflowId: workflow.id, + outcome: "blocked", + claim, + observations: [testCase.mutate(observation(context, claim.fact, true))], + now: NOW, + }); + + expect(result.transition, testCase.label).toBeUndefined(); + expect(readFileSync(loopPath), testCase.label).toEqual(loopBytes); + expect(readFileSync(taskPath), testCase.label).toEqual(taskBytes); + } + }); + + it("E8 allows a competing transition to win but rejects the stale confirmed admission", () => { + const { loopStore, workflow } = setup(); + const context = contextFor(workflow, "workspace-A"); + const result = attemptClaimedTransition({ + store: loopStore, + workflowId: workflow.id, + outcome: "blocked", + claim: environmentalClaim(context, "repository_dirty", true), + observations: [observation(context, "repository_dirty", true)], + now: NOW, + beforeCommit: (expected) => { + expect(loopStore.transitionWorkflow(workflow.id, { outcome: "continue" }, expected)).toMatchObject({ applied: true }); + }, + }); + + expect(result).toMatchObject({ + decision: { decision: "confirmed" }, + transition: { applied: false, error: expect.stringContaining("changed") }, + }); + expect(loopStore.get(workflow.id)).toMatchObject({ + status: "active", + workflow: { currentState: "review", transitionSeq: 1 }, + }); + }); + + it("E9 exposes administrative pause as distinct from semantic transition settlement", () => { + const { loopStore, workflow } = setup(); + const before = loopStore.get(workflow.id)!; + const paused = loopStore.pause(workflow.id)!; + + expect(paused).toMatchObject({ + status: "paused", + workflow: { + currentState: before.workflow?.currentState, + transitionSeq: before.workflow?.transitionSeq, + }, + }); + expect(paused.workflow?.lastTransition).toBeUndefined(); + expect(loopStore.resume(workflow.id)).toMatchObject({ status: "active" }); + }); +}); diff --git a/test/experiments/blocker-reconciliation-experiment.ts b/test/experiments/blocker-reconciliation-experiment.ts new file mode 100644 index 0000000..3028938 --- /dev/null +++ b/test/experiments/blocker-reconciliation-experiment.ts @@ -0,0 +1,187 @@ +import { LoopStore } from "../../src/store.js"; +import type { LoopEntry, WorkflowRuntimeActor } from "../../src/types.js"; + +export type ExperimentFactValue = string | number | boolean | null; +export type ExperimentClaimClass = "environmental" | "user_authority"; +export type ExperimentObservationSource = "deterministic" | "user_authority"; +export type ExperimentObservationStatus = "observed" | "abstained" | "error"; +export type ExperimentDecisionKind = "confirmed" | "contradicted" | "unresolved" | "requires_user_authority"; + +export interface ExperimentWorkflowContext { + workflowId: string; + currentState: string; + transitionSeq: number; + definitionRevision: number; + activeExecutionId?: string; + contextDigest: string; +} + +export interface ExperimentClaim { + class: ExperimentClaimClass; + fact: string; + expected: ExperimentFactValue; + context: ExperimentWorkflowContext; +} + +export interface ExperimentObservation { + fact: string; + actual: ExperimentFactValue; + sourceClass: ExperimentObservationSource; + provider: string; + providerVersion: string; + observedAt: number; + expiresAt: number; + context: ExperimentWorkflowContext; + status: ExperimentObservationStatus; +} + +export interface ExperimentDecision { + decision: ExperimentDecisionKind; + reason: string; + providers: string[]; +} + +interface WorkflowExpectedState { + currentState: string; + transitionSeq: number; + definitionRevision: number; + activeExecutionId?: string; +} + +export interface ExperimentTransitionAttempt { + store: LoopStore; + workflowId: string; + outcome: string; + claim: ExperimentClaim; + observations: ExperimentObservation[]; + now: number; + actor?: WorkflowRuntimeActor; + beforeCommit?: (expected: WorkflowExpectedState) => void; +} + +export function contextFor(entry: LoopEntry, contextDigest: string): ExperimentWorkflowContext { + if (!entry.workflow) throw new Error(`Loop #${entry.id} is not a workflow`); + return { + workflowId: entry.id, + currentState: entry.workflow.currentState, + transitionSeq: entry.workflow.transitionSeq, + definitionRevision: entry.workflow.definitionRevision, + activeExecutionId: entry.workflow.activeExecution?.id, + contextDigest, + }; +} + +function validContext(context: ExperimentWorkflowContext): boolean { + return Boolean(context.workflowId.trim() + && context.currentState.trim() + && context.contextDigest.trim() + && Number.isSafeInteger(context.transitionSeq) + && context.transitionSeq >= 0 + && Number.isSafeInteger(context.definitionRevision) + && context.definitionRevision >= 1); +} + +function validObservation(observation: ExperimentObservation): boolean { + return Boolean(observation.fact.trim() + && observation.provider.trim() + && observation.providerVersion.trim() + && validContext(observation.context) + && Number.isFinite(observation.observedAt) + && Number.isFinite(observation.expiresAt) + && observation.observedAt <= observation.expiresAt); +} + +function sameContext(left: ExperimentWorkflowContext, right: ExperimentWorkflowContext): boolean { + return left.workflowId === right.workflowId + && left.currentState === right.currentState + && left.transitionSeq === right.transitionSeq + && left.definitionRevision === right.definitionRevision + && left.activeExecutionId === right.activeExecutionId + && left.contextDigest === right.contextDigest; +} + +function currentObservations( + claim: ExperimentClaim, + observations: ExperimentObservation[], + now: number, +): ExperimentObservation[] { + const requiredSource: ExperimentObservationSource = claim.class === "user_authority" + ? "user_authority" + : "deterministic"; + return observations.filter((observation) => validObservation(observation) + && observation.status === "observed" + && observation.fact === claim.fact + && observation.sourceClass === requiredSource + && observation.observedAt <= now + && observation.expiresAt >= now + && sameContext(observation.context, claim.context)); +} + +export function reconcileClaim( + claim: ExperimentClaim, + observations: ExperimentObservation[], + now: number, +): ExperimentDecision { + if (!claim.fact.trim() || !validContext(claim.context) || !Number.isFinite(now)) { + return { decision: "unresolved", reason: "invalid_claim", providers: [] }; + } + const current = currentObservations(claim, observations, now); + if (current.length === 0) { + return { + decision: claim.class === "user_authority" ? "requires_user_authority" : "unresolved", + reason: "no_current_observation", + providers: [], + }; + } + + const values = new Set(current.map((observation) => `${typeof observation.actual}:${String(observation.actual)}`)); + if (values.size > 1) { + return { + decision: "unresolved", + reason: "conflicting_observations", + providers: current.map((observation) => `${observation.provider}@${observation.providerVersion}`), + }; + } + + return { + decision: Object.is(current[0]?.actual, claim.expected) ? "confirmed" : "contradicted", + reason: "exact_value_comparison", + providers: current.map((observation) => `${observation.provider}@${observation.providerVersion}`), + }; +} + +export function attemptClaimedTransition(input: ExperimentTransitionAttempt): { + decision: ExperimentDecision; + transition?: ReturnType; +} { + const entry = input.store.get(input.workflowId); + if (!entry?.workflow) { + return { + decision: { decision: "unresolved", reason: "workflow_unavailable", providers: [] }, + }; + } + + const currentContext = contextFor(entry, input.claim.context.contextDigest); + if (!sameContext(currentContext, input.claim.context)) { + return { + decision: { decision: "unresolved", reason: "stale_claim_context", providers: [] }, + }; + } + + const decision = reconcileClaim(input.claim, input.observations, input.now); + if (decision.decision !== "confirmed") return { decision }; + + const expected: WorkflowExpectedState = { + currentState: entry.workflow.currentState, + transitionSeq: entry.workflow.transitionSeq, + definitionRevision: entry.workflow.definitionRevision, + activeExecutionId: entry.workflow.activeExecution?.id, + }; + input.beforeCommit?.(expected); + const transition = input.store.transitionWorkflow(input.workflowId, { + outcome: input.outcome, + actor: input.actor, + evidence: `experiment:${input.claim.class}:${input.claim.fact}:${decision.decision}`, + }, expected); + return { decision, transition }; +} From ca19ac6da7309696281153d972d38a4e7522cd4c Mon Sep 17 00:00:00 2001 From: trvon Date: Sun, 30 Aug 2026 11:44:43 -0600 Subject: [PATCH 2/3] test(workflow): harden reconciliation experiment --- .../blocker-reconciliation-experiment.test.ts | 140 +++++++++++++++++- .../blocker-reconciliation-experiment.ts | 22 ++- 2 files changed, 154 insertions(+), 8 deletions(-) diff --git a/test/blocker-reconciliation-experiment.test.ts b/test/blocker-reconciliation-experiment.test.ts index 6ac3597..5e2de04 100644 --- a/test/blocker-reconciliation-experiment.test.ts +++ b/test/blocker-reconciliation-experiment.test.ts @@ -6,6 +6,7 @@ import { LoopStore } from "../src/store.js"; import { TaskStore } from "../src/task-store.js"; import { attemptClaimedTransition, + classifyWorkflowPause, contextFor, type ExperimentClaim, type ExperimentObservation, @@ -135,6 +136,12 @@ describe("test-only blocker reconciliation experiment", () => { observation(context, claim.fact, false, { provider: "monitor-A" }), observation(context, claim.fact, true, { provider: "monitor-B" }), ], NOW)).toMatchObject({ decision: "unresolved", reason: "conflicting_observations" }); + + const signedZeroClaim = environmentalClaim(context, "signed_zero", 0); + expect(reconcileClaim(signedZeroClaim, [ + observation(context, signedZeroClaim.fact, 0, { provider: "number-A" }), + observation(context, signedZeroClaim.fact, -0, { provider: "number-B" }), + ], NOW)).toMatchObject({ decision: "unresolved", reason: "conflicting_observations" }); }); it("E6 leaves absent, abstained, and provider-error evidence unresolved", () => { @@ -170,6 +177,7 @@ describe("test-only blocker reconciliation experiment", () => { const result = attemptClaimedTransition({ store: loopStore, workflowId: workflow.id, + runtimeContextDigest: "workspace-A", outcome: "blocked", claim: environmentalClaim(context, "repository_dirty", true), observations: [observation(context, "repository_dirty", true)], @@ -182,6 +190,44 @@ describe("test-only blocker reconciliation experiment", () => { expect(readFileSync(taskPath)).toEqual(taskBytes); }); + it("E7 rejects claim replay across workflow, state, revision, execution, and workspace scope", () => { + const mutations: Array<{ + label: string; + mutate: (context: ReturnType) => ReturnType; + }> = [ + { label: "workflow", mutate: (context) => ({ ...context, workflowId: "other-workflow" }) }, + { label: "state", mutate: (context) => ({ ...context, currentState: "other-state" }) }, + { label: "revision", mutate: (context) => ({ ...context, definitionRevision: context.definitionRevision + 1 }) }, + { label: "execution", mutate: (context) => ({ ...context, activeExecutionId: "other-execution" }) }, + { label: "workspace", mutate: (context) => ({ ...context, contextDigest: "workspace-B" }) }, + ]; + + for (const replay of mutations) { + const { loopPath, taskPath, loopStore, workflow } = setup(); + const actualContext = contextFor(workflow, "workspace-A"); + const replayedContext = replay.mutate(actualContext); + const loopBytes = readFileSync(loopPath); + const taskBytes = readFileSync(taskPath); + const claim = environmentalClaim(replayedContext, "repository_dirty", true); + const result = attemptClaimedTransition({ + store: loopStore, + workflowId: workflow.id, + runtimeContextDigest: "workspace-A", + outcome: "blocked", + claim, + observations: [observation(replayedContext, claim.fact, true)], + now: NOW, + }); + + expect(result, replay.label).toMatchObject({ + decision: { decision: "unresolved", reason: "stale_claim_context" }, + }); + expect(result.transition, replay.label).toBeUndefined(); + expect(readFileSync(loopPath), replay.label).toEqual(loopBytes); + expect(readFileSync(taskPath), replay.label).toEqual(taskBytes); + } + }); + it.each([ ["contradicted", false, "observed"], ["unresolved", true, "abstained"], @@ -195,6 +241,7 @@ describe("test-only blocker reconciliation experiment", () => { const result = attemptClaimedTransition({ store: loopStore, workflowId: workflow.id, + runtimeContextDigest: "workspace-A", outcome: "blocked", claim: environmentalClaim(context, "repository_dirty", true), observations: [observation(context, "repository_dirty", actual, { status })], @@ -238,6 +285,7 @@ describe("test-only blocker reconciliation experiment", () => { const result = attemptClaimedTransition({ store: loopStore, workflowId: workflow.id, + runtimeContextDigest: "workspace-A", outcome: "blocked", claim, observations: [testCase.mutate(observation(context, claim.fact, true))], @@ -256,6 +304,7 @@ describe("test-only blocker reconciliation experiment", () => { const result = attemptClaimedTransition({ store: loopStore, workflowId: workflow.id, + runtimeContextDigest: "workspace-A", outcome: "blocked", claim: environmentalClaim(context, "repository_dirty", true), observations: [observation(context, "repository_dirty", true)], @@ -275,11 +324,87 @@ describe("test-only blocker reconciliation experiment", () => { }); }); - it("E9 exposes administrative pause as distinct from semantic transition settlement", () => { - const { loopStore, workflow } = setup(); - const before = loopStore.get(workflow.id)!; - const paused = loopStore.pause(workflow.id)!; + it("resubmits delayed environmental evidence safely after file-backed store recreation", () => { + const { loopPath, taskPath, loopStore, workflow } = setup(); + const context = contextFor(workflow, "workspace-A"); + const claim = environmentalClaim(context, "validation_process_failed", true); + const loopBytes = readFileSync(loopPath); + const taskBytes = readFileSync(taskPath); + + const waiting = attemptClaimedTransition({ + store: loopStore, + workflowId: workflow.id, + runtimeContextDigest: "workspace-A", + outcome: "blocked", + claim, + observations: [], + now: NOW, + }); + expect(waiting).toMatchObject({ decision: { decision: "unresolved" } }); + expect(waiting.transition).toBeUndefined(); + expect(readFileSync(loopPath)).toEqual(loopBytes); + const restartedStore = new LoopStore(loopPath); + const resumed = attemptClaimedTransition({ + store: restartedStore, + workflowId: workflow.id, + runtimeContextDigest: "workspace-A", + outcome: "blocked", + claim, + observations: [observation(context, claim.fact, true, { provider: "delayed-monitor-status" })], + now: NOW, + }); + + expect(resumed).toMatchObject({ decision: { decision: "confirmed" }, transition: { applied: true } }); + expect(readFileSync(taskPath)).toEqual(taskBytes); + }); + + it("resubmits a scoped user-authority decision safely after store recreation", () => { + const { loopPath, taskPath, loopStore, workflow } = setup(); + const context = contextFor(workflow, "workspace-A"); + const claim: ExperimentClaim = { + class: "user_authority", + fact: "destructive_change_approved", + expected: true, + context, + }; + const taskBytes = readFileSync(taskPath); + const waiting = attemptClaimedTransition({ + store: loopStore, + workflowId: workflow.id, + runtimeContextDigest: "workspace-A", + outcome: "blocked", + claim, + observations: [observation(context, claim.fact, true)], + now: NOW, + }); + expect(waiting).toMatchObject({ decision: { decision: "requires_user_authority" } }); + expect(waiting.transition).toBeUndefined(); + + const restartedStore = new LoopStore(loopPath); + const resumed = attemptClaimedTransition({ + store: restartedStore, + workflowId: workflow.id, + runtimeContextDigest: "workspace-A", + outcome: "blocked", + claim, + observations: [observation(context, claim.fact, true, { + sourceClass: "user_authority", + provider: "explicit-user-decision", + })], + now: NOW, + }); + + expect(resumed).toMatchObject({ decision: { decision: "confirmed" }, transition: { applied: true } }); + expect(readFileSync(taskPath)).toEqual(taskBytes); + }); + + it("E9 classifies semantic terminal pause separately from unattributed nonsemantic pause", () => { + const administrative = setup(); + const before = administrative.loopStore.get(administrative.workflow.id)!; + expect(classifyWorkflowPause(before)).toBe("not_paused"); + + const paused = administrative.loopStore.pause(administrative.workflow.id)!; expect(paused).toMatchObject({ status: "paused", workflow: { @@ -288,6 +413,11 @@ describe("test-only blocker reconciliation experiment", () => { }, }); expect(paused.workflow?.lastTransition).toBeUndefined(); - expect(loopStore.resume(workflow.id)).toMatchObject({ status: "active" }); + expect(classifyWorkflowPause(paused)).toBe("nonsemantic_unattributed"); + + const semantic = setup(); + const transitioned = semantic.loopStore.transitionWorkflow(semantic.workflow.id, { outcome: "blocked" }); + expect(transitioned).toMatchObject({ applied: true, terminal: "paused" }); + expect(classifyWorkflowPause(semantic.loopStore.get(semantic.workflow.id)!)).toBe("semantic_terminal"); }); }); diff --git a/test/experiments/blocker-reconciliation-experiment.ts b/test/experiments/blocker-reconciliation-experiment.ts index 3028938..b2f5ee5 100644 --- a/test/experiments/blocker-reconciliation-experiment.ts +++ b/test/experiments/blocker-reconciliation-experiment.ts @@ -48,9 +48,12 @@ interface WorkflowExpectedState { activeExecutionId?: string; } +export type ExperimentPauseClass = "not_paused" | "semantic_terminal" | "nonsemantic_unattributed"; + export interface ExperimentTransitionAttempt { store: LoopStore; workflowId: string; + runtimeContextDigest: string; outcome: string; claim: ExperimentClaim; observations: ExperimentObservation[]; @@ -59,6 +62,19 @@ export interface ExperimentTransitionAttempt { beforeCommit?: (expected: WorkflowExpectedState) => void; } +export function classifyWorkflowPause(entry: LoopEntry): ExperimentPauseClass { + if (entry.status !== "paused") return "not_paused"; + const workflow = entry.workflow; + if (!workflow) return "nonsemantic_unattributed"; + const state = workflow.definition.states[workflow.currentState]; + const transition = workflow.lastTransition; + return state?.terminal === "paused" + && transition?.to === workflow.currentState + && transition.sequence === workflow.transitionSeq + ? "semantic_terminal" + : "nonsemantic_unattributed"; +} + export function contextFor(entry: LoopEntry, contextDigest: string): ExperimentWorkflowContext { if (!entry.workflow) throw new Error(`Loop #${entry.id} is not a workflow`); return { @@ -134,8 +150,8 @@ export function reconcileClaim( }; } - const values = new Set(current.map((observation) => `${typeof observation.actual}:${String(observation.actual)}`)); - if (values.size > 1) { + const firstValue = current[0]!.actual; + if (current.some((observation) => !Object.is(observation.actual, firstValue))) { return { decision: "unresolved", reason: "conflicting_observations", @@ -161,7 +177,7 @@ export function attemptClaimedTransition(input: ExperimentTransitionAttempt): { }; } - const currentContext = contextFor(entry, input.claim.context.contextDigest); + const currentContext = contextFor(entry, input.runtimeContextDigest); if (!sameContext(currentContext, input.claim.context)) { return { decision: { decision: "unresolved", reason: "stale_claim_context", providers: [] }, From 8718ddcddbf8d43c8e4f68c0e9ed25ccd3590e16 Mon Sep 17 00:00:00 2001 From: trvon Date: Sun, 30 Aug 2026 12:35:01 -0600 Subject: [PATCH 3/3] feat(workflow): ground blocker transitions --- AGENTS.md | 5 +- README.md | 2 +- benchmarks/workloads.ts | 12 +- docs/REFERENCE.md | 6 +- docs/USAGE_GUIDE.md | 9 +- src/api.ts | 3 + src/index.ts | 11 +- src/loop-format.ts | 8 + src/loop-reducer.ts | 18 +- src/runtime/subagent-orchestration-runtime.ts | 4 +- src/runtime/workflow-admission-providers.ts | 35 ++ src/scheduler.ts | 2 +- src/store.ts | 68 ++- src/tools/loop-tools.ts | 1 + src/tools/workflow-tools.ts | 80 +++- src/trigger-system.ts | 2 +- src/types.ts | 20 + src/workflow-admission.ts | 274 +++++++++++ src/workflow-reducer.ts | 26 ++ .../blocker-reconciliation-experiment.test.ts | 423 ----------------- .../blocker-reconciliation-experiment.ts | 203 --------- test/index.test.ts | 1 + test/loop-command.test.ts | 13 +- test/loop-reducer.test.ts | 4 +- test/loop-tools.test.ts | 72 ++- test/property/reducers.property.test.ts | 2 +- test/store.test.ts | 114 ++++- test/workflow-admission-providers.test.ts | 64 +++ test/workflow-admission.test.ts | 430 ++++++++++++++++++ 29 files changed, 1223 insertions(+), 689 deletions(-) create mode 100644 src/runtime/workflow-admission-providers.ts create mode 100644 src/workflow-admission.ts delete mode 100644 test/blocker-reconciliation-experiment.test.ts delete mode 100644 test/experiments/blocker-reconciliation-experiment.ts create mode 100644 test/workflow-admission-providers.test.ts create mode 100644 test/workflow-admission.test.ts diff --git a/AGENTS.md b/AGENTS.md index d0905b3..ebc7d5e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -26,6 +26,7 @@ src/index.ts extension registration and runtime wiring src/api.ts supported @trevonistrevon/pi-loop/api surface src/types.ts loop, workflow, revision, monitor contracts src/store.ts LoopStore workflow/orchestration atomic mutations +src/workflow-admission.ts provider-neutral blocker transition admission src/task-store.ts standalone native task persistence src/*-reducer.ts pure state transitions src/coordinator.ts reducer/effect coordination @@ -48,10 +49,12 @@ A workflow is one dynamic `LoopEntry` with a version-1 named-state definition. - The creator owns the initial execution lease. - Every destination/retry execution starts unowned and requires `WorkflowClaim`. - `WorkflowTransition` validates the live owner, settles source work, records evidence, advances state, and creates destination work in one locked write. +- Paused terminal outcomes require a typed blocker claim; trusted providers run outside the LoopStore lock, then the transition uses exact state/revision/execution CAS. +- Machine observations never grant user authority. Rejected, stale, or contradicted claims are state-preserving; restart recovery is explicit resubmission, not a persisted proposal. - `WorkflowRevise` applies typed additive changes with definition/state/sequence CAS, immutable prior-definition history, and no scheduler or TaskStore effect. - Current materialized state content is immutable. Current outgoing edges and future state content may be revised. - Transition CAS includes definition revision so transition/revision races fail closed in either order. -- Terminal completed workflows are deleted; terminal paused workflows remain inspectable. +- Terminal completed workflows are deleted; terminal paused workflows remain inspectable with bounded admission and pause provenance. ## Subagent orchestration contract diff --git a/README.md b/README.md index 2a1a54e..eb5c9f8 100644 --- a/README.md +++ b/README.md @@ -64,7 +64,7 @@ OrchestrationGet id="1" | `/loop` | Create or manage scheduled, event, and dynamic goal loops | | `/tasks` | Manage native fallback tasks when `pi-tasks` is absent | | `LoopCreate`, `LoopList`, `LoopUpdate`, `LoopDelete` | Create and control ordinary loops | -| `WorkflowCreate`, `WorkflowClaim`, `WorkflowRevise`, `WorkflowTransition` | Create, claim, revise, and advance task-driven workflows; inspect them with `LoopList` | +| `WorkflowCreate`, `WorkflowClaim`, `WorkflowRevise`, `WorkflowTransition` | Create, claim, revise, and advance workflows; paused terminals require trusted blocker admission | | `OrchestrationCreate`, `OrchestrationGet` | Run and inspect a finite batch of independent async subagent work; cancel with `LoopDelete` | | `MonitorCreate`, `MonitorList`, `MonitorStop` | Run and inspect background commands | | `TaskCreate`, `TaskList`, `TaskClaim`, `TaskHeartbeat`, `TaskUpdate`, `TaskDelete` | Native fallback task management | diff --git a/benchmarks/workloads.ts b/benchmarks/workloads.ts index d5514b0..99872de 100644 --- a/benchmarks/workloads.ts +++ b/benchmarks/workloads.ts @@ -33,13 +33,11 @@ function buildLoopState(): LoopReducerState { const loopState = buildLoopState(); const loopEvents: LoopReducerEvent[] = Array.from({ length: 1_000 }, (_, index) => { const id = String((index % 25) + 1); - const type = ["LOOP_FIRED", "LOOP_PAUSED", "LOOP_RESUMED"] as const; - return { - type: type[index % type.length] ?? "LOOP_FIRED", - at: 1_000 + index, - source: "system", - payload: { id }, - }; + const types = ["LOOP_FIRED", "LOOP_PAUSED", "LOOP_RESUMED"] as const; + const type = types[index % types.length] ?? "LOOP_FIRED"; + return type === "LOOP_PAUSED" + ? { type, at: 1_000 + index, source: "system", payload: { id, kind: "administrative" } } + : { type, at: 1_000 + index, source: "system", payload: { id } }; }); function buildTaskState(): TaskReducerState { diff --git a/docs/REFERENCE.md b/docs/REFERENCE.md index d8d791e..75c2da4 100644 --- a/docs/REFERENCE.md +++ b/docs/REFERENCE.md @@ -41,7 +41,7 @@ Project scope shares durable state but does not yet elect one scheduler owner ac ## Loop model -`LoopEntry.status` is `active` or `paused`. Triggers are: +`LoopEntry.status` is `active` or `paused`. New pauses persist optional provenance as `pause:{kind,at,reason?}`. Kinds distinguish `administrative`, `controller_limit`, `semantic_terminal`, and `orchestration_settlement`; older paused snapshots may remain unattributed. Resume clears the record. Triggers are: - cron: `{type:"cron", schedule}` - event: `{type:"event", source, filter?}` @@ -81,7 +81,9 @@ A run persists current state, transition sequence, attempts, state fire counts, The initial task execution is leased to the creating runtime. Every destination execution, including self-loop retries, starts unowned. `WorkflowClaim` claims unowned work, renews the same owner, or takes over an expired lease. Live foreign ownership fails closed. -`WorkflowTransition` validates the current lease, declared available outcome, attempt limit, active execution, and definition revision. One locked write settles source work, records evidence, advances state, and creates the destination execution. A missing or exhausted route is handled through `WorkflowRevise`, not a fabricated transition. Completed terminal states delete the controller; paused terminal states preserve it for inspection and represent a declared blocker or required user authority—not a progress notification. +`WorkflowTransition` validates the current lease, declared available outcome, attempt limit, active execution, and definition revision. Ordinary transitions proceed directly. A transition into a `paused` terminal state additionally requires `claim:{class,provider,subject,fact,expected}`. A trusted provider observes the fact outside the LoopStore lock; admission rejects missing, unavailable, stale, expired, conflicting, contradicted, or cross-context observations without writing. The built-in `monitor` provider exposes only `status`, `exitCode`, and `stopReason`; monitor output is never evidence. Admission confirms the exact fact, not whether workflow policy should treat that fact as a blocker—the declared edge remains the workflow author's policy. No user-authority provider is built in, and machine providers cannot grant `user_authority`, so those claims fail closed. After confirmation, the existing state/revision/execution CAS protects the locked transition. + +No pending claim or general evidence ledger is persisted. A confirmed transition stores only a bounded admission receipt on `lastTransition` (claim class/provider/subject/fact/expected value, provider versions, and decision time). After restart, callers inspect current state and explicitly resubmit; stale context is rejected. One locked transition write settles source work, records evidence and the receipt, advances state, and creates the destination execution. A missing or exhausted route is handled through `WorkflowRevise`, not a fabricated transition. Completed terminals delete the controller; paused terminals preserve it with `semantic_terminal` pause provenance and represent a declared blocker—not a progress notification. ### Adaptive revision diff --git a/docs/USAGE_GUIDE.md b/docs/USAGE_GUIDE.md index d4491f3..f9ce9c7 100644 --- a/docs/USAGE_GUIDE.md +++ b/docs/USAGE_GUIDE.md @@ -104,9 +104,12 @@ The initial state must be non-terminal. Each wake presents the current state, st ```text WorkflowTransition id="1" outcome="root_cause_found" evidence="A null config reaches the parser." WorkflowTransition id="1" outcome="tests_pass" evidence="Targeted and full test suites pass." +WorkflowTransition id="1" outcome="blocked" evidence="Monitor #m1 failed." claim='{"class":"environmental","provider":"monitor","subject":"m1","fact":"status","expected":"error"}' ``` -`WorkflowTransition` validates the branch, settles the current execution, records evidence, and activates the next state's execution in the same locked write. Newly entered task phases start unowned so another agent sharing project scope can claim the next phase immediately; whichever agent continues must call `WorkflowClaim id="1"` first. Workflow work is embedded in the loop controller — never call `TaskClaim` or `TaskUpdate` for it. `WorkflowClaim` also renews the current runtime's lease or takes over an expired lease after a restart. A self-loop creates a fresh unowned attempt execution and increments the displayed attempt count. When a target reaches `maxAttempts`, only outcomes leading to that target become unavailable; other declared outcomes remain selectable. Reaching a `completed` terminal state deletes the workflow loop; reaching a `paused` terminal state preserves it in paused state for inspection or deletion. Terminal workflow states cannot be resumed. Task status does not guess an outcome—the model selects one explicitly. LoopList and workflow wakes omit outcomes whose target state has exhausted `maxAttempts`. +`WorkflowTransition` validates the branch, settles the current execution, records evidence, and activates the next state's execution in the same locked write. A `paused` terminal target requires the typed claim shown above. The provider runs before the LoopStore lock, observations are scoped to the current workflow/state/revision/execution/workspace, and the final write uses the existing CAS. The built-in `monitor` provider exposes only `status`, `exitCode`, and `stopReason`; raw output is never admission evidence. Missing, stale, conflicting, contradicted, or unavailable evidence leaves every store unchanged. User-authority claims fail closed because machine evidence cannot manufacture consent. + +Newly entered task phases start unowned so another agent sharing project scope can claim the next phase immediately; whichever agent continues must call `WorkflowClaim id="1"` first. Workflow work is embedded in the loop controller — never call `TaskClaim` or `TaskUpdate` for it. `WorkflowClaim` also renews the current runtime's lease or takes over an expired lease after a restart. A self-loop creates a fresh unowned attempt execution and increments the displayed attempt count. When a target reaches `maxAttempts`, only outcomes leading to that target become unavailable; other declared outcomes remain selectable. Reaching a `completed` terminal state deletes the workflow loop; reaching an admitted `paused` terminal state preserves it with `semantic_terminal` pause provenance. Administrative, controller-limit, and orchestration-settlement pauses carry distinct provenance; legacy snapshots may be unattributed. Terminal workflow states cannot be resumed. Task status does not guess an outcome—the model selects one explicitly. LoopList and workflow wakes omit outcomes whose target state has exhausted `maxAttempts`. When active work discovers a missing prerequisite or supersedes future instructions, inspect `LoopList` and submit one typed revision against its exact definition revision, state, and transition sequence: @@ -120,7 +123,7 @@ WorkflowRevise id="1" expectedRevision=1 expectedState="investigate" expectedTra `WorkflowRevise` stores the prior definition, reason, accepted changes, timestamp, and runtime actor as immutable history. It preserves current execution and lease state, changes only future work or current outgoing edges, and rejects stale revisions or transitions. It never creates standalone tasks. See the [reference](./REFERENCE.md#adaptive-revision) for operation and graph rules. -A missing prerequisite, missing route, or exhausted route is a plan gap—not automatically a blocker. Persist an actionable recovery route with `WorkflowRevise`, then continue through the revised transition and claim while work remains actionable. Call `WorkflowTransition` only when an available declared outcome is supported by evidence; never fabricate one. Do not stop or move the controller to terminal `paused` merely to report progress. Reserve that state for a declared blocker or required user authority. +A missing prerequisite, missing route, or exhausted route is a plan gap—not automatically a blocker. Persist an actionable recovery route with `WorkflowRevise`, then continue through the revised transition and claim while work remains actionable. Call `WorkflowTransition` only when an available declared outcome is supported by evidence; never fabricate one. Do not stop or move the controller to terminal `paused` merely to report progress. Environmental blockers require trusted admission. If user authority is required, report the exact decision needed; machine observations cannot authorize the transition. To repeat a state until evidence supports an outcome, add a cron-only state policy: `"loop":{"schedule":"0 7 * * *","maxFires":10,"startImmediately":false}`. Only the active state's policy is armed. Scheduled wakes retain the active execution; `WorkflowTransition` remains the only operation that settles it and unlocks the destination execution and cadence. Below the fire cap, a no-change iteration leaves the workflow active; persist material future-plan changes with `WorkflowRevise`. Reaching `maxFires` pauses the workflow and schedules no next cadence. Transition when evidence supports an available outcome; otherwise revise in a bounded recovery state/route, then transition and claim it. State policies do not wake immediately unless `startImmediately` is `true`. @@ -300,6 +303,6 @@ Session files live under `.pi/loops/` and `.pi/tasks/`. Keep `session` as the no ## Status line and limits -The TUI status line summarizes ordinary loops, workflows, orchestrations, running monitors, and native tasks. Use `LoopList`, `OrchestrationGet`, `MonitorList`, and `/tasks` for detail. `LoopList` reports active-loop `age` as wall-clock time since creation, including pause and process downtime; paused loops omit the field. The status clears when no work is active. +The TUI status line summarizes ordinary loops, workflows, orchestrations, running monitors, and native tasks. Use `LoopList`, `OrchestrationGet`, `MonitorList`, and `/tasks` for detail. `LoopList` reports active-loop `age` as wall-clock time since creation, including pause and process downtime; paused loops omit age and show available pause provenance. The status clears when no work is active. The runtime allows at most 25 active loops and 25 running monitors. Each orchestration batch allows up to 32 work items, 8 local workers, and 3 attempts per item. diff --git a/src/api.ts b/src/api.ts index 4592114..c009bd9 100644 --- a/src/api.ts +++ b/src/api.ts @@ -41,6 +41,8 @@ export type { TaskClaimInput, TaskClaimResult } from "./task-store.js"; export { TaskStore } from "./task-store.js"; export type { TaskClaim, TaskEntry, TaskStatus, TaskStoreData } from "./task-types.js"; export type { + LoopPauseKind, + LoopPauseRecord, MonitorOutcome, OrchestrationActor, OrchestrationConsumeStatus, @@ -55,6 +57,7 @@ export type { OrchestrationWakeReason, OrchestrationWorkItem, OrchestrationWorkStatus, + WorkflowAdmissionRecord, WorkflowDefinition, WorkflowDefinitionRevision, WorkflowMonitorWait, diff --git a/src/index.ts b/src/index.ts index dd16301..5551dc2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -32,6 +32,7 @@ import { isStaleExtensionContextError } from "./runtime/stale-context.js"; import { createSubagentOrchestrationRuntime, type SubagentOrchestrationRuntime } from "./runtime/subagent-orchestration-runtime.js"; import { createTaskBacklogRuntime } from "./runtime/task-backlog-runtime.js"; import { createTaskProviderRuntime, type TaskProviderRuntime } from "./runtime/task-provider-runtime.js"; +import { createMonitorWorkflowAdmissionProvider } from "./runtime/workflow-admission-providers.js"; import { CronScheduler } from "./scheduler.js"; import { LoopStore } from "./store.js"; import { registerLoopTools } from "./tools/loop-tools.js"; @@ -64,6 +65,7 @@ export default function (pi: ExtensionAPI) { let store = new LoopStore(resolveLoopStorePath(getScopeOptions())); const memoryLoopStores = new Map(); const monitorManager = new MonitorManager(pi); + const monitorWorkflowAdmissionProvider = createMonitorWorkflowAdmissionProvider((id) => monitorManager.get(id)); let scheduler: CronScheduler; let triggerSystem: TriggerSystem; const widget = new LoopWidget(store, monitorManager); @@ -256,7 +258,7 @@ export default function (pi: ExtensionAPI) { if (atMaxFires(current)) { debug(`loop #${current.id} — reached maxFires ${current.maxFires}, retiring`); triggerSystem.remove(current.id); - if (current.workflow || current.taskBacklog) store.pause(current.id); + if (current.workflow || current.taskBacklog) store.pause(current.id, "controller_limit", "loop fire cap reached"); else store.delete(current.id); widget.update(); return; @@ -280,14 +282,14 @@ export default function (pi: ExtensionAPI) { if (atMaxFires(firedEntry)) { triggerSystem.remove(firedEntry.id); - if (firedEntry.workflow || firedEntry.taskBacklog) store.pause(firedEntry.id); + if (firedEntry.workflow || firedEntry.taskBacklog) store.pause(firedEntry.id, "controller_limit", "loop fire cap reached"); else store.delete(firedEntry.id); widget.update(); } if (firedEntry.workflow && atWorkflowStateFireLimit(firedEntry.workflow)) { triggerSystem.remove(firedEntry.id); - store.pause(firedEntry.id); + store.pause(firedEntry.id, "controller_limit", "workflow state fire cap reached"); widget.update(); } @@ -443,6 +445,9 @@ export default function (pi: ExtensionAPI) { onLoopFire(entry); }, getActor: () => _sessionId ? { sessionId: _sessionId, runtimeId } : undefined, + getAdmissionContextDigest: () => resolveLoopStorePath(getScopeOptions(), _sessionId) + ?? `memory:${process.cwd()}:${_sessionId ?? "unbound"}`, + getAdmissionProviders: () => [monitorWorkflowAdmissionProvider], }); function handleMonitorDoneLoop(doneLoop: LoopEntry, monitorId: string): void { diff --git a/src/loop-format.ts b/src/loop-format.ts index b380262..88a7beb 100644 --- a/src/loop-format.ts +++ b/src/loop-format.ts @@ -7,6 +7,14 @@ export function formatLastTransitionLines(lastTransition: WorkflowTransitionReco const { from, to, outcome, evidence } = lastTransition; const lines = [`Last transition: ${from} → ${to} via ${outcome}`]; if (evidence) lines.push(`Evidence: ${evidence.replace(/\s+/g, " ")}`); + if (lastTransition.admission) { + const admission = lastTransition.admission; + const provider = admission.provider.replace(/\s+/g, " "); + const subject = admission.subject.replace(/\s+/g, " "); + const fact = admission.fact.replace(/\s+/g, " "); + const observations = admission.observations.map((observation) => observation.replace(/\s+/g, " ")).join(", "); + lines.push(`Admission: ${admission.claimClass} · ${provider}:${subject}.${fact} = ${JSON.stringify(admission.expected)} · ${observations}`); + } return lines; } diff --git a/src/loop-reducer.ts b/src/loop-reducer.ts index c7bd3e6..ded5d84 100644 --- a/src/loop-reducer.ts +++ b/src/loop-reducer.ts @@ -1,5 +1,5 @@ import { applyOrchestrationEvent, createOrchestrationState, type OrchestrationEvent } from "./orchestration-reducer.js"; -import type { DynamicLoopState, LoopEntry, LoopFireOrigin, OrchestrationActor, OrchestrationDefinitionInput, Trigger, WorkflowDefinition, WorkflowMonitorWait, WorkflowRevisionChange, WorkflowRuntimeActor } from "./types.js"; +import type { DynamicLoopState, LoopEntry, LoopFireOrigin, LoopPauseKind, OrchestrationActor, OrchestrationDefinitionInput, Trigger, WorkflowAdmissionRecord, WorkflowDefinition, WorkflowMonitorWait, WorkflowRevisionChange, WorkflowRuntimeActor } from "./types.js"; import { createWorkflowRun, transitionWorkflowRun } from "./workflow-reducer.js"; import { reviseWorkflowRun, type WorkflowRevisionSummary } from "./workflow-revision.js"; @@ -43,9 +43,16 @@ export type LoopReducerEvent = orchestration?: { definition: OrchestrationDefinitionInput; owner: OrchestrationActor }; }; } + | { + type: "LOOP_PAUSED"; + at: number; + source: ReducerSource; + entityType?: "loop"; + entityId?: string; + payload: { id: string; kind: LoopPauseKind; reason?: string }; + } | { type: - | "LOOP_PAUSED" | "LOOP_RESUMED" | "LOOP_FIRED" | "LOOP_DELETED" @@ -114,6 +121,7 @@ export type LoopReducerEvent = id: string; outcome: string; evidence?: string; + admission?: WorkflowAdmissionRecord; actor?: WorkflowRuntimeActor; }; } @@ -127,6 +135,7 @@ export type LoopReducerEvent = id: string; outcome: string; evidence?: string; + admission?: WorkflowAdmissionRecord; actor?: WorkflowRuntimeActor; terminal: "completed" | "paused"; }; @@ -259,11 +268,13 @@ export function reduceLoopState(state: LoopReducerState, event: LoopReducerEvent if (event.type === "LOOP_PAUSED") { loop.status = "paused"; + loop.pause = { kind: event.payload.kind, at: event.at, ...(event.payload.reason ? { reason: event.payload.reason } : {}) }; loop.updatedAt = event.at; } if (event.type === "LOOP_RESUMED") { loop.status = "active"; + loop.pause = undefined; loop.updatedAt = event.at; } @@ -336,6 +347,7 @@ export function reduceLoopState(state: LoopReducerState, event: LoopReducerEvent const result = transitionWorkflowRun(loop.workflow, { outcome: event.payload.outcome, evidence: event.payload.evidence, + admission: event.payload.admission, actor: event.payload.actor, }, event.at); if (!result.applied) return { state, effects: [] }; @@ -358,6 +370,7 @@ export function reduceLoopState(state: LoopReducerState, event: LoopReducerEvent const result = transitionWorkflowRun(loop.workflow, { outcome: event.payload.outcome, evidence: event.payload.evidence, + admission: event.payload.admission, actor: event.payload.actor, }, event.at); if (!result.applied || result.terminal !== event.payload.terminal) return { state, effects: [] }; @@ -381,6 +394,7 @@ export function reduceLoopState(state: LoopReducerState, event: LoopReducerEvent lastUpdatedAt: event.at, }; loop.status = "paused"; + loop.pause = { kind: "semantic_terminal", at: event.at }; loop.updatedAt = event.at; } diff --git a/src/runtime/subagent-orchestration-runtime.ts b/src/runtime/subagent-orchestration-runtime.ts index ab31ac1..602ff2c 100644 --- a/src/runtime/subagent-orchestration-runtime.ts +++ b/src/runtime/subagent-orchestration-runtime.ts @@ -453,7 +453,7 @@ export function createSubagentOrchestrationRuntime( if (wakeQueued.has(key)) continue; wakeQueued.add(key); if (state.status === "completed" || (state.status === "needs_attention" && getOrchestrationCounts(state).active === 0)) { - getStore().pause(entry.id); + getStore().pause(entry.id, "orchestration_settlement", `orchestration ${state.status}`); } emitWake(getStore().get(entry.id) ?? entry, wake); } @@ -597,7 +597,7 @@ export function createSubagentOrchestrationRuntime( await Promise.all(agentIds.map((agentId) => stopCancelledAgent(id, agentId))); await drainDispatches(id); if (action === "delete") getStore().delete(id); - else getStore().pause(id); + else getStore().pause(id, "administrative", "orchestration cancelled by operator"); for (const key of wakeQueued) { if (key.startsWith(`${id}:`)) wakeQueued.delete(key); } diff --git a/src/runtime/workflow-admission-providers.ts b/src/runtime/workflow-admission-providers.ts new file mode 100644 index 0000000..c974bb7 --- /dev/null +++ b/src/runtime/workflow-admission-providers.ts @@ -0,0 +1,35 @@ +import type { MonitorEntry } from "../types.js"; +import type { WorkflowAdmissionProvider, WorkflowFactValue } from "../workflow-admission.js"; + +const OBSERVATION_TTL_MS = 5_000; + +function monitorFact(entry: MonitorEntry, fact: string): WorkflowFactValue | undefined { + if (fact === "status") return entry.status; + if (fact === "exitCode") return entry.exitCode ?? null; + if (fact === "stopReason") return entry.stopReason ?? null; + return undefined; +} + +export function createMonitorWorkflowAdmissionProvider( + getMonitor: (id: string) => MonitorEntry | undefined, +): WorkflowAdmissionProvider { + return { + id: "monitor", + sourceClass: "environmental", + async observe({ claim, context, now }) { + const entry = getMonitor(claim.subject); + const actual = entry && monitorFact(entry, claim.fact); + return [{ + fact: claim.fact, + actual: actual ?? null, + sourceClass: "environmental", + provider: "monitor", + providerVersion: "1", + observedAt: now, + expiresAt: now + OBSERVATION_TTL_MS, + context, + status: entry && actual !== undefined ? "observed" : "abstained", + }]; + }, + }; +} diff --git a/src/scheduler.ts b/src/scheduler.ts index 452f1b3..b0bfd13 100644 --- a/src/scheduler.ts +++ b/src/scheduler.ts @@ -61,7 +61,7 @@ export class CronScheduler { } private retire(entry: LoopEntry): void { - if (entry.workflow || entry.taskBacklog) this.store.pause(entry.id); + if (entry.workflow || entry.taskBacklog) this.store.pause(entry.id, "controller_limit", "scheduler fire cap reached"); else this.store.delete(entry.id); this.fireTimes.delete(entry.id); } diff --git a/src/store.ts b/src/store.ts index cb43167..7b6ed99 100644 --- a/src/store.ts +++ b/src/store.ts @@ -3,9 +3,9 @@ import { join } from "node:path"; import { type LoopReducerEffect, type LoopReducerEvent, type LoopReducerState, reduceLoopState } from "./loop-reducer.js"; import { applyOrchestrationEvent, type OrchestrationEvent, validateOrchestrationDefinition, validatePersistedOrchestration } from "./orchestration-reducer.js"; import { ReducerBackedStore } from "./reducer-backed-store.js"; -import type { DynamicLoopState, LoopDeletionTombstone, LoopDeletionTombstoneInput, LoopEntry, LoopFireOrigin, LoopStoreData, OrchestrationActor, OrchestrationDefinitionInput, Trigger, WorkflowDefinition, WorkflowMonitorWait, WorkflowRevisionFailure, WorkflowRunState, WorkflowRuntimeActor, WorkflowTerminalStatus } from "./types.js"; +import type { DynamicLoopState, LoopDeletionTombstone, LoopDeletionTombstoneInput, LoopEntry, LoopFireOrigin, LoopPauseKind, LoopPauseRecord, LoopStoreData, OrchestrationActor, OrchestrationDefinitionInput, Trigger, WorkflowDefinition, WorkflowMonitorWait, WorkflowRevisionFailure, WorkflowRunState, WorkflowRuntimeActor, WorkflowTerminalStatus } from "./types.js"; import { validateWorkflowDefinition } from "./workflow-definition.js"; -import { isTerminalWorkflowRun, transitionWorkflowRun, type WorkflowTransitionFailure, type WorkflowTransitionInput } from "./workflow-reducer.js"; +import { isTerminalWorkflowRun, transitionWorkflowRun, validateWorkflowAdmissionRecord, type WorkflowTransitionFailure, type WorkflowTransitionInput } from "./workflow-reducer.js"; import { validatePersistedWorkflowRevision, type WorkflowRevisionInput, type WorkflowRevisionSummary } from "./workflow-revision.js"; const LOOPS_DIR = join(homedir(), ".pi", "loops"); @@ -19,6 +19,9 @@ const LOOPS_DIR = join(homedir(), ".pi", "loops"); */ function normalizeWorkflowRunState(workflow: WorkflowRunState): WorkflowRunState { const legacy = workflow as WorkflowRunState & { activeTaskId?: string }; + if (workflow.lastTransition?.admission && validateWorkflowAdmissionRecord(workflow.lastTransition.admission)) { + throw new Error("Malformed workflow admission provenance"); + } const hasRevision = Object.hasOwn(workflow, "definitionRevision"); const hasHistory = Object.hasOwn(workflow, "revisionHistory"); if (hasRevision !== hasHistory) throw new Error("Malformed workflow revision metadata: revision and history must appear together"); @@ -61,9 +64,30 @@ function normalizeWorkflowRunState(workflow: WorkflowRunState): WorkflowRunState } : {}), }; } +const LOOP_PAUSE_KINDS = new Set([ + "administrative", + "controller_limit", + "semantic_terminal", + "orchestration_settlement", +]); + +function normalizePauseRecord(entry: LoopEntry): LoopPauseRecord | undefined { + const pause = entry.pause; + if (!pause) return undefined; + if (entry.status !== "paused" + || !LOOP_PAUSE_KINDS.has(pause.kind) + || !Number.isFinite(pause.at) + || (pause.reason !== undefined && (typeof pause.reason !== "string" || pause.reason.length > 512))) { + throw new Error("Malformed loop pause provenance"); + } + return pause; +} + function normalizeLoopEntry(entry: LoopEntry): LoopEntry { + const pause = normalizePauseRecord(entry); return { ...entry, + ...(pause ? { pause } : {}), ...(entry.workflow ? { workflow: normalizeWorkflowRunState(entry.workflow) } : {}), ...(entry.orchestration ? { orchestration: validatePersistedOrchestration(entry.orchestration) } : {}), }; @@ -151,17 +175,19 @@ export class LoopStore extends ReducerBackedStore { const entry = this.entries.get(id); if (!entry) return undefined; + if (entry.status === "paused") return entry; + const boundedReason = reason?.trim().slice(0, 512); this.applyReducerEvent({ type: "LOOP_PAUSED", at: Date.now(), source: "tool", entityType: "loop", entityId: id, - payload: { id }, + payload: { id, kind, ...(boundedReason ? { reason: boundedReason } : {}) }, }); return this.entries.get(id); }); @@ -306,14 +332,24 @@ export class LoopStore extends ReducerBackedStore WorkflowStoreLike; getTriggerSystem: () => TriggerSystemLike; getActor: () => WorkflowRuntimeActor | undefined; + getAdmissionContextDigest: () => string; + getAdmissionProviders: () => WorkflowAdmissionProvider[]; updateWidget: () => void; onDynamicLoopActivated?: (entry: LoopEntry) => void; } @@ -113,6 +116,15 @@ function parseWorkflowDefinition(input: string): { definition?: WorkflowDefiniti const WORKFLOW_DEFINITION_EXAMPLE = '{"version":1,"initialState":"collect","states":{"collect":{"prompt":"Collect evidence.","on":{"ready":"publish"}},"publish":{"prompt":"Publish the result.","terminal":"completed"}}}'; +const WorkflowFactValueSchema = Type.Union([Type.String({ maxLength: 1_024 }), Type.Number(), Type.Boolean(), Type.Null()]); +const WorkflowBlockerClaimSchema = Type.Object({ + class: Type.Union([Type.Literal("environmental"), Type.Literal("user_authority")]), + provider: Type.String({ minLength: 1, maxLength: 64, description: "Provider ID (built-in: monitor)" }), + subject: Type.String({ minLength: 1, maxLength: 256, description: "Provider subject or monitor ID" }), + fact: Type.String({ minLength: 1, maxLength: 64, description: "Fact; monitor: status, exitCode, stopReason" }), + expected: WorkflowFactValueSchema, +}); + function workflowDefaultMaxFires(definition: WorkflowDefinition): number { const loopBudget = Object.values(definition.states).reduce((total, state) => total + (state.loop?.maxFires ?? 0), 0); return loopBudget > 0 ? loopBudget : 30; @@ -138,6 +150,7 @@ export function formatWorkflowSummary(entry: LoopEntry, heading: string, failure const attempt = workflow.attemptsByState[workflow.currentState] ?? 1; const attemptLabel = state?.maxAttempts ? `${attempt}/${state.maxAttempts}` : String(attempt); let message = `${heading}\nGoal: ${entry.prompt}\nDefinition revision: ${workflow.definitionRevision ?? 1}\nCurrent state: ${workflow.currentState}\nTransition sequence: ${workflow.transitionSeq}\nAttempt: ${attemptLabel}`; + if (entry.pause) message += `\nPause cause: ${entry.pause.kind}${entry.pause.reason ? ` — ${entry.pause.reason}` : ""}`; if (workflow.lastTransition) message += `\n${formatLastTransitionLines(workflow.lastTransition).join("\n")}`; if (state?.prompt) message += `\nInstruction: ${state.prompt}`; const execution = workflow.activeExecution; @@ -169,7 +182,16 @@ export function formatWorkflowSummary(entry: LoopEntry, heading: string, failure } export function registerWorkflowTools(options: WorkflowToolsOptions): void { - const { pi, getStore, getTriggerSystem, getActor, updateWidget, onDynamicLoopActivated } = options; + const { + pi, + getStore, + getTriggerSystem, + getActor, + getAdmissionContextDigest, + getAdmissionProviders, + updateWidget, + onDynamicLoopActivated, + } = options; pi.registerTool({ name: "WorkflowCreate", @@ -352,27 +374,57 @@ export function registerWorkflowTools(options: WorkflowToolsOptions): void { label: "WorkflowTransition", renderCall: renderToolCall("Workflow", (args) => `transition · #${String(toolArg(args, "id") ?? "?")} → ${String(toolArg(args, "outcome") ?? "?")}`), renderResult: renderToolResult, - description: "Advance one declared workflow outcome. The controller authorizes the current runtime lease; it never accepts a claim token.", + description: "Advance a declared outcome; it never accepts a claim token. Paused terminal outcomes require trusted blocker admission outside LoopStore.", promptGuidelines: [ - "WorkflowTransition uses id, outcome, and optional evidence; claimId is invalid. Outcome must be available and declared; inspect LoopList first.", - "If no outcome fits or the plan changed, use WorkflowRevise first—never fabricate an outcome or terminal-pause merely to report progress.", + "WorkflowTransition uses id, outcome, and optional evidence; paused terminal outcomes require a typed claim; claimId is invalid.", + "Machine evidence cannot grant user authority. If no outcome fits, use WorkflowRevise; never fabricate an outcome or terminal pause.", ], parameters: Type.Object({ id: Type.String({ description: "Workflow loop ID" }), outcome: Type.String({ description: "Declared outcome for the current workflow state" }), evidence: Type.Optional(Type.String({ description: "Concise evidence supporting this transition" })), + claim: Type.Optional(WorkflowBlockerClaimSchema), }), async execute(_toolCallId, params) { const store = getStore(); - const current = store.get(params.id); const actor = getActor(); - const expected = current?.workflow ? { - currentState: current.workflow.currentState, - transitionSeq: current.workflow.transitionSeq, - definitionRevision: current.workflow.definitionRevision, - activeExecutionId: current.workflow.activeExecution?.id, - } : undefined; - const result = store.transitionWorkflow(params.id, { outcome: params.outcome, evidence: params.evidence, actor }, expected); + const contextDigest = getAdmissionContextDigest(); + const admission = await admitWorkflowTransition({ + store, + workflowId: params.id, + outcome: params.outcome, + evidence: params.evidence, + actor, + claim: params.claim, + contextDigest, + providers: getAdmissionProviders(), + isContextCurrent: () => { + const currentActor = getActor(); + return getStore() === store + && getAdmissionContextDigest() === contextDigest + && currentActor?.sessionId === actor?.sessionId + && currentActor?.runtimeId === actor?.runtimeId; + }, + }); + const result = admission.transition; + if (!result) { + const entry = store.get(params.id); + const next = admission.decision.decision === "requires_user_authority" + ? "Ask the user for the explicit decision and keep the workflow active; this runtime has no authority provider." + : admission.decision.reason === "claim_required" + ? "Retry with a typed claim backed by a trusted provider." + : "Refresh the scoped evidence and retry; unresolved or contradicted claims never mutate workflow state."; + const message = `Workflow #${params.id} did not transition\nAdmission: ${admission.decision.decision} (${admission.decision.reason})\nNext: ${next}`; + return textResult(message, entry?.workflow + ? workflowDisplayDetails({ + entry, + action: "transition", + tone: "error", + summary: `Workflow #${params.id} transition admission rejected`, + extra: [`Admission: ${admission.decision.decision} (${admission.decision.reason})`, `Next: ${next}`], + }) + : { kind: "workflow", action: "transition", tone: "error", summary: `Workflow #${params.id} transition admission rejected`, expanded: message.split("\n").slice(1) }); + } if (!result.applied || !result.entry) { const entry = store.get(params.id); const error = result.error ?? "unknown transition error"; diff --git a/src/trigger-system.ts b/src/trigger-system.ts index 2788736..ce51952 100644 --- a/src/trigger-system.ts +++ b/src/trigger-system.ts @@ -106,7 +106,7 @@ export class TriggerSystem { private retire(entry: LoopEntry): void { this.remove(entry.id); - if (entry.workflow || entry.taskBacklog) this.store.pause(entry.id); + if (entry.workflow || entry.taskBacklog) this.store.pause(entry.id, "controller_limit", "trigger fire cap reached"); else this.store.delete(entry.id); } diff --git a/src/types.ts b/src/types.ts index dcc6ad1..7ecea99 100644 --- a/src/types.ts +++ b/src/types.ts @@ -12,6 +12,14 @@ export type LoopDeletionTombstoneInput = Omit; +} + +interface WorkflowAdmissionStore { + get(id: string): LoopEntry | undefined; + transitionWorkflow: LoopStore["transitionWorkflow"]; +} + +export interface WorkflowAdmissionRequest { + store: WorkflowAdmissionStore; + workflowId: string; + outcome: string; + evidence?: string; + actor?: WorkflowRuntimeActor; + claim?: WorkflowBlockerClaim; + contextDigest: string; + providers: WorkflowAdmissionProvider[]; + now?: number | (() => number); + isContextCurrent?: () => boolean; +} + +export interface WorkflowAdmissionResult { + decision: WorkflowAdmissionDecision; + transition?: ReturnType; +} + +function workflowContext(entry: LoopEntry, contextDigest: string): WorkflowAdmissionContext | undefined { + const workflow = entry.workflow; + if (!workflow || !contextDigest.trim()) return undefined; + return { + workflowId: entry.id, + currentState: workflow.currentState, + transitionSeq: workflow.transitionSeq, + definitionRevision: workflow.definitionRevision, + activeExecutionId: workflow.activeExecution?.id, + contextDigest, + }; +} + +function sameContext(left: WorkflowAdmissionContext, right: WorkflowAdmissionContext): boolean { + return left.workflowId === right.workflowId + && left.currentState === right.currentState + && left.transitionSeq === right.transitionSeq + && left.definitionRevision === right.definitionRevision + && left.activeExecutionId === right.activeExecutionId + && left.contextDigest === right.contextDigest; +} + +function validFactValue(value: WorkflowFactValue): boolean { + return value === null + || (typeof value === "string" && value.length <= MAX_FACT_VALUE_STRING_LENGTH) + || typeof value === "boolean" + || (typeof value === "number" && Number.isFinite(value)); +} + +function validClaim(claim: WorkflowBlockerClaim): boolean { + return Boolean(claim.provider.trim() + && claim.provider.length <= MAX_PROVIDER_ID_LENGTH + && claim.subject.trim() + && claim.subject.length <= MAX_SUBJECT_LENGTH + && claim.fact.trim() + && claim.fact.length <= MAX_FACT_LENGTH + && validFactValue(claim.expected)); +} + +function validObservation(observation: WorkflowAdmissionObservation): boolean { + return Boolean(observation.fact.trim() + && observation.fact.length <= MAX_FACT_LENGTH + && validFactValue(observation.actual) + && observation.provider.trim() + && observation.provider.length <= MAX_PROVIDER_ID_LENGTH + && observation.providerVersion.trim() + && observation.providerVersion.length <= MAX_PROVIDER_ID_LENGTH + && Number.isFinite(observation.observedAt) + && Number.isFinite(observation.expiresAt) + && observation.observedAt <= observation.expiresAt); +} + +export function reconcileWorkflowClaim( + claim: WorkflowBlockerClaim, + observations: WorkflowAdmissionObservation[], + context: WorkflowAdmissionContext, + now: number, +): WorkflowAdmissionDecision { + if (!validClaim(claim) || !Number.isFinite(now)) { + return { decision: "unresolved", reason: "invalid_claim", providers: [] }; + } + const current = observations.filter((observation) => validObservation(observation) + && observation.status === "observed" + && observation.fact === claim.fact + && observation.provider === claim.provider + && observation.sourceClass === claim.class + && observation.observedAt <= now + && observation.expiresAt >= now + && sameContext(observation.context, context)); + if (current.length === 0) { + return { + decision: claim.class === "user_authority" ? "requires_user_authority" : "unresolved", + reason: "no_current_observation", + providers: [], + }; + } + const firstValue = current[0]!.actual; + if (current.some((observation) => !Object.is(observation.actual, firstValue))) { + return { + decision: "unresolved", + reason: "conflicting_observations", + providers: current.map((observation) => `${observation.provider}@${observation.providerVersion}`), + }; + } + return { + decision: Object.is(firstValue, claim.expected) ? "confirmed" : "contradicted", + reason: "exact_value_comparison", + providers: current.map((observation) => `${observation.provider}@${observation.providerVersion}`), + }; +} + +function expectedState(context: WorkflowAdmissionContext) { + return { + currentState: context.currentState, + transitionSeq: context.transitionSeq, + definitionRevision: context.definitionRevision, + activeExecutionId: context.activeExecutionId, + }; +} + +function terminalPauseTarget(entry: LoopEntry, outcome: string): boolean { + const workflow = entry.workflow; + const state = workflow?.definition.states[workflow.currentState]; + const targetId = state?.on?.[outcome]; + return targetId !== undefined && workflow?.definition.states[targetId]?.terminal === "paused"; +} + +export async function admitWorkflowTransition(input: WorkflowAdmissionRequest): Promise { + const entry = input.store.get(input.workflowId); + const context = entry && workflowContext(entry, input.contextDigest); + if (!entry?.workflow || !context) { + return { decision: { decision: "unresolved", reason: "workflow_unavailable", providers: [] } }; + } + const expected = expectedState(context); + if (input.isContextCurrent && !input.isContextCurrent()) { + return { decision: { decision: "unresolved", reason: "stale_runtime_context", providers: [] } }; + } + if (!terminalPauseTarget(entry, input.outcome)) { + return { + decision: { decision: "not_required", reason: "ordinary_transition", providers: [] }, + transition: input.store.transitionWorkflow(input.workflowId, { + outcome: input.outcome, + evidence: input.evidence, + actor: input.actor, + }, expected), + }; + } + const claim = input.claim; + if (!claim) { + return { decision: { decision: "unresolved", reason: "claim_required", providers: [] } }; + } + if (!validClaim(claim)) { + return { decision: { decision: "unresolved", reason: "invalid_claim", providers: [] } }; + } + const matchingProviders = input.providers.filter((provider) => provider.id === claim.provider + && provider.sourceClass === claim.class).slice(0, MAX_ADMISSION_PROVIDERS); + if (matchingProviders.length === 0) { + return { + decision: { + decision: claim.class === "user_authority" ? "requires_user_authority" : "unresolved", + reason: "provider_unavailable", + providers: [], + }, + }; + } + + const clock = typeof input.now === "function" + ? input.now + : input.now === undefined + ? Date.now + : () => input.now as number; + const observationTime = clock(); + // Providers may block or inspect other runtimes, so observation must finish before the CAS-protected store mutation begins. + const settled = await Promise.allSettled(matchingProviders.map((provider) => provider.observe({ claim, context, now: observationTime }))); + const observations: WorkflowAdmissionObservation[] = []; + for (const result of settled) { + if (result.status !== "fulfilled" || !Array.isArray(result.value)) continue; + observations.push(...result.value.slice(0, MAX_ADMISSION_OBSERVATIONS - observations.length)); + if (observations.length === MAX_ADMISSION_OBSERVATIONS) break; + } + const decisionTime = clock(); + const decision = reconcileWorkflowClaim(claim, observations, context, decisionTime); + if (decision.decision !== "confirmed") return { decision }; + if (input.isContextCurrent && !input.isContextCurrent()) { + return { decision: { decision: "unresolved", reason: "stale_runtime_context", providers: decision.providers } }; + } + + return { + decision, + transition: input.store.transitionWorkflow(input.workflowId, { + outcome: input.outcome, + evidence: input.evidence, + admission: { + claimClass: claim.class, + provider: claim.provider, + subject: claim.subject, + fact: claim.fact, + expected: claim.expected, + observations: decision.providers, + decidedAt: decisionTime, + }, + actor: input.actor, + }, expected), + }; +} diff --git a/src/workflow-reducer.ts b/src/workflow-reducer.ts index 427f5c9..ddf1188 100644 --- a/src/workflow-reducer.ts +++ b/src/workflow-reducer.ts @@ -1,4 +1,5 @@ import type { + WorkflowAdmissionRecord, WorkflowDefinition, WorkflowExecutionRecord, WorkflowRunState, @@ -8,9 +9,29 @@ import type { WorkflowTransitionRecord, } from "./types.js"; +export function validateWorkflowAdmissionRecord(record: WorkflowAdmissionRecord | undefined): string | undefined { + if (!record) return "Paused terminal transitions require trusted admission"; + const validValue = record.expected === null + || (typeof record.expected === "string" && record.expected.length <= 1_024) + || typeof record.expected === "boolean" + || (typeof record.expected === "number" && Number.isFinite(record.expected)); + if ((record.claimClass !== "environmental" && record.claimClass !== "user_authority") + || !record.provider || record.provider.length > 64 + || !record.subject || record.subject.length > 256 + || !record.fact || record.fact.length > 64 + || !validValue + || !Array.isArray(record.observations) || record.observations.length < 1 || record.observations.length > 8 + || record.observations.some((observation) => typeof observation !== "string" || !observation || observation.length > 129) + || !Number.isFinite(record.decidedAt)) { + return "Paused terminal transition admission is malformed"; + } + return undefined; +} + export interface WorkflowTransitionInput { outcome: string; evidence?: string; + admission?: WorkflowAdmissionRecord; actor?: WorkflowRuntimeActor; } @@ -187,6 +208,10 @@ export function transitionWorkflowRun( const target = resolved.target; const targetState = run.definition.states[target]; if (!targetState) return { applied: false, error: `Transition target "${target}" is not defined` }; + if (targetState.terminal === "paused" || input.admission) { + const admissionError = validateWorkflowAdmissionRecord(input.admission); + if (admissionError) return { applied: false, error: admissionError }; + } const sequence = run.transitionSeq + 1; const settled = settleExecution(run.activeExecution, input.evidence, at); const destination = targetState.task && !targetState.terminal @@ -197,6 +222,7 @@ export function transitionWorkflowRun( to: target, outcome: input.outcome, evidence: input.evidence, + admission: input.admission, at, sequence, }; diff --git a/test/blocker-reconciliation-experiment.test.ts b/test/blocker-reconciliation-experiment.test.ts deleted file mode 100644 index 5e2de04..0000000 --- a/test/blocker-reconciliation-experiment.test.ts +++ /dev/null @@ -1,423 +0,0 @@ -import { mkdtempSync, readFileSync, rmSync } from "node:fs"; -import { tmpdir } from "node:os"; -import { join } from "node:path"; -import { afterEach, describe, expect, it } from "vitest"; -import { LoopStore } from "../src/store.js"; -import { TaskStore } from "../src/task-store.js"; -import { - attemptClaimedTransition, - classifyWorkflowPause, - contextFor, - type ExperimentClaim, - type ExperimentObservation, - reconcileClaim, -} from "./experiments/blocker-reconciliation-experiment.js"; - -const NOW = 1_800_000_000_000; -const directories: string[] = []; - -function setup() { - const directory = mkdtempSync(join(tmpdir(), "pi-loop-reconciliation-experiment-")); - directories.push(directory); - const loopPath = join(directory, "loops.json"); - const taskPath = join(directory, "tasks.json"); - const loopStore = new LoopStore(loopPath); - const taskStore = new TaskStore(taskPath); - const workflow = loopStore.create({ type: "dynamic" }, "Assess a claimed blocker", { - recurring: true, - workflow: { - version: 1, - initialState: "assess", - states: { - assess: { - prompt: "Assess the blocker claim.", - on: { blocked: "blocked", continue: "review" }, - }, - review: { - prompt: "Continue reviewing.", - on: { blocked: "blocked" }, - }, - blocked: { prompt: "Wait for resolution.", terminal: "paused" }, - }, - }, - }); - taskStore.create("Sentinel task", "This standalone task must remain byte-identical during workflow reconciliation."); - return { directory, loopPath, taskPath, loopStore, taskStore, workflow }; -} - -function environmentalClaim( - context: ReturnType, - fact: string, - expected: string | number | boolean | null, -): ExperimentClaim { - return { class: "environmental", fact, expected, context }; -} - -function observation( - context: ReturnType, - fact: string, - actual: string | number | boolean | null, - overrides: Partial = {}, -): ExperimentObservation { - return { - fact, - actual, - sourceClass: "deterministic", - provider: "fixture", - providerVersion: "1", - observedAt: NOW - 10, - expiresAt: NOW + 1_000, - context, - status: "observed", - ...overrides, - }; -} - -afterEach(() => { - for (const directory of directories.splice(0)) rmSync(directory, { recursive: true, force: true }); -}); - -describe("test-only blocker reconciliation experiment", () => { - it("E1/E4 resolves normalized environmental facts without provider-specific core logic", () => { - const { workflow } = setup(); - const context = contextFor(workflow, "workspace-A"); - - expect(reconcileClaim( - environmentalClaim(context, "repository_dirty", true), - [observation(context, "repository_dirty", false)], - NOW, - )).toMatchObject({ decision: "contradicted" }); - expect(reconcileClaim( - environmentalClaim(context, "artifact_present", true), - [observation(context, "artifact_present", true, { provider: "repository-fact" })], - NOW, - )).toMatchObject({ decision: "confirmed" }); - }); - - it("E2/E5 never derives user authority from machine evidence or another workflow scope", () => { - const { workflow } = setup(); - const context = contextFor(workflow, "workspace-A"); - const claim: ExperimentClaim = { - class: "user_authority", - fact: "destructive_change_approved", - expected: true, - context, - }; - - expect(reconcileClaim(claim, [observation(context, claim.fact, true)], NOW)).toMatchObject({ - decision: "requires_user_authority", - }); - expect(reconcileClaim(claim, [observation( - { ...context, workflowId: "other-workflow" }, - claim.fact, - true, - { sourceClass: "user_authority", provider: "explicit-user-decision" }, - )], NOW)).toMatchObject({ decision: "requires_user_authority" }); - expect(reconcileClaim(claim, [observation( - context, - claim.fact, - true, - { sourceClass: "user_authority", provider: "explicit-user-decision" }, - )], NOW)).toMatchObject({ decision: "confirmed" }); - }); - - it("E3 treats monitor status as a narrow fact and conflicts as unresolved", () => { - const { workflow } = setup(); - const context = contextFor(workflow, "workspace-A"); - const claim = environmentalClaim(context, "validation_process_failed", true); - - expect(reconcileClaim(claim, [observation( - context, - claim.fact, - false, - { provider: "monitor-terminal-status" }, - )], NOW)).toMatchObject({ decision: "contradicted" }); - expect(reconcileClaim(claim, [ - observation(context, claim.fact, false, { provider: "monitor-A" }), - observation(context, claim.fact, true, { provider: "monitor-B" }), - ], NOW)).toMatchObject({ decision: "unresolved", reason: "conflicting_observations" }); - - const signedZeroClaim = environmentalClaim(context, "signed_zero", 0); - expect(reconcileClaim(signedZeroClaim, [ - observation(context, signedZeroClaim.fact, 0, { provider: "number-A" }), - observation(context, signedZeroClaim.fact, -0, { provider: "number-B" }), - ], NOW)).toMatchObject({ decision: "unresolved", reason: "conflicting_observations" }); - }); - - it("E6 leaves absent, abstained, and provider-error evidence unresolved", () => { - const { workflow } = setup(); - const context = contextFor(workflow, "workspace-A"); - const claim = environmentalClaim(context, "service_available", true); - - expect(reconcileClaim(claim, [], NOW)).toMatchObject({ decision: "unresolved" }); - expect(reconcileClaim(claim, [observation(context, claim.fact, true, { status: "abstained" })], NOW)) - .toMatchObject({ decision: "unresolved" }); - expect(reconcileClaim(claim, [observation(context, claim.fact, true, { status: "error" })], NOW)) - .toMatchObject({ decision: "unresolved" }); - }); - - it("E7 rejects expired and context-mismatched observations", () => { - const { workflow } = setup(); - const context = contextFor(workflow, "workspace-A"); - const claim = environmentalClaim(context, "repository_dirty", true); - - expect(reconcileClaim(claim, [observation(context, claim.fact, true, { expiresAt: NOW - 1 })], NOW)) - .toMatchObject({ decision: "unresolved", reason: "no_current_observation" }); - expect(reconcileClaim(claim, [observation( - { ...context, transitionSeq: context.transitionSeq + 1 }, - claim.fact, - true, - )], NOW)).toMatchObject({ decision: "unresolved", reason: "no_current_observation" }); - }); - - it("E8 admits one confirmed claim through the actual LoopStore CAS path without touching TaskStore", () => { - const { loopPath, taskPath, loopStore, workflow } = setup(); - const context = contextFor(workflow, "workspace-A"); - const taskBytes = readFileSync(taskPath); - const result = attemptClaimedTransition({ - store: loopStore, - workflowId: workflow.id, - runtimeContextDigest: "workspace-A", - outcome: "blocked", - claim: environmentalClaim(context, "repository_dirty", true), - observations: [observation(context, "repository_dirty", true)], - now: NOW, - }); - - expect(result).toMatchObject({ decision: { decision: "confirmed" }, transition: { applied: true, terminal: "paused" } }); - expect(loopStore.get(workflow.id)).toMatchObject({ status: "paused", workflow: { currentState: "blocked", transitionSeq: 1 } }); - expect(readFileSync(loopPath)).not.toHaveLength(0); - expect(readFileSync(taskPath)).toEqual(taskBytes); - }); - - it("E7 rejects claim replay across workflow, state, revision, execution, and workspace scope", () => { - const mutations: Array<{ - label: string; - mutate: (context: ReturnType) => ReturnType; - }> = [ - { label: "workflow", mutate: (context) => ({ ...context, workflowId: "other-workflow" }) }, - { label: "state", mutate: (context) => ({ ...context, currentState: "other-state" }) }, - { label: "revision", mutate: (context) => ({ ...context, definitionRevision: context.definitionRevision + 1 }) }, - { label: "execution", mutate: (context) => ({ ...context, activeExecutionId: "other-execution" }) }, - { label: "workspace", mutate: (context) => ({ ...context, contextDigest: "workspace-B" }) }, - ]; - - for (const replay of mutations) { - const { loopPath, taskPath, loopStore, workflow } = setup(); - const actualContext = contextFor(workflow, "workspace-A"); - const replayedContext = replay.mutate(actualContext); - const loopBytes = readFileSync(loopPath); - const taskBytes = readFileSync(taskPath); - const claim = environmentalClaim(replayedContext, "repository_dirty", true); - const result = attemptClaimedTransition({ - store: loopStore, - workflowId: workflow.id, - runtimeContextDigest: "workspace-A", - outcome: "blocked", - claim, - observations: [observation(replayedContext, claim.fact, true)], - now: NOW, - }); - - expect(result, replay.label).toMatchObject({ - decision: { decision: "unresolved", reason: "stale_claim_context" }, - }); - expect(result.transition, replay.label).toBeUndefined(); - expect(readFileSync(loopPath), replay.label).toEqual(loopBytes); - expect(readFileSync(taskPath), replay.label).toEqual(taskBytes); - } - }); - - it.each([ - ["contradicted", false, "observed"], - ["unresolved", true, "abstained"], - ["provider error", true, "error"], - ] as const)("E8 preserves exact store bytes when a claim is %s", (_label, actual, status) => { - const { loopPath, taskPath, loopStore, workflow } = setup(); - const context = contextFor(workflow, "workspace-A"); - const loopBytes = readFileSync(loopPath); - const taskBytes = readFileSync(taskPath); - - const result = attemptClaimedTransition({ - store: loopStore, - workflowId: workflow.id, - runtimeContextDigest: "workspace-A", - outcome: "blocked", - claim: environmentalClaim(context, "repository_dirty", true), - observations: [observation(context, "repository_dirty", actual, { status })], - now: NOW, - }); - - expect(result.transition).toBeUndefined(); - expect(readFileSync(loopPath)).toEqual(loopBytes); - expect(readFileSync(taskPath)).toEqual(taskBytes); - }); - - it("E7/E8 preserves exact bytes for expired, stale-scope, unauthorized, and malformed evidence", () => { - const cases: Array<{ - label: string; - claimClass?: ExperimentClaim["class"]; - mutate: (item: ExperimentObservation) => ExperimentObservation; - }> = [ - { label: "expired", mutate: (item) => ({ ...item, expiresAt: NOW - 1 }) }, - { - label: "stale scope", - mutate: (item) => ({ - ...item, - context: { ...item.context, definitionRevision: item.context.definitionRevision + 1 }, - }), - }, - { label: "unauthorized", claimClass: "user_authority", mutate: (item) => item }, - { label: "malformed provider", mutate: (item) => ({ ...item, provider: "" }) }, - ]; - - for (const testCase of cases) { - const { loopPath, taskPath, loopStore, workflow } = setup(); - const context = contextFor(workflow, "workspace-A"); - const loopBytes = readFileSync(loopPath); - const taskBytes = readFileSync(taskPath); - const claim: ExperimentClaim = { - class: testCase.claimClass ?? "environmental", - fact: "repository_dirty", - expected: true, - context, - }; - const result = attemptClaimedTransition({ - store: loopStore, - workflowId: workflow.id, - runtimeContextDigest: "workspace-A", - outcome: "blocked", - claim, - observations: [testCase.mutate(observation(context, claim.fact, true))], - now: NOW, - }); - - expect(result.transition, testCase.label).toBeUndefined(); - expect(readFileSync(loopPath), testCase.label).toEqual(loopBytes); - expect(readFileSync(taskPath), testCase.label).toEqual(taskBytes); - } - }); - - it("E8 allows a competing transition to win but rejects the stale confirmed admission", () => { - const { loopStore, workflow } = setup(); - const context = contextFor(workflow, "workspace-A"); - const result = attemptClaimedTransition({ - store: loopStore, - workflowId: workflow.id, - runtimeContextDigest: "workspace-A", - outcome: "blocked", - claim: environmentalClaim(context, "repository_dirty", true), - observations: [observation(context, "repository_dirty", true)], - now: NOW, - beforeCommit: (expected) => { - expect(loopStore.transitionWorkflow(workflow.id, { outcome: "continue" }, expected)).toMatchObject({ applied: true }); - }, - }); - - expect(result).toMatchObject({ - decision: { decision: "confirmed" }, - transition: { applied: false, error: expect.stringContaining("changed") }, - }); - expect(loopStore.get(workflow.id)).toMatchObject({ - status: "active", - workflow: { currentState: "review", transitionSeq: 1 }, - }); - }); - - it("resubmits delayed environmental evidence safely after file-backed store recreation", () => { - const { loopPath, taskPath, loopStore, workflow } = setup(); - const context = contextFor(workflow, "workspace-A"); - const claim = environmentalClaim(context, "validation_process_failed", true); - const loopBytes = readFileSync(loopPath); - const taskBytes = readFileSync(taskPath); - - const waiting = attemptClaimedTransition({ - store: loopStore, - workflowId: workflow.id, - runtimeContextDigest: "workspace-A", - outcome: "blocked", - claim, - observations: [], - now: NOW, - }); - expect(waiting).toMatchObject({ decision: { decision: "unresolved" } }); - expect(waiting.transition).toBeUndefined(); - expect(readFileSync(loopPath)).toEqual(loopBytes); - - const restartedStore = new LoopStore(loopPath); - const resumed = attemptClaimedTransition({ - store: restartedStore, - workflowId: workflow.id, - runtimeContextDigest: "workspace-A", - outcome: "blocked", - claim, - observations: [observation(context, claim.fact, true, { provider: "delayed-monitor-status" })], - now: NOW, - }); - - expect(resumed).toMatchObject({ decision: { decision: "confirmed" }, transition: { applied: true } }); - expect(readFileSync(taskPath)).toEqual(taskBytes); - }); - - it("resubmits a scoped user-authority decision safely after store recreation", () => { - const { loopPath, taskPath, loopStore, workflow } = setup(); - const context = contextFor(workflow, "workspace-A"); - const claim: ExperimentClaim = { - class: "user_authority", - fact: "destructive_change_approved", - expected: true, - context, - }; - const taskBytes = readFileSync(taskPath); - const waiting = attemptClaimedTransition({ - store: loopStore, - workflowId: workflow.id, - runtimeContextDigest: "workspace-A", - outcome: "blocked", - claim, - observations: [observation(context, claim.fact, true)], - now: NOW, - }); - expect(waiting).toMatchObject({ decision: { decision: "requires_user_authority" } }); - expect(waiting.transition).toBeUndefined(); - - const restartedStore = new LoopStore(loopPath); - const resumed = attemptClaimedTransition({ - store: restartedStore, - workflowId: workflow.id, - runtimeContextDigest: "workspace-A", - outcome: "blocked", - claim, - observations: [observation(context, claim.fact, true, { - sourceClass: "user_authority", - provider: "explicit-user-decision", - })], - now: NOW, - }); - - expect(resumed).toMatchObject({ decision: { decision: "confirmed" }, transition: { applied: true } }); - expect(readFileSync(taskPath)).toEqual(taskBytes); - }); - - it("E9 classifies semantic terminal pause separately from unattributed nonsemantic pause", () => { - const administrative = setup(); - const before = administrative.loopStore.get(administrative.workflow.id)!; - expect(classifyWorkflowPause(before)).toBe("not_paused"); - - const paused = administrative.loopStore.pause(administrative.workflow.id)!; - expect(paused).toMatchObject({ - status: "paused", - workflow: { - currentState: before.workflow?.currentState, - transitionSeq: before.workflow?.transitionSeq, - }, - }); - expect(paused.workflow?.lastTransition).toBeUndefined(); - expect(classifyWorkflowPause(paused)).toBe("nonsemantic_unattributed"); - - const semantic = setup(); - const transitioned = semantic.loopStore.transitionWorkflow(semantic.workflow.id, { outcome: "blocked" }); - expect(transitioned).toMatchObject({ applied: true, terminal: "paused" }); - expect(classifyWorkflowPause(semantic.loopStore.get(semantic.workflow.id)!)).toBe("semantic_terminal"); - }); -}); diff --git a/test/experiments/blocker-reconciliation-experiment.ts b/test/experiments/blocker-reconciliation-experiment.ts deleted file mode 100644 index b2f5ee5..0000000 --- a/test/experiments/blocker-reconciliation-experiment.ts +++ /dev/null @@ -1,203 +0,0 @@ -import { LoopStore } from "../../src/store.js"; -import type { LoopEntry, WorkflowRuntimeActor } from "../../src/types.js"; - -export type ExperimentFactValue = string | number | boolean | null; -export type ExperimentClaimClass = "environmental" | "user_authority"; -export type ExperimentObservationSource = "deterministic" | "user_authority"; -export type ExperimentObservationStatus = "observed" | "abstained" | "error"; -export type ExperimentDecisionKind = "confirmed" | "contradicted" | "unresolved" | "requires_user_authority"; - -export interface ExperimentWorkflowContext { - workflowId: string; - currentState: string; - transitionSeq: number; - definitionRevision: number; - activeExecutionId?: string; - contextDigest: string; -} - -export interface ExperimentClaim { - class: ExperimentClaimClass; - fact: string; - expected: ExperimentFactValue; - context: ExperimentWorkflowContext; -} - -export interface ExperimentObservation { - fact: string; - actual: ExperimentFactValue; - sourceClass: ExperimentObservationSource; - provider: string; - providerVersion: string; - observedAt: number; - expiresAt: number; - context: ExperimentWorkflowContext; - status: ExperimentObservationStatus; -} - -export interface ExperimentDecision { - decision: ExperimentDecisionKind; - reason: string; - providers: string[]; -} - -interface WorkflowExpectedState { - currentState: string; - transitionSeq: number; - definitionRevision: number; - activeExecutionId?: string; -} - -export type ExperimentPauseClass = "not_paused" | "semantic_terminal" | "nonsemantic_unattributed"; - -export interface ExperimentTransitionAttempt { - store: LoopStore; - workflowId: string; - runtimeContextDigest: string; - outcome: string; - claim: ExperimentClaim; - observations: ExperimentObservation[]; - now: number; - actor?: WorkflowRuntimeActor; - beforeCommit?: (expected: WorkflowExpectedState) => void; -} - -export function classifyWorkflowPause(entry: LoopEntry): ExperimentPauseClass { - if (entry.status !== "paused") return "not_paused"; - const workflow = entry.workflow; - if (!workflow) return "nonsemantic_unattributed"; - const state = workflow.definition.states[workflow.currentState]; - const transition = workflow.lastTransition; - return state?.terminal === "paused" - && transition?.to === workflow.currentState - && transition.sequence === workflow.transitionSeq - ? "semantic_terminal" - : "nonsemantic_unattributed"; -} - -export function contextFor(entry: LoopEntry, contextDigest: string): ExperimentWorkflowContext { - if (!entry.workflow) throw new Error(`Loop #${entry.id} is not a workflow`); - return { - workflowId: entry.id, - currentState: entry.workflow.currentState, - transitionSeq: entry.workflow.transitionSeq, - definitionRevision: entry.workflow.definitionRevision, - activeExecutionId: entry.workflow.activeExecution?.id, - contextDigest, - }; -} - -function validContext(context: ExperimentWorkflowContext): boolean { - return Boolean(context.workflowId.trim() - && context.currentState.trim() - && context.contextDigest.trim() - && Number.isSafeInteger(context.transitionSeq) - && context.transitionSeq >= 0 - && Number.isSafeInteger(context.definitionRevision) - && context.definitionRevision >= 1); -} - -function validObservation(observation: ExperimentObservation): boolean { - return Boolean(observation.fact.trim() - && observation.provider.trim() - && observation.providerVersion.trim() - && validContext(observation.context) - && Number.isFinite(observation.observedAt) - && Number.isFinite(observation.expiresAt) - && observation.observedAt <= observation.expiresAt); -} - -function sameContext(left: ExperimentWorkflowContext, right: ExperimentWorkflowContext): boolean { - return left.workflowId === right.workflowId - && left.currentState === right.currentState - && left.transitionSeq === right.transitionSeq - && left.definitionRevision === right.definitionRevision - && left.activeExecutionId === right.activeExecutionId - && left.contextDigest === right.contextDigest; -} - -function currentObservations( - claim: ExperimentClaim, - observations: ExperimentObservation[], - now: number, -): ExperimentObservation[] { - const requiredSource: ExperimentObservationSource = claim.class === "user_authority" - ? "user_authority" - : "deterministic"; - return observations.filter((observation) => validObservation(observation) - && observation.status === "observed" - && observation.fact === claim.fact - && observation.sourceClass === requiredSource - && observation.observedAt <= now - && observation.expiresAt >= now - && sameContext(observation.context, claim.context)); -} - -export function reconcileClaim( - claim: ExperimentClaim, - observations: ExperimentObservation[], - now: number, -): ExperimentDecision { - if (!claim.fact.trim() || !validContext(claim.context) || !Number.isFinite(now)) { - return { decision: "unresolved", reason: "invalid_claim", providers: [] }; - } - const current = currentObservations(claim, observations, now); - if (current.length === 0) { - return { - decision: claim.class === "user_authority" ? "requires_user_authority" : "unresolved", - reason: "no_current_observation", - providers: [], - }; - } - - const firstValue = current[0]!.actual; - if (current.some((observation) => !Object.is(observation.actual, firstValue))) { - return { - decision: "unresolved", - reason: "conflicting_observations", - providers: current.map((observation) => `${observation.provider}@${observation.providerVersion}`), - }; - } - - return { - decision: Object.is(current[0]?.actual, claim.expected) ? "confirmed" : "contradicted", - reason: "exact_value_comparison", - providers: current.map((observation) => `${observation.provider}@${observation.providerVersion}`), - }; -} - -export function attemptClaimedTransition(input: ExperimentTransitionAttempt): { - decision: ExperimentDecision; - transition?: ReturnType; -} { - const entry = input.store.get(input.workflowId); - if (!entry?.workflow) { - return { - decision: { decision: "unresolved", reason: "workflow_unavailable", providers: [] }, - }; - } - - const currentContext = contextFor(entry, input.runtimeContextDigest); - if (!sameContext(currentContext, input.claim.context)) { - return { - decision: { decision: "unresolved", reason: "stale_claim_context", providers: [] }, - }; - } - - const decision = reconcileClaim(input.claim, input.observations, input.now); - if (decision.decision !== "confirmed") return { decision }; - - const expected: WorkflowExpectedState = { - currentState: entry.workflow.currentState, - transitionSeq: entry.workflow.transitionSeq, - definitionRevision: entry.workflow.definitionRevision, - activeExecutionId: entry.workflow.activeExecution?.id, - }; - input.beforeCommit?.(expected); - const transition = input.store.transitionWorkflow(input.workflowId, { - outcome: input.outcome, - actor: input.actor, - evidence: `experiment:${input.claim.class}:${input.claim.fact}:${decision.decision}`, - }, expected); - return { decision, transition }; -} diff --git a/test/index.test.ts b/test/index.test.ts index 591b367..a16340a 100644 --- a/test/index.test.ts +++ b/test/index.test.ts @@ -62,6 +62,7 @@ describe("workflow runtime wiring", () => { const wake = sentMessages.find((item) => item.message.content.includes("fired (workflow)")); expect(loops.content[0].text).toContain("[paused]"); + expect(loops.content[0].text).toContain("Pause cause: controller_limit"); expect(wake?.message.content).toContain("has reached its fire cap; this workflow is paused and no next cadence is scheduled"); expect(wake?.message.content).toContain("Otherwise add a bounded recovery state/route with WorkflowRevise, then transition and claim it"); expect(wake?.message.content).not.toContain("leave the workflow active for its next cadence"); diff --git a/test/loop-command.test.ts b/test/loop-command.test.ts index 6f57aad..e32c159 100644 --- a/test/loop-command.test.ts +++ b/test/loop-command.test.ts @@ -420,7 +420,18 @@ describe("registerLoopCommand", () => { }, }, }); - h.store.transitionWorkflow("1", { outcome: "blocked" }); + h.store.transitionWorkflow("1", { + outcome: "blocked", + admission: { + claimClass: "environmental", + provider: "test", + subject: "release", + fact: "failed", + expected: true, + observations: ["test@1"], + decidedAt: 1, + }, + }); h.store.pause("1"); const actionChoices: string[][] = []; diff --git a/test/loop-reducer.test.ts b/test/loop-reducer.test.ts index 9fb89a9..4cb700d 100644 --- a/test/loop-reducer.test.ts +++ b/test/loop-reducer.test.ts @@ -283,7 +283,7 @@ describe("loop reducer", () => { source: "tool", entityType: "loop", entityId: "1", - payload: { id: "1" }, + payload: { id: "1", kind: "administrative" }, }); expect(state.loopsById["1"].status).toBe("paused"); @@ -522,7 +522,7 @@ describe("loop reducer", () => { source: "tool", entityType: "loop", entityId: "99", - payload: { id: "99" }, + payload: { id: "99", kind: "administrative" }, }); expect(state).toEqual(initial); diff --git a/test/loop-tools.test.ts b/test/loop-tools.test.ts index 67e9dd6..82ba137 100644 --- a/test/loop-tools.test.ts +++ b/test/loop-tools.test.ts @@ -11,6 +11,7 @@ function setup() { const scheduler = { nextFire: vi.fn(() => undefined) }; const monitorManager = { get: vi.fn(() => undefined) }; const onDynamicLoopActivated = vi.fn(); + const admissionProviders: import("../src/workflow-admission.js").WorkflowAdmissionProvider[] = []; const maybeBootstrapTaskLoop = vi.fn(async () => false); const isTaskSystemReady = vi.fn(() => true); const cancelOrchestration = vi.fn(async (id: string, action: "pause" | "delete") => { @@ -34,12 +35,14 @@ function setup() { getStore: () => store, getTriggerSystem: () => triggerSystem, getActor: () => ({ sessionId: "test-session", runtimeId: "test-runtime" }), + getAdmissionContextDigest: () => "workspace-A", + getAdmissionProviders: () => admissionProviders, updateWidget: vi.fn(), onDynamicLoopActivated, }); const result = async (name: string, args: any) => await toolMap.get(name)!.execute!("t", args); const text = async (name: string, args: any) => (await result(name, args)).content[0].text as string; - return { store, triggerSystem, text, result, toolMap, maybeBootstrapTaskLoop, isTaskSystemReady, onDynamicLoopActivated, cancelOrchestration }; + return { store, triggerSystem, text, result, toolMap, admissionProviders, maybeBootstrapTaskLoop, isTaskSystemReady, onDynamicLoopActivated, cancelOrchestration }; } describe("LoopCreate", () => { @@ -549,6 +552,64 @@ describe("Workflow tools", () => { expect(h.store.get("1")?.workflow?.activeExecution?.lease).toBeUndefined(); }); + it("requires trusted blocker admission before a paused terminal transition", async () => { + const definition = JSON.stringify({ + version: 1, + initialState: "work", + states: { + work: { prompt: "Check release.", on: { blocked: "blocked" } }, + blocked: { prompt: "Report blocker.", terminal: "paused" }, + }, + }); + await h.text("WorkflowCreate", { goal: "Check release", definition }); + + const rejected = await h.text("WorkflowTransition", { id: "1", outcome: "blocked" }); + + expect(rejected).toContain("Admission: unresolved (claim_required)"); + expect(h.store.get("1")).toMatchObject({ status: "active", workflow: { currentState: "work" } }); + }); + + it("admits a paused terminal transition through a trusted provider", async () => { + const definition = JSON.stringify({ + version: 1, + initialState: "work", + states: { + work: { prompt: "Check release.", on: { blocked: "blocked" } }, + blocked: { prompt: "Report blocker.", terminal: "paused" }, + }, + }); + h.admissionProviders.push({ + id: "test", + sourceClass: "environmental", + async observe({ claim, context, now }) { + return [{ + fact: claim.fact, + actual: true, + sourceClass: "environmental", + provider: "test", + providerVersion: "1", + observedAt: now, + expiresAt: now + 1_000, + context, + status: "observed", + }]; + }, + }); + await h.text("WorkflowCreate", { goal: "Check release", definition }); + + const admitted = await h.text("WorkflowTransition", { + id: "1", + outcome: "blocked", + claim: { class: "environmental", provider: "test", subject: "release", fact: "failed", expected: true }, + }); + + expect(admitted).toContain("Workflow #1 paused"); + expect(h.store.get("1")?.pause?.kind).toBe("semantic_terminal"); + const listed = await h.text("LoopList", {}); + expect(listed).toContain("Pause cause: semantic_terminal"); + expect(listed).toContain("Admission: environmental · test:release.failed = true · test@1"); + }); + it("exposes typed workflow revision changes without replacement or ownership fields", () => { const revise = h.toolMap.get("WorkflowRevise") as any; @@ -737,7 +798,11 @@ describe("Workflow tools", () => { const create = h.toolMap.get("WorkflowCreate") as any; expect(create.description).toContain("embedded atomically"); expect(create.promptGuidelines.join("\n")).toContain("maxAttempts"); - expect((h.toolMap.get("WorkflowTransition") as any).parameters.properties.evidence).toBeDefined(); + const transition = h.toolMap.get("WorkflowTransition") as any; + expect(transition.parameters.properties.evidence).toBeDefined(); + expect(transition.parameters.properties.claim).toBeDefined(); + expect(transition.parameters.properties.contextDigest).toBeUndefined(); + expect(transition.parameters.properties.claim.properties.context).toBeUndefined(); expect(await h.text("WorkflowCreate", { goal: "Fix the regression", definition })).toContain("Definition revision: 1"); expect(await h.text("LoopList", {})).toContain("Transition sequence: 0"); }); @@ -760,7 +825,8 @@ describe("LoopDelete", () => { it("pauses a loop without removing it", async () => { const out = await h.text("LoopDelete", { id: "1", action: "pause" }); expect(out).toBe("Loop #1 paused"); - expect(h.store.get("1")?.status).toBe("paused"); + expect(h.store.get("1")).toMatchObject({ status: "paused", pause: { kind: "administrative" } }); + expect(await h.text("LoopList", {})).toContain("[pause:administrative]"); }); it("delegates orchestration cancellation before deletion", async () => { diff --git a/test/property/reducers.property.test.ts b/test/property/reducers.property.test.ts index f47adb4..dbb75de 100644 --- a/test/property/reducers.property.test.ts +++ b/test/property/reducers.property.test.ts @@ -22,7 +22,7 @@ function eventFor(command: LoopCommand, at: number): LoopReducerEvent { case "fire": return { type: "LOOP_FIRED", at, source: "system", payload: { id: "1" } }; case "pause": - return { type: "LOOP_PAUSED", at, source: "system", payload: { id: "1" } }; + return { type: "LOOP_PAUSED", at, source: "system", payload: { id: "1", kind: "administrative" } }; case "resume": return { type: "LOOP_RESUMED", at, source: "system", payload: { id: "1" } }; case "update": diff --git a/test/store.test.ts b/test/store.test.ts index a8d1ade..609f5eb 100644 --- a/test/store.test.ts +++ b/test/store.test.ts @@ -7,6 +7,15 @@ import { LoopStore } from "../src/store.js"; import type { Trigger, WorkflowRunState } from "../src/types.js"; const cronTrigger: Trigger = { type: "cron", schedule: "*/5 * * * *" }; +const trustedAdmission = { + claimClass: "environmental" as const, + provider: "test", + subject: "release", + fact: "failed", + expected: true, + observations: ["test@1"], + decidedAt: 1, +}; describe("LoopStore (in-memory)", () => { let store: LoopStore; @@ -68,7 +77,20 @@ describe("LoopStore (in-memory)", () => { store.create(cronTrigger, "test", { recurring: true }); const entry = store.pause("1"); - expect(entry!.status).toBe("paused"); + expect(entry).toMatchObject({ + status: "paused", + pause: { kind: "administrative" }, + }); + }); + + it("records controller-limit pause provenance", () => { + store.create(cronTrigger, "test", { recurring: true }); + const entry = store.pause("1", "controller_limit", "workflow fire cap reached"); + + expect(entry).toMatchObject({ + status: "paused", + pause: { kind: "controller_limit", reason: "workflow fire cap reached" }, + }); }); it("resumes loops explicitly", () => { @@ -76,10 +98,11 @@ describe("LoopStore (in-memory)", () => { store.pause("1"); const entry = store.resume("1"); - expect(entry!.status).toBe("active"); + expect(entry).toMatchObject({ status: "active" }); + expect(entry?.pause).toBeUndefined(); }); - it("atomically pauses a workflow that reaches a paused terminal state", () => { + it("rejects a paused terminal transition without trusted admission", () => { store.create({ type: "dynamic" }, "Investigate", { recurring: true, workflow: { @@ -93,9 +116,32 @@ describe("LoopStore (in-memory)", () => { }); const result = store.transitionWorkflow("1", { outcome: "blocked" }); + expect(result).toMatchObject({ applied: false, error: expect.stringContaining("require trusted admission") }); + expect(store.get("1")).toMatchObject({ status: "active", workflow: { currentState: "investigate" } }); + }); + + it("atomically pauses a workflow that reaches a paused terminal state", () => { + store.create({ type: "dynamic" }, "Investigate", { + recurring: true, + workflow: { + version: 1, + initialState: "investigate", + states: { + investigate: { prompt: "Find the blocker.", on: { blocked: "blocked" } }, + blocked: { prompt: "Report the blocker.", terminal: "paused" }, + }, + }, + }); + const result = store.transitionWorkflow("1", { outcome: "blocked", admission: trustedAdmission }); + expect(result.terminal).toBe("paused"); expect(store.resume("1")).toBeUndefined(); - expect(store.get("1")?.status).toBe("paused"); + expect(store.get("1")).toMatchObject({ + status: "paused", + pause: { kind: "semantic_terminal" }, + }); + store.pause("1", "administrative", "later duplicate pause"); + expect(store.get("1")?.pause).toMatchObject({ kind: "semantic_terminal" }); }); it("atomically removes a workflow that reaches a completed terminal state", () => { @@ -467,7 +513,65 @@ describe("LoopStore (file-backed)", () => { store1.pause("1"); const store2 = new LoopStore(filePath); - expect(store2.get("1")!.status).toBe("paused"); + expect(store2.get("1")).toMatchObject({ status: "paused", pause: { kind: "administrative" } }); + }); + + it("loads legacy paused snapshots without synthesized provenance", () => { + const store = new LoopStore(filePath); + store.create(cronTrigger, "test", { recurring: true }); + store.pause("1"); + const data = JSON.parse(readFileSync(filePath, "utf8")); + delete data.loops[0].pause; + writeFileSync(filePath, JSON.stringify(data)); + rmSync(`${filePath}.prev`, { force: true }); + + expect(new LoopStore(filePath).get("1")).toMatchObject({ status: "paused" }); + expect(new LoopStore(filePath).get("1")?.pause).toBeUndefined(); + }); + + it("fails closed on malformed persisted pause provenance", () => { + const store = new LoopStore(filePath); + store.create(cronTrigger, "test", { recurring: true }); + store.pause("1"); + const data = JSON.parse(readFileSync(filePath, "utf8")); + data.loops[0].pause.kind = "invented"; + writeFileSync(filePath, JSON.stringify(data)); + rmSync(`${filePath}.prev`, { force: true }); + + expect(() => new LoopStore(filePath)).toThrow("Corrupt store"); + }); + + it("fails closed on malformed persisted admission provenance", () => { + const store = new LoopStore(filePath); + store.create({ type: "dynamic" }, "Investigate", { + recurring: true, + workflow: { + version: 1, + initialState: "work", + states: { + work: { prompt: "Work.", on: { blocked: "blocked" } }, + blocked: { prompt: "Blocked.", terminal: "paused" }, + }, + }, + }); + store.transitionWorkflow("1", { + outcome: "blocked", + admission: { + claimClass: "environmental", + provider: "monitor", + subject: "m1", + fact: "status", + expected: "error", + observations: ["monitor@1"], + decidedAt: Date.now(), + }, + }); + const data = JSON.parse(readFileSync(filePath, "utf8")); + data.loops[0].workflow.lastTransition.admission.observations = Array.from({ length: 9 }, () => "monitor@1"); + writeFileSync(filePath, JSON.stringify(data)); + rmSync(`${filePath}.prev`, { force: true }); + + expect(() => new LoopStore(filePath)).toThrow("Corrupt store"); }); it("persists deletions", () => { diff --git a/test/workflow-admission-providers.test.ts b/test/workflow-admission-providers.test.ts new file mode 100644 index 0000000..1e70524 --- /dev/null +++ b/test/workflow-admission-providers.test.ts @@ -0,0 +1,64 @@ +import { describe, expect, it } from "vitest"; +import { createMonitorWorkflowAdmissionProvider } from "../src/runtime/workflow-admission-providers.js"; +import type { MonitorEntry } from "../src/types.js"; +import type { WorkflowAdmissionContext, WorkflowBlockerClaim } from "../src/workflow-admission.js"; + +const NOW = 1_800_000_000_000; +const context: WorkflowAdmissionContext = { + workflowId: "1", + currentState: "validate", + transitionSeq: 0, + definitionRevision: 1, + contextDigest: "workspace-A", +}; + +function claim(fact: string): WorkflowBlockerClaim { + return { class: "environmental", provider: "monitor", subject: "monitor-1", fact, expected: true }; +} + +function monitor(): MonitorEntry { + return { + id: "monitor-1", + command: "npm test", + timeout: 1_000, + status: "error", + startedAt: NOW - 100, + completedAt: NOW, + exitCode: 1, + outputLines: 1, + outputBuffer: ["secret output must not become admission evidence"], + }; +} + +describe("monitor workflow admission provider", () => { + it.each([ + ["status", "error"], + ["exitCode", 1], + ["stopReason", null], + ])("exposes bounded %s facts", async (fact, expected) => { + const provider = createMonitorWorkflowAdmissionProvider(() => monitor()); + + const observations = await provider.observe({ claim: claim(fact), context, now: NOW }); + + expect(observations).toEqual([expect.objectContaining({ + fact, + actual: expected, + provider: "monitor", + providerVersion: "1", + status: "observed", + context, + })]); + expect(JSON.stringify(observations)).not.toContain("secret output"); + }); + + it.each([ + ["missing monitor", () => undefined, "status"], + ["unsupported fact", () => monitor(), "output"], + ])("abstains for %s", async (_label, getMonitor, fact) => { + const provider = createMonitorWorkflowAdmissionProvider(getMonitor); + + const observations = await provider.observe({ claim: claim(fact), context, now: NOW }); + + expect(observations).toEqual([expect.objectContaining({ status: "abstained" })]); + }); +}); diff --git a/test/workflow-admission.test.ts b/test/workflow-admission.test.ts new file mode 100644 index 0000000..d7e1775 --- /dev/null +++ b/test/workflow-admission.test.ts @@ -0,0 +1,430 @@ +import { mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { LoopStore } from "../src/store.js"; +import { TaskStore } from "../src/task-store.js"; +import { + admitWorkflowTransition, + reconcileWorkflowClaim, + type WorkflowAdmissionContext, + type WorkflowAdmissionObservation, + type WorkflowAdmissionProvider, + type WorkflowBlockerClaim, +} from "../src/workflow-admission.js"; + +const NOW = 1_800_000_000_000; +const directories: string[] = []; + +function createStore() { + const directory = mkdtempSync(join(tmpdir(), "pi-loop-workflow-admission-")); + directories.push(directory); + const path = join(directory, "loops.json"); + const store = new LoopStore(path); + const entry = store.create({ type: "dynamic" }, "Validate release", { + recurring: true, + workflow: { + version: 1, + initialState: "validate", + states: { + validate: { + prompt: "Validate release.", + on: { continue: "ship", race: "review", blocked: "blocked" }, + }, + review: { prompt: "Review.", on: { continue: "ship" } }, + ship: { prompt: "Ship.", terminal: "completed" }, + blocked: { prompt: "Report blocker.", terminal: "paused" }, + }, + }, + }); + return { path, store, entry }; +} + +function context(entry: ReturnType, contextDigest = "workspace-A"): WorkflowAdmissionContext { + if (!entry?.workflow) throw new Error("expected workflow"); + return { + workflowId: entry.id, + currentState: entry.workflow.currentState, + transitionSeq: entry.workflow.transitionSeq, + definitionRevision: entry.workflow.definitionRevision, + activeExecutionId: entry.workflow.activeExecution?.id, + contextDigest, + }; +} + +function claim(): WorkflowBlockerClaim { + return { + class: "environmental", + provider: "test", + subject: "release-check", + fact: "failed", + expected: true, + }; +} + +function observed( + scoped: WorkflowAdmissionContext, + actual: boolean | number = true, + provider = "test", +): WorkflowAdmissionObservation { + return { + fact: "failed", + actual, + sourceClass: "environmental", + provider, + providerVersion: "1", + observedAt: NOW, + expiresAt: NOW + 1_000, + context: scoped, + status: "observed", + }; +} + +function provider(observations: (input: { context: WorkflowAdmissionContext }) => WorkflowAdmissionObservation[]): WorkflowAdmissionProvider { + return { + id: "test", + sourceClass: "environmental", + observe: vi.fn(async (input) => observations(input)), + }; +} + +afterEach(() => { + for (const directory of directories.splice(0)) rmSync(directory, { recursive: true, force: true }); +}); + +describe("workflow transition admission", () => { + it("allows ordinary declared transitions without blocker admission", async () => { + const { store, entry } = createStore(); + const result = await admitWorkflowTransition({ + store, + workflowId: entry.id, + outcome: "continue", + contextDigest: "workspace-A", + providers: [], + now: NOW, + }); + + expect(result).toMatchObject({ + decision: { decision: "not_required" }, + transition: { applied: true, terminal: "completed" }, + }); + }); + + it("requires a grounded claim before entering a paused terminal state", async () => { + const { path, store, entry } = createStore(); + const before = readFileSync(path); + const result = await admitWorkflowTransition({ + store, + workflowId: entry.id, + outcome: "blocked", + contextDigest: "workspace-A", + providers: [], + now: NOW, + }); + + expect(result).toMatchObject({ decision: { decision: "unresolved", reason: "claim_required" } }); + expect(result.transition).toBeUndefined(); + expect(readFileSync(path)).toEqual(before); + }); + + it("runs trusted providers before the CAS transition and records semantic pause provenance", async () => { + const { store, entry } = createStore(); + const trusted = provider(({ context: scoped }) => { + expect(store.get(entry.id)?.status).toBe("active"); + return [observed(scoped)]; + }); + const result = await admitWorkflowTransition({ + store, + workflowId: entry.id, + outcome: "blocked", + claim: claim(), + contextDigest: "workspace-A", + providers: [trusted], + now: NOW, + }); + + expect(trusted.observe).toHaveBeenCalledOnce(); + expect(result).toMatchObject({ decision: { decision: "confirmed" }, transition: { applied: true, terminal: "paused" } }); + expect(store.get(entry.id)).toMatchObject({ + pause: { kind: "semantic_terminal" }, + workflow: { + lastTransition: { + admission: { + claimClass: "environmental", + provider: "test", + subject: "release-check", + fact: "failed", + expected: true, + observations: ["test@1"], + decidedAt: NOW, + }, + }, + }, + }); + }); + + it("uses one equality relation for conflicts and final comparison", () => { + const { entry } = createStore(); + const scoped = context(entry); + expect(reconcileWorkflowClaim(claim(), [observed(scoped, 0), observed(scoped, -0)], scoped, NOW)) + .toMatchObject({ decision: "unresolved", reason: "conflicting_observations" }); + }); + + it("rejects stale provider context and preserves the workflow bytes", async () => { + const { path, store, entry } = createStore(); + const before = readFileSync(path); + const stale = provider(({ context: scoped }) => [observed({ ...scoped, contextDigest: "workspace-B" })]); + const result = await admitWorkflowTransition({ + store, + workflowId: entry.id, + outcome: "blocked", + claim: claim(), + contextDigest: "workspace-A", + providers: [stale], + now: NOW, + }); + + expect(result).toMatchObject({ decision: { decision: "unresolved", reason: "no_current_observation" } }); + expect(result.transition).toBeUndefined(); + expect(readFileSync(path)).toEqual(before); + }); + + it("never treats environmental providers as user authority", async () => { + const { path, store, entry } = createStore(); + const before = readFileSync(path); + const authorityClaim: WorkflowBlockerClaim = { ...claim(), class: "user_authority" }; + const result = await admitWorkflowTransition({ + store, + workflowId: entry.id, + outcome: "blocked", + claim: authorityClaim, + contextDigest: "workspace-A", + providers: [provider(({ context: scoped }) => [observed(scoped)])], + now: NOW, + }); + + expect(result).toMatchObject({ decision: { decision: "requires_user_authority", reason: "provider_unavailable" } }); + expect(result.transition).toBeUndefined(); + expect(readFileSync(path)).toEqual(before); + }); + + it("rechecks expiry after provider execution", async () => { + const { path, store, entry } = createStore(); + const before = readFileSync(path); + let clockReads = 0; + const expiring = provider(({ context: scoped }) => [observed(scoped)]); + const result = await admitWorkflowTransition({ + store, + workflowId: entry.id, + outcome: "blocked", + claim: claim(), + contextDigest: "workspace-A", + providers: [expiring], + now: () => clockReads++ === 0 ? NOW : NOW + 2_000, + }); + + expect(result).toMatchObject({ decision: { decision: "unresolved", reason: "no_current_observation" } }); + expect(result.transition).toBeUndefined(); + expect(readFileSync(path)).toEqual(before); + }); + + it("rejects oversized fact values without writing", async () => { + const { path, store, entry } = createStore(); + const before = readFileSync(path); + const result = await admitWorkflowTransition({ + store, + workflowId: entry.id, + outcome: "blocked", + claim: { ...claim(), expected: "x".repeat(1_025) }, + contextDigest: "workspace-A", + providers: [provider(({ context: scoped }) => [observed(scoped)])], + now: NOW, + }); + + expect(result).toMatchObject({ decision: { decision: "unresolved", reason: "invalid_claim" } }); + expect(result.transition).toBeUndefined(); + expect(readFileSync(path)).toEqual(before); + }); + + it("rejects contradicted, expired, abstained, and errored evidence without writing", async () => { + const cases: Array<{ + label: string; + mutate: (item: WorkflowAdmissionObservation) => WorkflowAdmissionObservation; + decision: string; + }> = [ + { label: "contradicted", mutate: (item) => ({ ...item, actual: false }), decision: "contradicted" }, + { label: "expired", mutate: (item) => ({ ...item, expiresAt: NOW - 1 }), decision: "unresolved" }, + { label: "abstained", mutate: (item) => ({ ...item, status: "abstained" }), decision: "unresolved" }, + { label: "error", mutate: (item) => ({ ...item, status: "error" }), decision: "unresolved" }, + ]; + + for (const testCase of cases) { + const { path, store, entry } = createStore(); + const before = readFileSync(path); + const resolver = provider(({ context: scoped }) => [testCase.mutate(observed(scoped))]); + const result = await admitWorkflowTransition({ + store, + workflowId: entry.id, + outcome: "blocked", + claim: claim(), + contextDigest: "workspace-A", + providers: [resolver], + now: NOW, + }); + + expect(result.decision.decision, testCase.label).toBe(testCase.decision); + expect(result.transition, testCase.label).toBeUndefined(); + expect(readFileSync(path), testCase.label).toEqual(before); + } + }); + + it("rejects replay across workflow, state, revision, execution, and workspace boundaries", async () => { + const mutations: Array<{ + label: string; + mutate: (item: WorkflowAdmissionContext) => WorkflowAdmissionContext; + }> = [ + { label: "workflow", mutate: (item) => ({ ...item, workflowId: "other-workflow" }) }, + { label: "state", mutate: (item) => ({ ...item, currentState: "other-state" }) }, + { label: "revision", mutate: (item) => ({ ...item, definitionRevision: item.definitionRevision + 1 }) }, + { label: "execution", mutate: (item) => ({ ...item, activeExecutionId: "other-execution" }) }, + { label: "workspace", mutate: (item) => ({ ...item, contextDigest: "workspace-B" }) }, + ]; + + for (const replay of mutations) { + const { path, store, entry } = createStore(); + const before = readFileSync(path); + const stale = provider(({ context: scoped }) => [observed(replay.mutate(scoped))]); + const result = await admitWorkflowTransition({ + store, + workflowId: entry.id, + outcome: "blocked", + claim: claim(), + contextDigest: "workspace-A", + providers: [stale], + now: NOW, + }); + + expect(result, replay.label).toMatchObject({ decision: { decision: "unresolved", reason: "no_current_observation" } }); + expect(result.transition, replay.label).toBeUndefined(); + expect(readFileSync(path), replay.label).toEqual(before); + } + }); + + it("resubmits after file-backed store recreation without pending proposal state", async () => { + const { path, store, entry } = createStore(); + const taskPath = join(path, "..", "tasks.json"); + const taskStore = new TaskStore(taskPath); + taskStore.create({ subject: "Independent", description: "Must remain untouched" }); + const taskBytes = readFileSync(taskPath); + const waiting = await admitWorkflowTransition({ + store, + workflowId: entry.id, + outcome: "blocked", + claim: claim(), + contextDigest: "workspace-A", + providers: [], + now: NOW, + }); + expect(waiting).toMatchObject({ decision: { decision: "unresolved", reason: "provider_unavailable" } }); + + const restarted = new LoopStore(path); + const fresh = provider(({ context: scoped }) => [observed(scoped)]); + const resumed = await admitWorkflowTransition({ + store: restarted, + workflowId: entry.id, + outcome: "blocked", + claim: claim(), + contextDigest: "workspace-A", + providers: [fresh], + now: NOW, + }); + + expect(resumed).toMatchObject({ decision: { decision: "confirmed" }, transition: { applied: true } }); + expect(readFileSync(taskPath)).toEqual(taskBytes); + }); + + it("admits user authority only through a matching trusted authority provider", async () => { + const { store, entry } = createStore(); + const authorityClaim: WorkflowBlockerClaim = { + class: "user_authority", + provider: "approval", + subject: "release", + fact: "approved", + expected: true, + }; + const authority: WorkflowAdmissionProvider = { + id: "approval", + sourceClass: "user_authority", + async observe({ claim: requested, context: scoped, now }) { + return [{ + fact: requested.fact, + actual: true, + sourceClass: "user_authority", + provider: "approval", + providerVersion: "1", + observedAt: now, + expiresAt: now + 1_000, + context: scoped, + status: "observed", + }]; + }, + }; + const result = await admitWorkflowTransition({ + store, + workflowId: entry.id, + outcome: "blocked", + claim: authorityClaim, + contextDigest: "workspace-A", + providers: [authority], + now: NOW, + }); + + expect(result).toMatchObject({ decision: { decision: "confirmed" }, transition: { applied: true } }); + }); + + it("rejects a provider result after the runtime context changes", async () => { + const { path, store, entry } = createStore(); + const before = readFileSync(path); + let current = true; + const switching = provider(({ context: scoped }) => { + current = false; + return [observed(scoped)]; + }); + const result = await admitWorkflowTransition({ + store, + workflowId: entry.id, + outcome: "blocked", + claim: claim(), + contextDigest: "workspace-A", + providers: [switching], + now: NOW, + isContextCurrent: () => current, + }); + + expect(result).toMatchObject({ decision: { decision: "unresolved", reason: "stale_runtime_context" } }); + expect(result.transition).toBeUndefined(); + expect(readFileSync(path)).toEqual(before); + }); + + it("rejects a confirmed claim when a competing transition wins the CAS race", async () => { + const { store, entry } = createStore(); + const racing = provider(({ context: scoped }) => { + expect(store.transitionWorkflow(entry.id, { outcome: "race" })).toMatchObject({ applied: true }); + return [observed(scoped)]; + }); + const result = await admitWorkflowTransition({ + store, + workflowId: entry.id, + outcome: "blocked", + claim: claim(), + contextDigest: "workspace-A", + providers: [racing], + now: NOW, + }); + + expect(result).toMatchObject({ + decision: { decision: "confirmed" }, + transition: { applied: false, error: expect.stringContaining("changed") }, + }); + }); +});