diff --git a/.github/tests/test_detached_supervision.py b/.github/tests/test_detached_supervision.py index cd27eda..15ca6f9 100644 --- a/.github/tests/test_detached_supervision.py +++ b/.github/tests/test_detached_supervision.py @@ -90,6 +90,24 @@ def run_helper( def helper_json(self, *args: object, cwd: Path | None = None) -> dict: return json.loads(self.run_helper(*args, cwd=cwd).stdout) + def apply_prescribed( + self, transition: str, *args: object, cwd: Path | None = None + ) -> dict: + resolved = self.helper_json( + "next", "--transition", transition, *args, cwd=cwd + ) + prescription = resolved["prescription"] + correlation = resolved["snapshot"]["invocation"]["correlation_id"] + return self.helper_json( + "apply", "--transition", transition, *args, + "--correlation", correlation, + "--prescription-id", prescription["id"], + "--expected-state-revision", prescription["expected_state_revision"], + "--expected-program-fingerprint", prescription["expected_program_fingerprint"], + "--expected-snapshot-fingerprint", prescription["expected_snapshot_fingerprint"], + cwd=cwd, + ) + def porcelain(self, repository: Path | None = None) -> str: repository = repository or self.repo return subprocess.run( @@ -198,13 +216,13 @@ def test_detached_installation_and_engaged_guard_use_the_same_kernel(self) -> No ), ) - self.helper_json( - "apply", "--repo", self.repo, "--transition", "goal.configure", + self.apply_prescribed( + "goal.configure", "--repo", self.repo, *self.goal_flags(), "--human", "contract", "--param", "goal_kind=approved-plan", "--param", "delivery_id=bootstrap", ) - self.helper_json( - "apply", "--repo", self.repo, "--transition", "engagement.begin", + self.apply_prescribed( + "engagement.begin", "--repo", self.repo, *self.goal_flags(), "--repository-authority", ) ordinary = self.helper_json( @@ -249,14 +267,14 @@ def test_authority_free_frontier_does_not_block_authorized_plan_creation(self) - "init", "--repo", self.repo, *goal, *flow, "--human", "contract", "--param", f"config_path={config}", ) - self.helper_json( - "apply", "--repo", self.repo, "--transition", "goal.configure", + self.apply_prescribed( + "goal.configure", "--repo", self.repo, *goal, *flow, "--human", "contract", "--param", "goal_kind=open-or-updated-pr", "--param", "delivery_id=codex-driver-authority-triggers", ) - self.helper_json( - "apply", "--repo", self.repo, "--transition", "engagement.begin", + self.apply_prescribed( + "engagement.begin", "--repo", self.repo, *goal, *flow, "--repository-authority", ) @@ -377,8 +395,8 @@ def test_one_delivery_context_rematerializes_repository_authority_after_initiali for field in ('"admission"', '"receipt"', '"snapshot"', '"target_fingerprint"', '"recovery"'): self.assertIn(field, initialized_process.stdout) - configured = self.helper_json( - "apply", "--repo", self.repo, "--transition", "goal.configure", + configured = self.apply_prescribed( + "goal.configure", "--repo", self.repo, *goal, *flow, *actor, "--param", "goal_kind=open-or-updated-pr", "--param", "delivery_id=preserve-repository-authority-context", @@ -392,19 +410,19 @@ def test_one_delivery_context_rematerializes_repository_authority_after_initiali self.assertEqual(engagement["decision"]["kind"], "PRESCRIBED") self.assertEqual(engagement["decision"]["transition"]["id"], "engagement.begin") - engaged_process = self.run_helper( - "apply", "--repo", self.repo, "--transition", "engagement.begin", + engaged = self.apply_prescribed( + "engagement.begin", "--repo", self.repo, *goal, *flow, *actor, "--repository-authority", ) - engaged = json.loads(engaged_process.stdout) self.assertEqual(engaged["receipt"]["transition_id"], "engagement.begin") self.assertEqual(engaged["receipt"]["flow_id"], flow[1]) self.assertEqual( {receipt["class"] for receipt in engaged["admission"]["authority"]["receipts"]}, {"human", "repository-policy"}, ) + engaged_output = json.dumps(engaged) for field in ('"admission"', '"receipt"', '"snapshot"', '"target_fingerprint"', '"recovery"'): - self.assertIn(field, engaged_process.stdout) + self.assertIn(field, engaged_output) plan = self.helper_json( "next", "--repo", self.repo, *goal, *flow, *actor, diff --git a/.github/tests/test_repository_contract.py b/.github/tests/test_repository_contract.py index 7d90921..b62077c 100644 --- a/.github/tests/test_repository_contract.py +++ b/.github/tests/test_repository_contract.py @@ -79,6 +79,33 @@ def run_helper( self.helper, *args, env=env, stdin=stdin, expected=expected ) + def apply_prescribed( + self, + binary: Path, + transition: str, + *args: object, + cwd: Path | None = None, + env: dict[str, str] | None = None, + ) -> dict: + resolved = json.loads( + self.run_command( + binary, "next", "--transition", transition, *args, + cwd=cwd, env=env, + ).stdout + ) + prescription = resolved["prescription"] + correlation = resolved["snapshot"]["invocation"]["correlation_id"] + applied = self.run_command( + binary, "apply", "--transition", transition, *args, + "--correlation", correlation, + "--prescription-id", prescription["id"], + "--expected-state-revision", prescription["expected_state_revision"], + "--expected-program-fingerprint", prescription["expected_program_fingerprint"], + "--expected-snapshot-fingerprint", prescription["expected_snapshot_fingerprint"], + cwd=cwd, env=env, + ) + return json.loads(applied.stdout) + def init_repository(self, root: Path) -> None: self.run_command("git", "init", "-b", "main", cwd=root) self.run_command("git", "config", "user.name", "Boatstack Test", cwd=root) @@ -494,15 +521,13 @@ def test_offline_installer_initializes_updates_and_guards_through_kernel(self) - "--goal-id", "bootstrap", "--goal-kind", "approved-plan", "--delivery", "bootstrap", ) - self.run_command( - launcher, "apply", "--repo", repository, - "--transition", "goal.configure", *goal, + self.apply_prescribed( + launcher, "goal.configure", "--repo", repository, *goal, "--human", "contract", "--param", "goal_kind=approved-plan", "--param", "delivery_id=bootstrap", env=env, ) - self.run_command( - launcher, "apply", "--repo", repository, - "--transition", "engagement.begin", *goal, + self.apply_prescribed( + launcher, "engagement.begin", "--repo", repository, *goal, "--repository-authority", env=env, ) ordinary = json.loads( @@ -594,7 +619,7 @@ def test_program_changing_update_is_explicit_atomic_and_dormant_safe(self) -> No prior_program = candidate_status["snapshot"]["recorded_program_fingerprint"] split_reconciliation = self.run_command( self.helper, - "apply", + "next", "--repo", repository, "--transition", @@ -612,7 +637,6 @@ def test_program_changing_update_is_explicit_atomic_and_dormant_safe(self) -> No "--param", "accept_obligation_change=true", env=env, - expected=1, ) self.assertIn( "catalog reconciliation cannot activate a different runtime", diff --git a/README.md b/README.md index 5a03013..f8ee895 100644 --- a/README.md +++ b/README.md @@ -36,6 +36,9 @@ the complete contract and the historical failure synthesis. The [Control Program ABI](docs/architecture/control-program-abi.md) defines the strict repository source, canonical fingerprint, compatibility gate, and program-qualified transition identity used by complete user-facing Flows. +The [prescription transaction boundary](docs/architecture/prescription-transactions.md) +defines the exact durable-state and executable-program compare-and-swap contract +between resolution and effects. ## Install @@ -72,12 +75,17 @@ See [Getting started](docs/getting-started.md) and ```sh boatstack status --repo . --format json boatstack catalog --format json -boatstack apply --repo . --transition --format json +boatstack next --repo . --goal-id --goal-kind --delivery --format json +boatstack apply --repo . --transition --flow \ + --prescription-id --expected-state-revision \ + --expected-program-fingerprint \ + --expected-snapshot-fingerprint --format json ``` - `status`, `next`, `doctor`, `catalog`, and `events` are read-only. -- `apply` and `recover` request stable transition IDs from the 63-event - executable catalog. +- `apply` and `recover` consume a stable transition ID plus the exact + prescription returned by `next`; stale state or program identity causes zero + effects and requires re-resolution. - Friendly aliases such as `plan-create`, `plan-approve`, `workspace-cut`, `record-test`, and `publish-pr` map to those IDs. - `guard` is the shared safety-hook query. It blocks high-confidence diff --git a/boatstack/cmd/boatstack-helper/main.go b/boatstack/cmd/boatstack-helper/main.go index 80f52fc..60bc00b 100644 --- a/boatstack/cmd/boatstack-helper/main.go +++ b/boatstack/cmd/boatstack-helper/main.go @@ -23,7 +23,6 @@ import ( "github.com/operatorstack/boatstack/boatstack/internal/kernel/catalog" "github.com/operatorstack/boatstack/boatstack/internal/kernel/model" "github.com/operatorstack/boatstack/boatstack/internal/kernel/protocol" - "github.com/operatorstack/boatstack/boatstack/internal/kernel/supervisor" boatstackruntime "github.com/operatorstack/boatstack/boatstack/internal/runtime" "github.com/operatorstack/boatstack/boatstack/internal/surfaces" ) @@ -37,22 +36,27 @@ func (s *stringList) Set(value string) error { } type commandOptions struct { - repository string - format string - goalID string - goalKind string - deliveryID string - flowID string - transitionID string - idempotencyKey string - humanActor string - repositoryPolicy bool - acceptProgramChange bool - parameters stringList - authorityReceipts stringList - follow bool - host string - command string + repository string + format string + goalID string + goalKind string + deliveryID string + flowID string + transitionID string + correlationID string + prescriptionID string + expectedStateRevision uint64 + expectedProgramFingerprint string + expectedSnapshotFingerprint string + idempotencyKey string + humanActor string + repositoryPolicy bool + acceptProgramChange bool + parameters stringList + authorityReceipts stringList + follow bool + host string + command string } func main() { @@ -101,13 +105,27 @@ func run(arguments []string) error { if err != nil { return err } - response, handleErr := kernel.Handle(context.Background(), request) - if command == "update" && options.acceptProgramChange && handleErr != nil && response.ProgramChange != nil && response.Decision != nil && - response.Decision.Kind == supervisor.DecisionUnresolved && response.Decision.Reason == supervisor.ReasonProgramDrift { - request.TransitionID = "installation.reconcile-update" - request.Parameters = append(request.Parameters, protocol.Parameter{Name: "accept_obligation_change", Value: "true"}).Canonical() - response, handleErr = kernel.Handle(context.Background(), request) + if (operation == surfaces.OperationApply || operation == surfaces.OperationRecover) && request.Prescription.ID == "" && command != "apply" && command != "recover" { + resolveRequest := request + resolveRequest.Operation = surfaces.OperationResolve + resolveRequest.FlowID = "" + resolveRequest.Prescription = protocol.Prescription{} + resolved, resolveErr := kernel.Handle(context.Background(), resolveRequest) + if resolveErr != nil || resolved.Prescription == nil { + if renderErr := renderResponse(resolved, options.format); renderErr != nil { + return renderErr + } + if resolveErr == nil { + if resolved.Decision != nil && resolved.Decision.Reason != "" { + return errors.New(resolved.Decision.Reason) + } + return fmt.Errorf("transition %q was not prescribed", request.TransitionID) + } + return resolveErr + } + request.Prescription = *resolved.Prescription } + response, handleErr := kernel.Handle(context.Background(), request) if command == "events" && options.follow { if options.format != "jsonl" { return fmt.Errorf("events --follow requires --format jsonl") @@ -229,6 +247,11 @@ func parseOptions(command string, arguments []string, transition catalog.Transit flags.StringVar(&options.deliveryID, "delivery", options.deliveryID, "delivery identity") flags.StringVar(&options.flowID, "flow", "", "flow identity") flags.StringVar(&options.transitionID, "transition", options.transitionID, "stable semantic transition id") + flags.StringVar(&options.correlationID, "correlation", "", "command-scoped correlation identity from resolution") + flags.StringVar(&options.prescriptionID, "prescription-id", "", "exact prescription identity from resolution") + flags.Uint64Var(&options.expectedStateRevision, "expected-state-revision", 0, "exact durable state revision observed during resolution") + flags.StringVar(&options.expectedProgramFingerprint, "expected-program-fingerprint", "", "exact executable control-program fingerprint observed during resolution") + flags.StringVar(&options.expectedSnapshotFingerprint, "expected-snapshot-fingerprint", "", "exact admission-relevant snapshot fingerprint observed during resolution") flags.StringVar(&options.idempotencyKey, "idempotency-key", "", "exact prior admission idempotency key for safe replay") flags.StringVar(&options.humanActor, "human", "", "explicit command-scoped human authority actor") flags.BoolVar(&options.repositoryPolicy, "repository-authority", false, "derive repository-policy authority from the V2 project configuration") @@ -253,10 +276,11 @@ func parseOptions(command string, arguments []string, transition catalog.Transit if err := populateRuntimeParameters(&options); err != nil { return commandOptions{}, err } - if command == "reconcile-update" { + if command == "reconcile-update" || (command == "update" && options.acceptProgramChange) { if !options.acceptProgramChange { return commandOptions{}, fmt.Errorf("reconcile-update requires explicit --accept-program-change") } + options.transitionID = "installation.reconcile-update" options.parameters = append(options.parameters, "accept_obligation_change=true") } case "correct-pr": @@ -405,7 +429,10 @@ func buildRevision() string { func buildRequest(operation surfaces.Operation, options commandOptions) (surfaces.Request, error) { now := time.Now().UTC() - correlation := fmt.Sprintf("cli-%d-%d", os.Getpid(), now.UnixNano()) + correlation := options.correlationID + if correlation == "" { + correlation = fmt.Sprintf("cli-%d-%d", os.Getpid(), now.UnixNano()) + } goal := model.Goal{} if options.goalKind != "" || options.goalID != "" || options.deliveryID != "" { goal = model.Goal{ID: options.goalID, Kind: model.GoalKind(options.goalKind), DeliveryID: options.deliveryID} @@ -431,6 +458,9 @@ func buildRequest(operation surfaces.Operation, options commandOptions) (surface return surfaces.Request{ SchemaVersion: surfaces.SchemaVersion, Operation: operation, Repository: options.repository, Host: options.host, CorrelationID: correlation, FlowID: flowID, Goal: goal, TransitionID: catalog.TransitionID(options.transitionID), Authority: authority, Parameters: parameters, + Prescription: protocol.Prescription{SchemaVersion: protocol.PrescriptionSchemaVersion, ID: options.prescriptionID, + TransitionID: catalog.TransitionID(options.transitionID), ExpectedStateRevision: options.expectedStateRevision, + ExpectedProgramFingerprint: options.expectedProgramFingerprint, ExpectedSnapshotFingerprint: options.expectedSnapshotFingerprint}, RepositoryAuthority: options.repositoryPolicy, IdempotencyKey: options.idempotencyKey, Command: options.command, }, nil } @@ -568,6 +598,15 @@ func renderResponse(response surfaces.Response, format string) error { fmt.Println("transition:", response.Decision.Transition.ID) } } + if response.Prescription != nil { + correlation := "" + if response.Snapshot != nil { + correlation = response.Snapshot.Invocation.Correlation + } + fmt.Printf("prescription=%s state_revision=%d program=%s snapshot=%s correlation=%s\n", response.Prescription.ID, + response.Prescription.ExpectedStateRevision, response.Prescription.ExpectedProgramFingerprint, + response.Prescription.ExpectedSnapshotFingerprint, correlation) + } if response.Receipt != nil { fmt.Println("receipt:", response.Receipt.ID) } diff --git a/boatstack/flow/standard/historical_test.go b/boatstack/flow/standard/historical_test.go index b51256d..179a46a 100644 --- a/boatstack/flow/standard/historical_test.go +++ b/boatstack/flow/standard/historical_test.go @@ -79,7 +79,7 @@ func snapshotFromFixture(t *testing.T, fixture historicalFixture) model.Snapshot InvokingPath: filepath.Join(t.TempDir(), "fixture", "repository"), Topology: model.Topology(facts["topology"]), Host: "corpus", Correlation: "correlation-" + fixture.Name, } observation := model.Observation{ - SchemaVersion: model.SnapshotSchemaVersion, Invocation: invocation, + SchemaVersion: model.SnapshotSchemaVersion, StateRevision: 1, Invocation: invocation, Phase: model.Known(model.ProtocolPhase(facts["phase"]), evidence), Engagement: model.Known(model.EngagementState(facts["engagement"]), evidence), Delivery: model.Known(model.DeliveryState(facts["delivery"]), evidence), Workspace: model.Known(model.WorkspaceState(facts["workspace"]), evidence), Plan: model.Known(model.PlanState(facts["plan"]), evidence), Configuration: model.Known(model.ConfigurationState(facts["configuration"]), evidence), diff --git a/boatstack/flow/standard/supervisor_parity_test.go b/boatstack/flow/standard/supervisor_parity_test.go index c02cb45..b774daa 100644 --- a/boatstack/flow/standard/supervisor_parity_test.go +++ b/boatstack/flow/standard/supervisor_parity_test.go @@ -29,9 +29,9 @@ func snapshotFor(t *testing.T, phase model.ProtocolPhase, terminal model.Termina t.Helper() e := model.Evidence{Source: "fixture", Fingerprint: "fixture", ObservedAt: time.Unix(10, 0).UTC()} o := model.Observation{ - SchemaVersion: model.SnapshotSchemaVersion, - Invocation: model.InvocationContext{RepositoryID: "repo", GitCommonID: "git", WorktreeID: "wt", Ref: "refs/heads/f", ControllerID: "ctl", InvokingPath: filepath.Join(t.TempDir(), "repo"), RuntimeVersion: "runtime-version", RuntimePath: filepath.Join(t.TempDir(), "runtime"), RuntimeFingerprint: "runtime", Topology: model.TopologyEmbedded, Host: "cli", Correlation: "c"}, - Phase: model.Known(phase, e), Engagement: model.Known(model.EngagementActive, e), Delivery: model.Known(model.DeliveryActive, e), + SchemaVersion: model.SnapshotSchemaVersion, StateRevision: 1, + Invocation: model.InvocationContext{RepositoryID: "repo", GitCommonID: "git", WorktreeID: "wt", Ref: "refs/heads/f", ControllerID: "ctl", InvokingPath: filepath.Join(t.TempDir(), "repo"), RuntimeVersion: "runtime-version", RuntimePath: filepath.Join(t.TempDir(), "runtime"), RuntimeFingerprint: "runtime", Topology: model.TopologyEmbedded, Host: "cli", Correlation: "c"}, + Phase: model.Known(phase, e), Engagement: model.Known(model.EngagementActive, e), Delivery: model.Known(model.DeliveryActive, e), Workspace: model.Known(model.WorkspaceActive, e), Plan: model.Known(model.PlanValid, e), Configuration: model.Known(model.ConfigurationVerified, e), Runtime: model.Known(model.RuntimeVerified, e), ConfigurationPolicy: model.Known(model.ConfigurationPolicy{PlanApproval: "human", VisualEvidence: "optional", ExternalEffectAuthority: "human-or-autonomy-plus-provider", Hosts: []string{"cli"}}, e), diff --git a/boatstack/internal/effects/cas_integration_test.go b/boatstack/internal/effects/cas_integration_test.go new file mode 100644 index 0000000..9d793f7 --- /dev/null +++ b/boatstack/internal/effects/cas_integration_test.go @@ -0,0 +1,204 @@ +package effects_test + +import ( + "bytes" + "context" + "errors" + "os" + "path/filepath" + "strings" + "testing" + "time" + + boatstack "github.com/operatorstack/boatstack/boatstack" + "github.com/operatorstack/boatstack/boatstack/control" + "github.com/operatorstack/boatstack/boatstack/core" + "github.com/operatorstack/boatstack/boatstack/extension/releasenote" + "github.com/operatorstack/boatstack/boatstack/flow/standard" + "github.com/operatorstack/boatstack/boatstack/internal/kernel/catalog" + "github.com/operatorstack/boatstack/boatstack/internal/kernel/durable" + "github.com/operatorstack/boatstack/boatstack/internal/kernel/engine" + "github.com/operatorstack/boatstack/boatstack/internal/kernel/model" + "github.com/operatorstack/boatstack/boatstack/internal/kernel/protocol" + "github.com/operatorstack/boatstack/boatstack/internal/plant" + "github.com/operatorstack/boatstack/boatstack/internal/surfaces" +) + +type concurrentApplyResult struct { + response surfaces.Response + err error +} + +func TestConcurrentApplyConsumesOneRevisionExactlyOnce(t *testing.T) { + // control-law: one prescribed durable revision has at most one successful committer + ctx := context.Background() + repository := testRepository(t) + externalRoot := t.TempDir() + program := testProgram() + kernelA, err := boatstack.NewKernel(externalRoot, program) + if err != nil { + t.Fatal(err) + } + kernelB, err := boatstack.NewKernel(externalRoot, program) + if err != nil { + t.Fatal(err) + } + executable, _ := os.Executable() + executable, _ = filepath.Abs(executable) + executable, _ = filepath.EvalSymlinks(executable) + runtimeRaw, _ := os.ReadFile(executable) + runtimeVersion := installTestRuntime(t, executable, runtimeRaw) + configPath := filepath.Join(t.TempDir(), "project.json") + configRaw := []byte("{\"schema_version\":2,\"project\":{\"name\":\"cas\",\"default_branch\":\"main\",\"commands\":{}},\"policy\":{\"plan_approval\":\"human\",\"visual_evidence\":\"optional\"},\"hosts\":[\"cli\"]}\n") + if err := os.WriteFile(configPath, configRaw, 0o600); err != nil { + t.Fatal(err) + } + now := time.Now().UTC() + human := protocol.AuthorityBundle{Receipts: []protocol.AuthorityReceipt{{ + ID: "cas-human", Class: catalog.AuthorityHuman, Subject: "operator", Fingerprint: "cas-human-proof", + IssuedAt: now.Add(-time.Minute), ExpiresAt: now.Add(time.Hour), + }}} + goal := model.Goal{ID: "cas-goal", Kind: model.GoalApprovedPlan, DeliveryID: "cas-delivery"} + request := surfaces.Request{ + SchemaVersion: surfaces.SchemaVersion, Operation: surfaces.OperationApply, Repository: repository, Host: "cli", CorrelationID: "cas-concurrent", + FlowID: "flow-cas", Goal: goal, TransitionID: "installation.initialize", Authority: human, + Parameters: protocol.Parameters{ + {Name: "source_revision", Value: "cas-fixture"}, {Name: "runtime_version", Value: runtimeVersion}, {Name: "runtime_sha256", Value: digestBytes(runtimeRaw)}, + {Name: "config_path", Value: configPath}, {Name: "config_sha256", Value: configFingerprint(t, configRaw)}, + }, + } + request = prescribeSurface(t, ctx, kernelA, request) + if request.Prescription.ExpectedStateRevision != 1 || request.Prescription.ExpectedProgramFingerprint != program.Fingerprint() { + t.Fatalf("initial prescription = %#v", request.Prescription) + } + resolver, err := plant.NewResolver(externalRoot) + if err != nil { + t.Fatal(err) + } + invocation, err := resolver.ResolveInvocation(ctx, repository, "cli", request.CorrelationID) + if err != nil { + t.Fatal(err) + } + layout, _, err := resolver.ResolveLayout(ctx, invocation) + if err != nil { + t.Fatal(err) + } + if _, err := os.Stat(layout.StatePath); !os.IsNotExist(err) { + t.Fatalf("read-only resolution changed durable state: %v", err) + } + + start := make(chan struct{}) + results := make(chan concurrentApplyResult, 3) + for _, kernel := range []boatstack.Kernel{kernelA, kernelA, kernelB} { + kernel := kernel + go func() { + <-start + response, applyErr := kernel.Handle(ctx, request) + results <- concurrentApplyResult{response: response, err: applyErr} + }() + } + close(start) + one, two, three := <-results, <-results, <-results + + successes, stale := 0, 0 + var committed surfaces.Response + for _, result := range []concurrentApplyResult{one, two, three} { + if result.err == nil { + successes++ + committed = result.response + continue + } + var staleErr engine.StalePrescriptionError + if errors.As(result.err, &staleErr) { + stale++ + if result.response.Receipt != nil { + t.Fatal("stale contender received a transition receipt") + } + } + } + if successes != 1 || stale != 2 { + t.Fatalf("concurrent results: success=%d stale=%d one=%v two=%v three=%v", successes, stale, one.err, two.err, three.err) + } + if committed.Receipt == nil || committed.Receipt.PriorStateRevision != 1 || committed.Receipt.ResultingStateRevision != 2 || + committed.Receipt.ProgramFingerprint != program.Fingerprint() || committed.Receipt.PrescriptionID != request.Prescription.ID { + t.Fatalf("commit receipt does not prove the consumed revision/program pair: %#v", committed.Receipt) + } + + stateRaw, err := os.ReadFile(layout.StatePath) + if err != nil { + t.Fatal(err) + } + state, err := durable.DecodeState(stateRaw) + if err != nil || state.Revision != 2 { + t.Fatalf("durable state = %#v, %v", state, err) + } + receiptRaw, err := os.ReadFile(layout.ReceiptPath) + if err != nil || bytes.Count(receiptRaw, []byte("\n")) != 1 { + t.Fatalf("receipt stream contains more than one commit: %v %q", err, receiptRaw) + } + + replayRequest := request + replayRequest.IdempotencyKey = committed.Receipt.IdempotencyKey + replayed, err := kernelA.Handle(ctx, replayRequest) + if err != nil || !replayed.Replayed || replayed.Receipt == nil || replayed.Receipt.ID != committed.Receipt.ID { + t.Fatalf("explicit idempotent replay = %#v, %v", replayed, err) + } + afterReplay, _ := os.ReadFile(layout.StatePath) + if !bytes.Equal(stateRaw, afterReplay) { + t.Fatal("idempotent replay created a second durable commit") + } +} + +func TestProgramChangeInvalidatesPriorPrescriptionBeforeEffects(t *testing.T) { + // control-law: apply executes only the canonical program observed by resolution + ctx := context.Background() + repository := testRepository(t) + externalRoot := t.TempDir() + programP := testProgram() + programQ, err := control.Compile(ctx, control.CompileRequest{ + KernelVersion: boatstack.Version, Core: core.System(), Runtime: standard.Definition(), Extensions: []control.Extension{releasenote.Definition()}, + }) + if err != nil { + t.Fatal(err) + } + kernelP, _ := boatstack.NewKernel(externalRoot, programP) + kernelQ, _ := boatstack.NewKernel(externalRoot, programQ) + executable, _ := os.Executable() + executable, _ = filepath.Abs(executable) + executable, _ = filepath.EvalSymlinks(executable) + runtimeRaw, _ := os.ReadFile(executable) + runtimeVersion := installTestRuntime(t, executable, runtimeRaw) + configPath := filepath.Join(t.TempDir(), "project.json") + configRaw := []byte("{\"schema_version\":2,\"project\":{\"name\":\"program-cas\",\"default_branch\":\"main\",\"commands\":{}},\"policy\":{\"plan_approval\":\"human\",\"visual_evidence\":\"optional\"},\"hosts\":[\"cli\"]}\n") + if err := os.WriteFile(configPath, configRaw, 0o600); err != nil { + t.Fatal(err) + } + now := time.Now().UTC() + request := surfaces.Request{ + SchemaVersion: surfaces.SchemaVersion, Operation: surfaces.OperationApply, Repository: repository, Host: "cli", CorrelationID: "program-cas", + FlowID: "flow-program-cas", Goal: model.Goal{ID: "program-cas", Kind: model.GoalApprovedPlan, DeliveryID: "program-cas"}, TransitionID: "installation.initialize", + Authority: protocol.AuthorityBundle{Receipts: []protocol.AuthorityReceipt{{ID: "program-cas-human", Class: catalog.AuthorityHuman, Subject: "operator", Fingerprint: "human", IssuedAt: now.Add(-time.Minute), ExpiresAt: now.Add(time.Hour)}}}, + Parameters: protocol.Parameters{ + {Name: "source_revision", Value: "program-cas"}, {Name: "runtime_version", Value: runtimeVersion}, {Name: "runtime_sha256", Value: digestBytes(runtimeRaw)}, + {Name: "config_path", Value: configPath}, {Name: "config_sha256", Value: configFingerprint(t, configRaw)}, + }, + } + request = prescribeSurface(t, ctx, kernelP, request) + response, err := kernelQ.Handle(ctx, request) + var stale engine.StalePrescriptionError + if !errors.As(err, &stale) || stale.ExpectedProgramFingerprint != programP.Fingerprint() || stale.ObservedProgramFingerprint != programQ.Fingerprint() { + t.Fatalf("program-change result = %#v, %v", response, err) + } + resolver, _ := plant.NewResolver(externalRoot) + invocation, _ := resolver.ResolveInvocation(ctx, repository, "cli", request.CorrelationID) + layout, _, _ := resolver.ResolveLayout(ctx, invocation) + if _, statErr := os.Stat(layout.StatePath); !os.IsNotExist(statErr) { + t.Fatalf("stale-program apply produced durable effects: %v", statErr) + } + if _, statErr := os.Stat(layout.ReceiptPath); !os.IsNotExist(statErr) { + t.Fatalf("stale-program apply produced a receipt: %v", statErr) + } + if !strings.Contains(err.Error(), "control program changed") { + t.Fatalf("stale-program diagnostic omitted the differing facet: %v", err) + } +} diff --git a/boatstack/internal/effects/driver.go b/boatstack/internal/effects/driver.go index 620decf..3c6ee76 100644 --- a/boatstack/internal/effects/driver.go +++ b/boatstack/internal/effects/driver.go @@ -76,7 +76,14 @@ func (d Driver) Prepare(ctx context.Context, admission protocol.Admission, trans if state.RepositoryID != admission.Invocation.RepositoryID || state.GitCommonID != admission.Invocation.GitCommonID || state.WorktreeID != admission.Invocation.WorktreeID { return nil, fmt.Errorf("durable state belongs to a different invocation") } - if state.ProgramFingerprint != "" && state.ProgramFingerprint != admission.ProgramFingerprint && !transition.Policy.ReconcilesProgram { + if state.Revision != admission.ExpectedStateRevision { + return nil, fmt.Errorf("durable state revision changed after admission") + } + resultingRevision, err := durable.NextRevision(state.Revision) + if err != nil { + return nil, err + } + if state.ProgramFingerprint != "" && state.ProgramFingerprint != admission.ExpectedProgramFingerprint && !transition.Policy.ReconcilesProgram { return nil, fmt.Errorf("compiled control program drifted; explicit program reconciliation is required") } if transition.ID == "catalog.reconcile" && (state.RuntimeVersion != admission.Invocation.RuntimeVersion || state.RuntimeFingerprint != admission.Invocation.RuntimeFingerprint) { @@ -90,7 +97,7 @@ func (d Driver) Prepare(ctx context.Context, admission protocol.Admission, trans } next := state if next.ProgramFingerprint == "" { - next.ProgramFingerprint = admission.ProgramFingerprint + next.ProgramFingerprint = admission.ExpectedProgramFingerprint } if err := d.boundary.PrepareObservation(ctx, admission, transition, layout, &next); err != nil { return nil, err @@ -98,7 +105,7 @@ func (d Driver) Prepare(ctx context.Context, admission protocol.Admission, trans if err := applyStateTransition(&next, admission, transition); err != nil { return nil, err } - next.Revision++ + next.Revision = resultingRevision next.UpdatedAt = d.clock.Now().UTC() var verificationInvocation *model.InvocationContext if transition.ID == "workspace.cut" { @@ -188,7 +195,7 @@ func (d Driver) Prepare(ctx context.Context, admission protocol.Admission, trans mutations = append(mutations, bindingMutation) } if transition.ID == "workspace.cut" { - parked := parkedSourceState(state, transition.ID, d.clock.Now()) + parked := parkedSourceState(state, next.Revision, transition.ID, d.clock.Now()) parkedRaw, encodeErr := durable.EncodeState(parked) if encodeErr != nil { return nil, encodeErr @@ -324,8 +331,8 @@ func canonicalWorkspaceDestination(admission protocol.Admission) (string, error) return canonical, nil } -func parkedSourceState(state durable.State, transition catalog.TransitionID, now time.Time) durable.State { - state.Revision++ +func parkedSourceState(state durable.State, resultingRevision uint64, transition catalog.TransitionID, now time.Time) durable.State { + state.Revision = resultingRevision state.Phase = model.PhaseDormant state.Engagement = model.EngagementDormant state.Delivery = model.DeliveryUninitialized diff --git a/boatstack/internal/effects/filelock_unix.go b/boatstack/internal/effects/filelock_unix.go index 56fc732..45b6cd3 100644 --- a/boatstack/internal/effects/filelock_unix.go +++ b/boatstack/internal/effects/filelock_unix.go @@ -10,7 +10,10 @@ import ( func lockFile(file *os.File) error { if err := syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { - return fmt.Errorf("kernel lock is held: %w", err) + if err == syscall.EWOULDBLOCK || err == syscall.EAGAIN { + return fmt.Errorf("%w: %v", errLockHeld, err) + } + return err } return nil } diff --git a/boatstack/internal/effects/filelock_windows.go b/boatstack/internal/effects/filelock_windows.go index 7e1a986..03079bc 100644 --- a/boatstack/internal/effects/filelock_windows.go +++ b/boatstack/internal/effects/filelock_windows.go @@ -26,7 +26,10 @@ func lockFile(file *os.File) error { 1, 0, uintptr(unsafe.Pointer(&overlapped)), ) if result == 0 { - return fmt.Errorf("kernel lock is held: %w", callErr) + if callErr == syscall.Errno(33) { + return fmt.Errorf("%w: %v", errLockHeld, callErr) + } + return callErr } return nil } diff --git a/boatstack/internal/effects/host_skills.go b/boatstack/internal/effects/host_skills.go index 9f9a4f5..eb592b4 100644 --- a/boatstack/internal/effects/host_skills.go +++ b/boatstack/internal/effects/host_skills.go @@ -107,7 +107,12 @@ Begin each cycle with an untargeted authority-bearing `+"`next`"+`. A `+"`CANDID identifies the next transition but is not permission to apply it: bind only its declared parameters and re-resolve that exact transition. Apply only the stable transition ID from the immediately preceding `+"`PRESCRIBED`"+` result and only its -declared parameters. Preserve the complete apply response and stderr, including +declared parameters. Carry that result's prescription ID, expected state revision, +expected program fingerprint, expected snapshot fingerprint, and correlation +unchanged into `+"`apply`"+` or `+"`recover`"+`. Never construct, reuse, or omit those +bindings. If the Kernel returns `+"`STALE_PRESCRIPTION`"+`, preserve its complete +diagnostic, perform no effect, discard the prescription, and re-resolve once from +the same command context. Preserve the complete apply response and stderr, including admission, receipt, postcondition, error, recovery, and transaction fields. Re-resolve with the same context after every complete receipt. diff --git a/boatstack/internal/effects/host_skills_test.go b/boatstack/internal/effects/host_skills_test.go index d2525dc..5890a46 100644 --- a/boatstack/internal/effects/host_skills_test.go +++ b/boatstack/internal/effects/host_skills_test.go @@ -52,6 +52,8 @@ func TestHostSkillProjectionPreservesAuthorityBoundaries(t *testing.T) { "requested authority sources separately from currently\nmaterialized authority receipts", "complete apply response and stderr", "authority-bearing `FRONTIER`", "Never synthesize missing\nauthority", "`CANDIDATE`", "immediately preceding `PRESCRIBED`", "every requested authority source is materialized\nor conclusively rejected against the post-receipt state", + "prescription ID, expected state revision", "expected program fingerprint, expected snapshot fingerprint", "correlation\nunchanged", + "`STALE_PRESCRIPTION`", "discard the prescription, and re-resolve once", } { if !strings.Contains(value, contract) { t.Fatalf("%s is missing authority contract %q", path, contract) diff --git a/boatstack/internal/effects/integration_test.go b/boatstack/internal/effects/integration_test.go index ff079fe..92d56f6 100644 --- a/boatstack/internal/effects/integration_test.go +++ b/boatstack/internal/effects/integration_test.go @@ -57,6 +57,38 @@ func testProgram() control.ControlProgram { return program } +func prescribeEngine(t *testing.T, ctx context.Context, kernel engine.Engine, request engine.ApplyRequest) engine.ApplyRequest { + t.Helper() + resolve := request.ResolveRequest + resolve.Parameters = request.Parameters + resolution, err := kernel.Resolve(ctx, resolve) + if err != nil { + t.Fatal(err) + } + if resolution.Decision.Kind != supervisor.DecisionPrescribed || resolution.Prescription.ID == "" { + t.Fatalf("resolution did not produce an exact prescription: %#v", resolution.Decision) + } + request.Prescription = resolution.Prescription + return request +} + +func prescribeSurface(t *testing.T, ctx context.Context, kernel boatstack.Kernel, request surfaces.Request) surfaces.Request { + t.Helper() + resolve := request + resolve.Operation = surfaces.OperationResolve + resolve.FlowID = "" + resolve.Prescription = protocol.Prescription{} + response, err := kernel.Handle(ctx, resolve) + if err != nil { + t.Fatal(err) + } + if response.Decision == nil || response.Decision.Kind != supervisor.DecisionPrescribed || response.Prescription == nil { + t.Fatalf("resolution did not produce an exact prescription: %#v", response.Decision) + } + request.Prescription = *response.Prescription + return request +} + func (c fixedClock) Now() time.Time { return c.value } func installTestRuntime(t *testing.T, executable string, raw []byte) string { @@ -135,10 +167,11 @@ func TestConcreteBoundaryAppliesAndReceiptsOneTransition(t *testing.T) { ID: "authority-1", Class: catalog.AuthorityHuman, Subject: invocation.RepositoryID, Fingerprint: "human-fingerprint", IssuedAt: clock.Now().Add(-time.Minute), ExpiresAt: clock.Now().Add(time.Hour), }}} - result, err := kernel.Apply(ctx, engine.ApplyRequest{ + request := engine.ApplyRequest{ ResolveRequest: engine.ResolveRequest{Invocation: invocation, Goal: goal, Authority: authority, Requested: "repository.attach"}, FlowID: "flow-1", Parameters: protocol.Parameters{{Name: "topology", Value: string(model.TopologyDetached)}, {Name: "config_authority", Value: "repository"}}, AdmissionLifetime: time.Minute, - }) + } + result, err := kernel.Apply(ctx, prescribeEngine(t, ctx, kernel, request)) if err != nil { t.Fatal(err) } @@ -173,11 +206,12 @@ func TestExternalConfigurationAuthorityTransfersAcrossAttachAndDetach(t *testing }}} apply := func(id catalog.TransitionID, authority protocol.AuthorityBundle, repositoryAuthority bool, parameters protocol.Parameters) surfaces.Response { t.Helper() - response, handleErr := kernel.Handle(ctx, surfaces.Request{ + request := surfaces.Request{ SchemaVersion: surfaces.SchemaVersion, Operation: surfaces.OperationApply, Repository: repository, Host: "cli", CorrelationID: "external-config-" + string(id), FlowID: "flow-external-config", Goal: goal, TransitionID: id, Authority: authority, RepositoryAuthority: repositoryAuthority, Parameters: parameters, - }) + } + response, handleErr := kernel.Handle(ctx, prescribeSurface(t, ctx, kernel, request)) if handleErr != nil { t.Fatalf("apply %s: %v", id, handleErr) } @@ -274,14 +308,15 @@ func TestProgramDriftRequiresAtomicInstallationReconciliation(t *testing.T) { if err := os.WriteFile(configPath, configRaw, 0o600); err != nil { t.Fatal(err) } - initialized, err := oldKernel.Handle(ctx, surfaces.Request{ + initializeRequest := surfaces.Request{ SchemaVersion: surfaces.SchemaVersion, Operation: surfaces.OperationApply, Repository: repository, Host: "cli", CorrelationID: "program-old", FlowID: "flow-program-drift", Goal: goal, TransitionID: "installation.initialize", Authority: human, Parameters: protocol.Parameters{ {Name: "source_revision", Value: "program-old"}, {Name: "runtime_version", Value: runtimeVersion}, {Name: "runtime_sha256", Value: digestBytes(runtimeRaw)}, {Name: "config_path", Value: configPath}, {Name: "config_sha256", Value: configFingerprint(t, configRaw)}, }, - }) + } + initialized, err := oldKernel.Handle(ctx, prescribeSurface(t, ctx, oldKernel, initializeRequest)) if err != nil { t.Fatal(err) } @@ -331,8 +366,10 @@ func TestProgramDriftRequiresAtomicInstallationReconciliation(t *testing.T) { {Name: "accept_obligation_change", Value: "true"}, }, } - frontier, err := newKernel.Handle(ctx, request) - if err == nil || frontier.Decision == nil || frontier.Decision.Kind != supervisor.DecisionFrontier { + frontierRequest := request + frontierRequest.Operation = surfaces.OperationResolve + frontier, err := newKernel.Handle(ctx, frontierRequest) + if err != nil || frontier.Decision == nil || frontier.Decision.Kind != supervisor.DecisionFrontier { t.Fatalf("authority-free reconciliation = response %#v error %v", frontier, err) } afterRejected, _ := os.ReadFile(layout.StatePath) @@ -349,7 +386,7 @@ func TestProgramDriftRequiresAtomicInstallationReconciliation(t *testing.T) { t.Fatal("invalid reconciliation mutated durable state") } request.Parameters[3].Value = "true" - reconciled, err := newKernel.Handle(ctx, request) + reconciled, err := newKernel.Handle(ctx, prescribeSurface(t, ctx, newKernel, request)) if err != nil { t.Fatal(err) } @@ -388,14 +425,15 @@ func TestProgramDriftRequiresAtomicInstallationReconciliation(t *testing.T) { if !bytes.Equal(afterSuccess, afterReplay) { t.Fatal("rejected repeated reconciliation mutated durable state") } - updated, err := newKernel.Handle(ctx, surfaces.Request{ + updateRequest := surfaces.Request{ SchemaVersion: surfaces.SchemaVersion, Operation: surfaces.OperationApply, Repository: repository, Host: "cli", CorrelationID: "program-current-update", FlowID: "flow-program-drift", Goal: model.Goal{ID: "ignored-command-goal", Kind: model.GoalOpenPR, DeliveryID: "ignored"}, TransitionID: "installation.update", Authority: human, Parameters: protocol.Parameters{ {Name: "source_revision", Value: "program-current"}, {Name: "runtime_version", Value: runtimeVersion}, {Name: "runtime_sha256", Value: digestBytes(runtimeRaw)}, }, - }) + } + updated, err := newKernel.Handle(ctx, prescribeSurface(t, ctx, newKernel, updateRequest)) if err != nil { t.Fatalf("current-program update after reconciliation: %v", err) } @@ -443,11 +481,12 @@ func TestReferenceExtensionUsesKernelAdmissionVerificationAndReceiptPath(t *test } apply := func(id catalog.TransitionID, authorization protocol.AuthorityBundle, parameters protocol.Parameters) surfaces.Response { t.Helper() - response, applyErr := kernel.Handle(ctx, surfaces.Request{ + request := surfaces.Request{ SchemaVersion: surfaces.SchemaVersion, Operation: surfaces.OperationApply, Repository: repository, Host: "cli", CorrelationID: "extension-" + string(id), FlowID: "flow-extension-receipt", Goal: goal, TransitionID: id, Authority: authorization, Parameters: parameters, - }) + } + response, applyErr := kernel.Handle(ctx, prescribeSurface(t, ctx, kernel, request)) if applyErr != nil { t.Fatalf("apply %s: %v", id, applyErr) } @@ -568,10 +607,11 @@ func TestConcreteWorkflowPreservesConfigurationProofAndGoalTerminals(t *testing. } apply := func(goal model.Goal, id catalog.TransitionID, auth protocol.AuthorityBundle, parameters protocol.Parameters) engine.ApplyResult { t.Helper() - result, applyErr := kernel.Apply(ctx, engine.ApplyRequest{ + request := engine.ApplyRequest{ ResolveRequest: engine.ResolveRequest{Invocation: invocation, Goal: goal, Authority: auth, Requested: id}, FlowID: "flow-workflow", Parameters: parameters, AdmissionLifetime: time.Minute, - }) + } + result, applyErr := kernel.Apply(ctx, prescribeEngine(t, ctx, kernel, request)) if applyErr != nil { t.Fatalf("apply %s: %v", id, applyErr) } @@ -702,10 +742,11 @@ func TestWorkspaceCutTransfersAuthorityToExactDestinationWorktree(t *testing.T) }}} apply := func(invocation model.InvocationContext, id catalog.TransitionID, authority protocol.AuthorityBundle, parameters protocol.Parameters) engine.ApplyResult { t.Helper() - result, applyErr := kernel.Apply(ctx, engine.ApplyRequest{ + request := engine.ApplyRequest{ ResolveRequest: engine.ResolveRequest{Invocation: invocation, Goal: goal, Authority: authority, Requested: id}, FlowID: "flow-workspace", Parameters: parameters, AdmissionLifetime: time.Minute, - }) + } + result, applyErr := kernel.Apply(ctx, prescribeEngine(t, ctx, kernel, request)) if applyErr != nil { t.Fatalf("apply %s: %v", id, applyErr) } diff --git a/boatstack/internal/effects/journal.go b/boatstack/internal/effects/journal.go index 0e329b8..6337317 100644 --- a/boatstack/internal/effects/journal.go +++ b/boatstack/internal/effects/journal.go @@ -73,7 +73,7 @@ func (j *Journal) Begin(ctx context.Context, admission protocol.Admission, trans return statErr } now := j.clock.Now().UTC() - record := journalRecord{SchemaVersion: 2, Admission: admission, TransitionID: transition.ID, TransitionClass: transition.Class, ReconcilesProgram: transition.Policy.ReconcilesProgram, Status: "begun", CreatedAt: now, UpdatedAt: now} + record := journalRecord{SchemaVersion: protocol.JournalSchemaVersion, Admission: admission, TransitionID: transition.ID, TransitionClass: transition.Class, ReconcilesProgram: transition.Policy.ReconcilesProgram, Status: "begun", CreatedAt: now, UpdatedAt: now} raw, err := encodeJSON(record) if err != nil { return err @@ -104,7 +104,7 @@ func readJournal(path string) (journalRecord, error) { if err := decoder.Decode(&trailing); err != io.EOF { return journalRecord{}, fmt.Errorf("transaction journal %s contains trailing JSON", path) } - if record.SchemaVersion != 2 || record.Admission.ID == "" || record.TransitionID == "" || !record.TransitionClass.Valid() || !record.TransitionClass.Controllable() || record.Status == "" { + if record.SchemaVersion != protocol.JournalSchemaVersion || record.Admission.ID == "" || record.TransitionID == "" || !record.TransitionClass.Valid() || !record.TransitionClass.Controllable() || record.Status == "" { return journalRecord{}, fmt.Errorf("invalid transaction journal %s", path) } if err := record.Admission.ValidateIdentity(); err != nil || record.Admission.TransitionID != record.TransitionID { diff --git a/boatstack/internal/effects/locker.go b/boatstack/internal/effects/locker.go index 8158e27..f3275dc 100644 --- a/boatstack/internal/effects/locker.go +++ b/boatstack/internal/effects/locker.go @@ -2,11 +2,13 @@ package effects import ( "context" + "errors" "fmt" "os" "path/filepath" "sort" "strings" + "time" "github.com/operatorstack/boatstack/boatstack/internal/kernel/model" "github.com/operatorstack/boatstack/boatstack/internal/kernel/ports" @@ -14,6 +16,8 @@ import ( type Locker struct{ resolver ports.InvocationResolver } +var errLockHeld = errors.New("kernel lock is held") + func NewLocker(resolver ports.InvocationResolver) (Locker, error) { if resolver == nil { return Locker{}, fmt.Errorf("effect locker requires an invocation resolver") @@ -75,10 +79,23 @@ func (l Locker) Acquire(ctx context.Context, invocation model.InvocationContext, _ = held.Release() return nil, fmt.Errorf("open lock %s: %w", path, openErr) } - if lockErr := lockFile(file); lockErr != nil { - _ = file.Close() - _ = held.Release() - return nil, fmt.Errorf("acquire lock %s: %w", path, lockErr) + for { + lockErr := lockFile(file) + if lockErr == nil { + break + } + if !errors.Is(lockErr, errLockHeld) { + _ = file.Close() + _ = held.Release() + return nil, fmt.Errorf("acquire lock %s: %w", path, lockErr) + } + select { + case <-ctx.Done(): + _ = file.Close() + _ = held.Release() + return nil, fmt.Errorf("acquire lock %s: %w", path, ctx.Err()) + case <-time.After(10 * time.Millisecond): + } } if truncateErr := file.Truncate(0); truncateErr != nil { _ = unlockFile(file) diff --git a/boatstack/internal/effects/locker_test.go b/boatstack/internal/effects/locker_test.go index 1e6f580..43e50b2 100644 --- a/boatstack/internal/effects/locker_test.go +++ b/boatstack/internal/effects/locker_test.go @@ -5,6 +5,7 @@ import ( "os" "path/filepath" "testing" + "time" "github.com/operatorstack/boatstack/boatstack/internal/plant" ) @@ -27,7 +28,9 @@ func TestKernelLockUsesProcessScopedHandleNotFilePresence(t *testing.T) { if err != nil { t.Fatal(err) } - if _, err := locker.Acquire(context.Background(), invocation, []string{"state"}); err == nil { + blocked, cancel := context.WithTimeout(context.Background(), 25*time.Millisecond) + defer cancel() + if _, err := locker.Acquire(blocked, invocation, []string{"state"}); err == nil { t.Fatal("concurrent acquisition unexpectedly succeeded") } if err := first.Release(); err != nil { diff --git a/boatstack/internal/effects/receipts.go b/boatstack/internal/effects/receipts.go index a4c14e3..a9dbdd9 100644 --- a/boatstack/internal/effects/receipts.go +++ b/boatstack/internal/effects/receipts.go @@ -131,23 +131,26 @@ func (s *ReceiptStore) FindByIdempotency(ctx context.Context, invocation model.I } type processEvent struct { - SchemaVersion int `json:"schema_version"` - FlowID string `json:"flow_id"` - Sequence uint64 `json:"sequence"` - Timestamp time.Time `json:"timestamp"` - GoalID string `json:"goal_id"` - GoalScope string `json:"goal_scope,omitempty"` - GoalStatus string `json:"goal_status,omitempty"` - TransitionID string `json:"transition_id"` - ProgramFingerprint string `json:"program_fingerprint"` - SourceFingerprint string `json:"source_fingerprint"` - TargetFingerprint string `json:"target_fingerprint"` - Outcome string `json:"outcome"` - DurationNanoseconds int64 `json:"duration_nanoseconds"` - AuthorityClasses []string `json:"authority_classes,omitempty"` - Recovery string `json:"recovery,omitempty"` - Terminal string `json:"terminal"` - FailureClass string `json:"failure_class,omitempty"` + SchemaVersion int `json:"schema_version"` + FlowID string `json:"flow_id"` + Sequence uint64 `json:"sequence"` + Timestamp time.Time `json:"timestamp"` + GoalID string `json:"goal_id"` + GoalScope string `json:"goal_scope,omitempty"` + GoalStatus string `json:"goal_status,omitempty"` + TransitionID string `json:"transition_id"` + ProgramFingerprint string `json:"program_fingerprint"` + PrescriptionID string `json:"prescription_id"` + PriorStateRevision uint64 `json:"prior_state_revision"` + ResultingStateRevision uint64 `json:"resulting_state_revision"` + SourceFingerprint string `json:"source_fingerprint"` + TargetFingerprint string `json:"target_fingerprint"` + Outcome string `json:"outcome"` + DurationNanoseconds int64 `json:"duration_nanoseconds"` + AuthorityClasses []string `json:"authority_classes,omitempty"` + Recovery string `json:"recovery,omitempty"` + Terminal string `json:"terminal"` + FailureClass string `json:"failure_class,omitempty"` } func appendLine(path string, value any, mode os.FileMode) error { @@ -182,9 +185,11 @@ func (s *ReceiptStore) Append(ctx context.Context, receipt protocol.TransitionRe return err } event := processEvent{ - SchemaVersion: 1, FlowID: receipt.FlowID, Sequence: receipt.Sequence, Timestamp: s.clock.Now().UTC(), GoalID: receipt.GoalID, + SchemaVersion: 2, FlowID: receipt.FlowID, Sequence: receipt.Sequence, Timestamp: s.clock.Now().UTC(), GoalID: receipt.GoalID, GoalScope: string(receipt.GoalScope), GoalStatus: string(receipt.GoalStatus), - TransitionID: string(receipt.TransitionID), ProgramFingerprint: receipt.ProgramFingerprint, SourceFingerprint: receipt.SourceFingerprint, TargetFingerprint: receipt.TargetFingerprint, + TransitionID: string(receipt.TransitionID), ProgramFingerprint: receipt.ProgramFingerprint, PrescriptionID: receipt.PrescriptionID, + PriorStateRevision: receipt.PriorStateRevision, ResultingStateRevision: receipt.ResultingStateRevision, + SourceFingerprint: receipt.SourceFingerprint, TargetFingerprint: receipt.TargetFingerprint, Outcome: string(receipt.Outcome), DurationNanoseconds: receipt.DurationNanoseconds, AuthorityClasses: append([]string(nil), receipt.AuthorityClasses...), Recovery: string(receipt.Recovery), Terminal: string(receipt.Terminal), FailureClass: receipt.FailureClass, } diff --git a/boatstack/internal/effects/recovery.go b/boatstack/internal/effects/recovery.go index f1aa9d3..7352dce 100644 --- a/boatstack/internal/effects/recovery.go +++ b/boatstack/internal/effects/recovery.go @@ -10,6 +10,7 @@ import ( "time" "github.com/operatorstack/boatstack/boatstack/internal/kernel/catalog" + "github.com/operatorstack/boatstack/boatstack/internal/kernel/durable" "github.com/operatorstack/boatstack/boatstack/internal/kernel/model" "github.com/operatorstack/boatstack/boatstack/internal/kernel/ports" "github.com/operatorstack/boatstack/boatstack/internal/kernel/protocol" @@ -56,6 +57,10 @@ func (d Driver) prepareRecoveryReplay(ctx context.Context, layout ports.Controll } mutations = append(mutations, mutation) } + mutations, err = d.advanceRecoveredState(layout, admission, transition.ID, mutations) + if err != nil { + return nil, err + } closure, err := prepareJournalClosureFromRecord(pendingPath, record, string(transition.ID), d.clock.Now()) if err != nil { return nil, err @@ -115,6 +120,10 @@ func (d Driver) prepareWorkspaceCutReconciliation(ctx context.Context, layout po } mutations = append(mutations, mutation) } + mutations, err = d.advanceRecoveredState(layout, admission, "workspace.reconcile", mutations) + if err != nil { + return nil, err + } closure, err := prepareJournalClosureFromRecord(pendingPath, record, string("workspace.reconcile"), d.clock.Now()) if err != nil { return nil, err @@ -123,6 +132,69 @@ func (d Driver) prepareWorkspaceCutReconciliation(ctx context.Context, layout po return &preparedEffect{mutations: mutations, verifyInvocation: verificationInvocation}, nil } +func (d Driver) advanceRecoveredState(layout ports.ControllerLayout, admission protocol.Admission, transition catalog.TransitionID, mutations []ports.ResourceMutation) ([]ports.ResourceMutation, error) { + resultingRevision, err := durable.NextRevision(admission.ExpectedStateRevision) + if err != nil { + return nil, err + } + advanced := false + for index := range mutations { + mutation := &mutations[index] + if mutation.Delete && mutation.TargetLink == "" && filepath.Clean(mutation.Path) == filepath.Clean(layout.StatePath) { + state := durable.Default(admission.Invocation, d.clock.Now()) + state.ProgramFingerprint = admission.ExpectedProgramFingerprint + state.Revision = resultingRevision + state.LastTransition = transition + state.UpdatedAt = d.clock.Now().UTC() + mutation.Target, err = durable.EncodeState(state) + if err != nil { + return nil, err + } + mutation.Delete = false + mutation.InstallLast = true + advanced = true + continue + } + if mutation.Delete || mutation.TargetLink != "" || filepath.Base(mutation.Path) != "state.json" || len(mutation.Target) == 0 { + continue + } + state, err := durable.DecodeState(mutation.Target) + if err != nil { + continue + } + state.Revision = resultingRevision + state.LastTransition = transition + state.UpdatedAt = d.clock.Now().UTC() + mutation.Target, err = durable.EncodeState(state) + if err != nil { + return nil, err + } + advanced = true + } + if advanced { + return mutations, nil + } + state, err := loadDurableState(layout.StatePath, admission.Invocation, d.clock.Now()) + if err != nil { + return nil, err + } + if state.Revision != admission.ExpectedStateRevision { + return nil, fmt.Errorf("recovery state revision changed after admission") + } + state.Revision = resultingRevision + state.LastTransition = transition + state.UpdatedAt = d.clock.Now().UTC() + raw, err := durable.EncodeState(state) + if err != nil { + return nil, err + } + mutation, err := mutationFor(layout.StatePath, raw, 0o600, true, false) + if err != nil { + return nil, err + } + return append(mutations, mutation), nil +} + func loadInterruptedJournal(layout ports.ControllerLayout, transactionID string) (journalRecord, string, error) { name, err := journalName(transactionID, ".pending") if err != nil { diff --git a/boatstack/internal/effects/recovery_test.go b/boatstack/internal/effects/recovery_test.go index 07a343e..7ae92a2 100644 --- a/boatstack/internal/effects/recovery_test.go +++ b/boatstack/internal/effects/recovery_test.go @@ -11,6 +11,7 @@ import ( "github.com/operatorstack/boatstack/boatstack/flow/standard" "github.com/operatorstack/boatstack/boatstack/internal/buildinfo" "github.com/operatorstack/boatstack/boatstack/internal/kernel/catalog" + "github.com/operatorstack/boatstack/boatstack/internal/kernel/durable" "github.com/operatorstack/boatstack/boatstack/internal/kernel/engine" "github.com/operatorstack/boatstack/boatstack/internal/kernel/model" "github.com/operatorstack/boatstack/boatstack/internal/kernel/ports" @@ -65,7 +66,7 @@ func recoveryRepository(t *testing.T) string { return repository } -func TestRestartRecoveryRollsBackExactPriorBytesAndArchivesJournal(t *testing.T) { +func TestRestartRecoveryRestoresPriorStateAndCommitsRecoveryRevision(t *testing.T) { // control-law: interrupted-local-effect-has-restart-safe-exact-rollback ctx := context.Background() clock := recoveryClock{value: time.Unix(3000, 0).UTC()} @@ -116,7 +117,11 @@ func TestRestartRecoveryRollsBackExactPriorBytesAndArchivesJournal(t *testing.T) {Name: "source_revision", Value: "recovery-fixture"}, {Name: "runtime_version", Value: runtimeIdentity.Version}, {Name: "runtime_sha256", Value: sha256Bytes(runtimeRaw)}, {Name: "config_path", Value: configPath}, {Name: "config_sha256", Value: configFingerprint}, } - admission, err := protocol.NewAdmission(initial, goal, transition, authority, parameters, clock.Now(), time.Minute) + prescription, err := protocol.NewPrescription(initial, transition) + if err != nil { + t.Fatal(err) + } + admission, err := protocol.NewAdmission(initial, goal, transition, prescription, authority, parameters, clock.Now(), time.Minute) if err != nil { t.Fatal(err) } @@ -165,13 +170,25 @@ func TestRestartRecoveryRollsBackExactPriorBytesAndArchivesJournal(t *testing.T) if err != nil { t.Fatal(err) } - result, err := restartedEngine.Apply(ctx, engine.ApplyRequest{ + recoveryRequest := engine.ApplyRequest{ ResolveRequest: engine.ResolveRequest{Invocation: restartedInvocation, Goal: goal, Authority: authority, Requested: "recovery.rollback"}, FlowID: "flow-recovery", Parameters: protocol.Parameters{{Name: "transaction_id", Value: admission.ID}}, AdmissionLifetime: time.Minute, - }) + } + recoveryResolve := recoveryRequest.ResolveRequest + recoveryResolve.Parameters = recoveryRequest.Parameters + resolvedRecovery, err := restartedEngine.Resolve(ctx, recoveryResolve) if err != nil { t.Fatal(err) } + recoveryRequest.Prescription = resolvedRecovery.Prescription + result, err := restartedEngine.Apply(ctx, recoveryRequest) + if err != nil { + t.Fatal(err) + } + if result.Receipt.PriorStateRevision != resolvedRecovery.Prescription.ExpectedStateRevision || + result.Receipt.ResultingStateRevision != resolvedRecovery.Prescription.ExpectedStateRevision+1 { + t.Fatalf("recovery receipt did not advance exactly once: %#v", result.Receipt) + } if result.Target.Phase.Value != model.PhaseDormant || result.Target.Recovery.Value != model.RecoveryNone || result.Target.Goal.Status != model.FactAbsent || result.Receipt.ID == "" || result.Receipt.GoalStatus != model.FactAbsent || result.Receipt.GoalID != "" { t.Fatalf("rollback target=%#v receipt=%q", result.Target, result.Receipt.ID) @@ -185,8 +202,16 @@ func TestRestartRecoveryRollsBackExactPriorBytesAndArchivesJournal(t *testing.T) if _, err := os.Stat(recovered); err != nil { t.Fatalf("recovered journal missing: %v", err) } - if _, err := os.Stat(layout.StatePath); !os.IsNotExist(err) { - t.Fatalf("rollback did not restore absent state file: %v", err) + stateRaw, err := os.ReadFile(layout.StatePath) + if err != nil { + t.Fatalf("rollback did not commit its recovery revision: %v", err) + } + recoveredState, err := durable.DecodeState(stateRaw) + if err != nil || recoveredState.Revision != result.Receipt.ResultingStateRevision { + t.Fatalf("rollback state revision is not receipt-bound: state=%#v err=%v receipt=%#v", recoveredState, err, result.Receipt) + } + if recoveredState.Goal.Validate() == nil || recoveredState.Runtime != model.RuntimeAbsent || recoveredState.Configuration != model.ConfigurationUnsupported { + t.Fatalf("rollback created product intent or retained initialized state: %#v", recoveredState) } if _, err := os.Stat(boatstackruntime.PinPath(repository)); !os.IsNotExist(err) { t.Fatalf("rollback did not restore the absent repository runtime pin: %v", err) diff --git a/boatstack/internal/effects/revision.go b/boatstack/internal/effects/revision.go new file mode 100644 index 0000000..c802e56 --- /dev/null +++ b/boatstack/internal/effects/revision.go @@ -0,0 +1,63 @@ +package effects + +import ( + "context" + "fmt" + + "github.com/operatorstack/boatstack/boatstack/internal/kernel/catalog" + "github.com/operatorstack/boatstack/boatstack/internal/kernel/durable" + "github.com/operatorstack/boatstack/boatstack/internal/kernel/ports" + "github.com/operatorstack/boatstack/boatstack/internal/kernel/protocol" +) + +const ( + kernelStateResource = "boatstack.kernel.state" + kernelStateOwner = "boatstack.kernel" +) + +// BindStateRevision appends the kernel-owned logical state commit to a +// protocol-backed program or extension effect. Native effects already include +// this mutation in Driver.Prepare. +func BindStateRevision(ctx context.Context, prepared ports.PreparedEffect, resolver ports.InvocationResolver, clock ports.Clock, admission protocol.Admission, transition catalog.Transition) (ports.PreparedEffect, error) { + effect, ok := prepared.(*preparedEffect) + if !ok { + return nil, fmt.Errorf("state revision binding requires a Boatstack prepared effect") + } + layout, invocation, err := resolver.ResolveLayout(ctx, admission.Invocation) + if err != nil { + return nil, err + } + if invocation.RepositoryID != admission.Invocation.RepositoryID || invocation.GitCommonID != admission.Invocation.GitCommonID || invocation.WorktreeID != admission.Invocation.WorktreeID { + return nil, fmt.Errorf("effect invocation identity changed before revision binding") + } + state, err := loadDurableState(layout.StatePath, admission.Invocation, clock.Now()) + if err != nil { + return nil, err + } + if state.Revision != admission.ExpectedStateRevision { + return nil, fmt.Errorf("durable state revision changed before revision binding") + } + if state.ProgramFingerprint != "" && state.ProgramFingerprint != admission.ExpectedProgramFingerprint { + return nil, fmt.Errorf("compiled control program changed before revision binding") + } + state.Revision, err = durable.NextRevision(state.Revision) + if err != nil { + return nil, err + } + if state.ProgramFingerprint == "" { + state.ProgramFingerprint = admission.ExpectedProgramFingerprint + } + state.LastTransition = transition.ID + state.UpdatedAt = clock.Now().UTC() + raw, err := durable.EncodeState(state) + if err != nil { + return nil, err + } + mutation, err := mutationFor(layout.StatePath, raw, 0o600, true, false) + if err != nil { + return nil, err + } + mutation.Resource, mutation.Owner = kernelStateResource, kernelStateOwner + effect.mutations = append(effect.mutations, mutation) + return effect, nil +} diff --git a/boatstack/internal/effects/state_reducer.go b/boatstack/internal/effects/state_reducer.go index de7f295..66d122a 100644 --- a/boatstack/internal/effects/state_reducer.go +++ b/boatstack/internal/effects/state_reducer.go @@ -57,7 +57,7 @@ func applyStateTransition(state *durable.State, admission protocol.Admission, tr if accepted != "true" || admission.PriorProgramFingerprint == "" || admission.ProgramDeltaFingerprint == "" || state.ProgramFingerprint != admission.PriorProgramFingerprint { return fmt.Errorf("reconciled installation update must bind and explicitly accept the exact prior-to-candidate program delta") } - state.ProgramFingerprint = admission.ProgramFingerprint + state.ProgramFingerprint = admission.ExpectedProgramFingerprint state.Runtime = model.RuntimeVerified state.RuntimeVersion, _ = admission.Parameters.Get("runtime_version") state.RuntimeFingerprint, _ = admission.Parameters.Get("runtime_sha256") @@ -83,7 +83,7 @@ func applyStateTransition(state *durable.State, admission protocol.Admission, tr if prior == "" || prior != state.ProgramFingerprint || accepted != "true" { return fmt.Errorf("catalog reconciliation must bind the prior program and explicitly accept obligation changes") } - state.ProgramFingerprint = admission.ProgramFingerprint + state.ProgramFingerprint = admission.ExpectedProgramFingerprint case "installation.initialize": state.Runtime, state.Configuration = model.RuntimeVerified, model.ConfigurationVerified state.RuntimeVersion, _ = admission.Parameters.Get("runtime_version") diff --git a/boatstack/internal/kernel/durable/state.go b/boatstack/internal/kernel/durable/state.go index 410543e..0d80c73 100644 --- a/boatstack/internal/kernel/durable/state.go +++ b/boatstack/internal/kernel/durable/state.go @@ -82,6 +82,16 @@ func Default(invocation model.InvocationContext, now time.Time) State { } } +func NextRevision(current uint64) (uint64, error) { + if current == 0 { + return 0, fmt.Errorf("durable state revision is absent") + } + if current == ^uint64(0) { + return 0, fmt.Errorf("durable state revision overflow") + } + return current + 1, nil +} + func (s State) Validate() error { if s.SchemaVersion != StateSchemaVersion { return fmt.Errorf("durable state schema %d, want %d", s.SchemaVersion, StateSchemaVersion) diff --git a/boatstack/internal/kernel/durable/state_revision_test.go b/boatstack/internal/kernel/durable/state_revision_test.go new file mode 100644 index 0000000..c37f81f --- /dev/null +++ b/boatstack/internal/kernel/durable/state_revision_test.go @@ -0,0 +1,17 @@ +package durable + +import "testing" + +func TestNextRevisionAdvancesExactlyOnceAndRejectsOverflow(t *testing.T) { + // control-law: every successful logical transition advances one durable revision + next, err := NextRevision(41) + if err != nil || next != 42 { + t.Fatalf("next revision = %d, %v", next, err) + } + if _, err := NextRevision(0); err == nil { + t.Fatal("absent revision advanced") + } + if _, err := NextRevision(^uint64(0)); err == nil { + t.Fatal("uint64 revision overflow wrapped") + } +} diff --git a/boatstack/internal/kernel/engine/engine.go b/boatstack/internal/kernel/engine/engine.go index a88ee2b..5f5b5fb 100644 --- a/boatstack/internal/kernel/engine/engine.go +++ b/boatstack/internal/kernel/engine/engine.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "strings" "time" "github.com/operatorstack/boatstack/boatstack/internal/kernel/catalog" @@ -45,9 +46,11 @@ type ResolveRequest struct { } type Resolution struct { - Snapshot model.Snapshot - Goal model.Goal - Decision supervisor.Decision + Snapshot model.Snapshot + Goal model.Goal + Decision supervisor.Decision + Prescription protocol.Prescription + Admission protocol.Admission } func (e Engine) Resolve(ctx context.Context, request ResolveRequest) (Resolution, error) { @@ -105,7 +108,14 @@ func (e Engine) Resolve(ctx context.Context, request ResolveRequest) (Resolution decision.Transition = nil } } else { - admission, admissionErr := protocol.NewAdmission(snapshot, goal, *decision.Transition, request.Authority, request.Parameters, now, 2*time.Minute) + prescription, prescriptionErr := protocol.NewPrescription(snapshot, *decision.Transition) + if prescriptionErr != nil { + decision.Kind = supervisor.DecisionUnresolved + decision.Reason = prescriptionErr.Error() + decision.Transition = nil + return Resolution{Snapshot: snapshot, Goal: goal, Decision: decision}, nil + } + admission, admissionErr := protocol.NewAdmission(snapshot, goal, *decision.Transition, prescription, request.Authority, request.Parameters, now, 2*time.Minute) if admissionErr != nil { decision.Kind = supervisor.DecisionUnresolved decision.Reason = admissionErr.Error() @@ -114,6 +124,8 @@ func (e Engine) Resolve(ctx context.Context, request ResolveRequest) (Resolution decision.Kind = supervisor.DecisionUnresolved decision.Reason = fmt.Sprintf("transition %q failed deterministic effect preflight: %v", admission.TransitionID, preflightErr) decision.Transition = nil + } else { + return Resolution{Snapshot: snapshot, Goal: goal, Decision: decision, Prescription: prescription, Admission: admission}, nil } } } @@ -127,6 +139,7 @@ func unresolvedResolution(goal model.Goal, reason string) Resolution { type ApplyRequest struct { ResolveRequest FlowID string + Prescription protocol.Prescription Parameters protocol.Parameters IdempotencyKey string AdmissionLifetime time.Duration @@ -153,6 +166,29 @@ type StaleAdmissionError struct{ Err error } func (e StaleAdmissionError) Error() string { return "stale admission: " + e.Err.Error() } func (e StaleAdmissionError) Unwrap() error { return e.Err } +type StalePrescriptionError struct { + PrescriptionID string + ExpectedStateRevision uint64 + ObservedStateRevision uint64 + ExpectedProgramFingerprint string + ObservedProgramFingerprint string + SnapshotChanged bool +} + +func (e StalePrescriptionError) Error() string { + facets := make([]string, 0, 3) + if e.ExpectedStateRevision != e.ObservedStateRevision { + facets = append(facets, fmt.Sprintf("state revision %d != %d", e.ExpectedStateRevision, e.ObservedStateRevision)) + } + if e.ExpectedProgramFingerprint != e.ObservedProgramFingerprint { + facets = append(facets, "control program changed") + } + if e.SnapshotChanged { + facets = append(facets, "admission-relevant snapshot changed") + } + return fmt.Sprintf("STALE_PRESCRIPTION %q: %s; re-resolve before apply", e.PrescriptionID, strings.Join(facets, ", ")) +} + type PostconditionError struct { Transition catalog.TransitionID Recovery catalog.TransitionID @@ -181,6 +217,12 @@ func (e Engine) Apply(ctx context.Context, request ApplyRequest) (result ApplyRe if request.FlowID == "" { return result, fmt.Errorf("kernel apply requires flow identity") } + if err := request.Prescription.Validate(); err != nil { + return result, err + } + if request.Requested != request.Prescription.TransitionID { + return result, fmt.Errorf("apply transition does not match prescription") + } if request.IdempotencyKey != "" { prior, ok, err := e.receipts.FindByIdempotency(ctx, request.Invocation, request.IdempotencyKey) if err != nil { @@ -220,13 +262,16 @@ func (e Engine) Apply(ctx context.Context, request ApplyRequest) (result ApplyRe if err != nil { return result, err } + if err := validatePrescriptionCurrent(request.Prescription, resolution.Snapshot); err != nil { + return result, err + } request.Goal = resolution.Goal if resolution.Decision.Kind != supervisor.DecisionPrescribed || resolution.Decision.Transition == nil { return result, DecisionError{Decision: resolution.Decision} } transition := *resolution.Decision.Transition now := e.clock.Now() - admission, err := protocol.NewAdmission(resolution.Snapshot, request.Goal, transition, request.Authority, request.Parameters, now, request.AdmissionLifetime) + admission, err := protocol.NewAdmission(resolution.Snapshot, request.Goal, transition, request.Prescription, request.Authority, request.Parameters, now, request.AdmissionLifetime) if err != nil { return result, err } @@ -234,32 +279,32 @@ func (e Engine) Apply(ctx context.Context, request ApplyRequest) (result ApplyRe if request.IdempotencyKey != "" && request.IdempotencyKey != admission.IdempotencyKey { return result, fmt.Errorf("supplied idempotency key does not match the exact admitted request") } - if err := e.receipts.Bind(ctx, request.FlowID, admission); err != nil { - return result, err - } - defer e.receipts.Unbind(request.FlowID) - if prior, ok, err := e.receipts.FindByIdempotency(ctx, request.Invocation, admission.IdempotencyKey); err != nil { - return result, fmt.Errorf("check idempotency: %w", err) - } else if ok { - if err := validateReplayRequest(prior, request, e.programFingerprint); err != nil { - return result, err - } - observation, observeErr := e.observer.Observe(ctx, ports.ObservationRequest{Invocation: request.Invocation}) - if observeErr != nil { - return result, observeErr - } - current, canonicalErr := e.canonicalize(observation) - if canonicalErr != nil { - return result, canonicalErr - } - if err := validateReplayGoalState(prior, current); err != nil { - return result, err + if request.IdempotencyKey != "" { + prior, ok, err := e.receipts.FindByIdempotency(ctx, request.Invocation, admission.IdempotencyKey) + if err != nil { + return result, fmt.Errorf("check idempotency: %w", err) } - if !replayStateSettled(current) { - return result, ReplayRecoveryError{ReceiptID: prior.ID} + if ok { + if err := validateReplayRequest(prior, request, e.programFingerprint); err != nil { + return result, err + } + observation, observeErr := e.observer.Observe(ctx, ports.ObservationRequest{Invocation: request.Invocation}) + if observeErr != nil { + return result, observeErr + } + current, canonicalErr := e.canonicalize(observation) + if canonicalErr != nil { + return result, canonicalErr + } + if err := validateReplayGoalState(prior, current); err != nil { + return result, err + } + if !replayStateSettled(current) { + return result, ReplayRecoveryError{ReceiptID: prior.ID} + } + result.Target, result.Receipt, result.Replayed = current, prior, true + return result, nil } - result.Target, result.Receipt, result.Replayed = current, prior, true - return result, nil } lock, err := e.locker.Acquire(ctx, request.Invocation, transition.OwnedResources) @@ -280,24 +325,35 @@ func (e Engine) Apply(ctx context.Context, request ApplyRequest) (result ApplyRe if err != nil { return result, err } - if prior, ok, findErr := e.receipts.FindByIdempotency(ctx, request.Invocation, admission.IdempotencyKey); findErr != nil { - return result, fmt.Errorf("check locked idempotency: %w", findErr) - } else if ok { - if err := validateReplayRequest(prior, request, e.programFingerprint); err != nil { - return result, err - } - if err := validateReplayGoalState(prior, lockedSnapshot); err != nil { - return result, err + if err := validatePrescriptionCurrent(request.Prescription, lockedSnapshot); err != nil { + return result, err + } + if request.IdempotencyKey != "" { + prior, ok, findErr := e.receipts.FindByIdempotency(ctx, request.Invocation, admission.IdempotencyKey) + if findErr != nil { + return result, fmt.Errorf("check locked idempotency: %w", findErr) } - if !replayStateSettled(lockedSnapshot) { - return result, ReplayRecoveryError{ReceiptID: prior.ID} + if ok { + if err := validateReplayRequest(prior, request, e.programFingerprint); err != nil { + return result, err + } + if err := validateReplayGoalState(prior, lockedSnapshot); err != nil { + return result, err + } + if !replayStateSettled(lockedSnapshot) { + return result, ReplayRecoveryError{ReceiptID: prior.ID} + } + result.Target, result.Receipt, result.Replayed = lockedSnapshot, prior, true + return result, nil } - result.Target, result.Receipt, result.Replayed = lockedSnapshot, prior, true - return result, nil } if err := admission.ValidateCurrent(lockedSnapshot, request.Goal, transition, e.clock.Now()); err != nil { return result, StaleAdmissionError{Err: err} } + if err := e.receipts.Bind(ctx, request.FlowID, admission); err != nil { + return result, err + } + defer e.receipts.Unbind(request.FlowID) if err := e.journal.Begin(ctx, admission, transition); err != nil { return result, fmt.Errorf("begin transaction journal: %w", err) } @@ -388,6 +444,25 @@ func (e Engine) Apply(ctx context.Context, request ApplyRequest) (result ApplyRe return result, nil } +func validatePrescriptionCurrent(prescription protocol.Prescription, snapshot model.Snapshot) error { + if err := prescription.Validate(); err != nil { + return err + } + snapshotChanged := prescription.ExpectedSnapshotFingerprint != snapshot.Fingerprint + if prescription.ExpectedStateRevision == snapshot.StateRevision && + prescription.ExpectedProgramFingerprint == snapshot.ProgramFingerprint && !snapshotChanged { + return nil + } + return StalePrescriptionError{ + PrescriptionID: prescription.ID, + ExpectedStateRevision: prescription.ExpectedStateRevision, + ObservedStateRevision: snapshot.StateRevision, + ExpectedProgramFingerprint: prescription.ExpectedProgramFingerprint, + ObservedProgramFingerprint: snapshot.ProgramFingerprint, + SnapshotChanged: snapshotChanged, + } +} + func validateReplayRequest(prior protocol.TransitionReceipt, request ApplyRequest, programFingerprint string) error { if prior.ProgramFingerprint != programFingerprint { return fmt.Errorf("idempotency receipt belongs to a different control program") @@ -395,6 +470,9 @@ func validateReplayRequest(prior protocol.TransitionReceipt, request ApplyReques if prior.FlowID != request.FlowID { return fmt.Errorf("idempotency receipt belongs to flow %q, not %q", prior.FlowID, request.FlowID) } + if prior.PrescriptionID != request.Prescription.ID { + return fmt.Errorf("idempotency receipt belongs to a different prescription") + } if prior.GoalScope != catalog.GoalScopeOptionalPreserve && request.Goal.Validate() == nil { if prior.GoalID != request.Goal.ID || prior.GoalKind != request.Goal.Kind || prior.DeliveryID != request.Goal.DeliveryID { return fmt.Errorf("idempotency receipt belongs to a different configured goal") diff --git a/boatstack/internal/kernel/engine/engine_test.go b/boatstack/internal/kernel/engine/engine_test.go index 4175eb4..a028163 100644 --- a/boatstack/internal/kernel/engine/engine_test.go +++ b/boatstack/internal/kernel/engine/engine_test.go @@ -156,15 +156,17 @@ func observation(phase model.ProtocolPhase, fingerprint string) model.Observatio e := model.Evidence{Source: "fixture", Fingerprint: fingerprint, ObservedAt: time.Unix(20, 0).UTC()} configurationEvidence := model.Evidence{Source: "configuration:/repo/.boatstack/project.json", Fingerprint: "config-fingerprint", ObservedAt: time.Unix(20, 0).UTC()} stage := "start" + revision := uint64(1) if phase == model.PhaseActive { stage = "terminal" + revision = 2 } else if phase == model.PhaseRecovery { stage = "verify" } return model.Observation{ - SchemaVersion: model.SnapshotSchemaVersion, - Invocation: model.InvocationContext{RepositoryID: "repo", GitCommonID: "git", WorktreeID: "wt", Ref: "refs/heads/f", ControllerID: "ctl", InvokingPath: fixtureAbsolutePath("test-fixture", "repo"), RuntimeVersion: "runtime-version", RuntimePath: fixtureAbsolutePath("test-fixture", "runtime"), RuntimeFingerprint: "runtime", Topology: model.TopologyEmbedded, Host: "cli", Correlation: "corr"}, - Phase: model.Known(phase, e), Engagement: model.Known(model.EngagementActive, e), Delivery: model.Known(model.DeliveryActive, e), Workspace: model.Known(model.WorkspaceActive, e), + SchemaVersion: model.SnapshotSchemaVersion, StateRevision: revision, + Invocation: model.InvocationContext{RepositoryID: "repo", GitCommonID: "git", WorktreeID: "wt", Ref: "refs/heads/f", ControllerID: "ctl", InvokingPath: fixtureAbsolutePath("test-fixture", "repo"), RuntimeVersion: "runtime-version", RuntimePath: fixtureAbsolutePath("test-fixture", "runtime"), RuntimeFingerprint: "runtime", Topology: model.TopologyEmbedded, Host: "cli", Correlation: "corr"}, + Phase: model.Known(phase, e), Engagement: model.Known(model.EngagementActive, e), Delivery: model.Known(model.DeliveryActive, e), Workspace: model.Known(model.WorkspaceActive, e), Plan: model.Known(model.PlanApproved, e), Configuration: model.Known(model.ConfigurationVerified, configurationEvidence), Runtime: model.Known(model.RuntimeVerified, e), ConfigurationPolicy: model.Known(model.ConfigurationPolicy{PlanApproval: "human", VisualEvidence: "optional", ExternalEffectAuthority: "human-or-autonomy-plus-provider", Hosts: []string{"cli"}}, configurationEvidence), Publication: model.Known(model.PublicationNone, e), Verification: model.Known(model.VerificationUnverified, e), Recovery: model.Known(model.RecoveryNone, e), @@ -176,6 +178,7 @@ func observation(phase model.ProtocolPhase, fingerprint string) model.Observatio func recoveryObservation(fingerprint string) model.Observation { value := observation(model.PhaseRecovery, fingerprint) + value.StateRevision = 2 evidence := value.Phase.Evidence[0] value.Recovery = model.Known(model.RecoveryReconcile, evidence) value.Transaction = model.Known(model.TransactionLocalApplied, evidence) @@ -237,12 +240,22 @@ func testRegistryWithAdvanceClass(t *testing.T, class catalog.EventClass) catalo return r } -func request(now time.Time) ApplyRequest { +func request(t *testing.T, now time.Time) ApplyRequest { + t.Helper() invocation := observation(model.PhaseObserved, "source").Invocation + snapshot, err := model.CanonicalizeForProgram(observation(model.PhaseObserved, "source"), syntheticProgramFingerprint) + if err != nil { + t.Fatal(err) + } + transition, _ := testRegistry(t).Lookup("test.advance") + prescription, err := protocol.NewPrescription(snapshot, transition) + if err != nil { + t.Fatal(err) + } return ApplyRequest{ResolveRequest: ResolveRequest{ Invocation: invocation, Goal: model.Goal{ID: "goal", Kind: model.GoalVerified, DeliveryID: "delivery"}, Requested: "test.advance", Authority: protocol.AuthorityBundle{Receipts: []protocol.AuthorityReceipt{{ID: "auth", Class: catalog.AuthorityRepository, Subject: "repo", Fingerprint: "config-fingerprint", IssuedAt: now.Add(-time.Minute), ExpiresAt: now.Add(time.Hour)}}}, - }, FlowID: "flow", AdmissionLifetime: time.Minute} + }, FlowID: "flow", Prescription: prescription, AdmissionLifetime: time.Minute} } func TestRequiredObserverFailureReturnsTypedUnresolvedDecision(t *testing.T) { @@ -258,7 +271,7 @@ func TestRequiredObserverFailureReturnsTypedUnresolvedDecision(t *testing.T) { t.Fatal(err) } - resolved, resolveErr := kernel.Resolve(context.Background(), request(now).ResolveRequest) + resolved, resolveErr := kernel.Resolve(context.Background(), request(t, now).ResolveRequest) if resolveErr == nil || !strings.Contains(resolveErr.Error(), "observer unavailable") { t.Fatalf("resolve error = %v, want observer failure", resolveErr) } @@ -266,7 +279,7 @@ func TestRequiredObserverFailureReturnsTypedUnresolvedDecision(t *testing.T) { t.Fatalf("resolve decision = %+v, want typed UNRESOLVED", resolved.Decision) } - applied, applyErr := kernel.Apply(context.Background(), request(now)) + applied, applyErr := kernel.Apply(context.Background(), request(t, now)) if applyErr == nil || !strings.Contains(applyErr.Error(), "observer unavailable") { t.Fatalf("apply error = %v, want observer failure", applyErr) } @@ -296,7 +309,7 @@ func TestResolutionDoesNotPrescribeBeforeRequiredParametersAreBound(t *testing.T if err != nil { t.Fatal(err) } - req := request(now).ResolveRequest + req := request(t, now).ResolveRequest req.Requested = "" candidate, err := kernel.Resolve(context.Background(), req) if err != nil { @@ -326,7 +339,7 @@ func TestResolutionDoesNotPrescribeAnEffectThatDeterministicPreflightRejects(t * if err != nil { t.Fatal(err) } - resolved, err := kernel.Resolve(context.Background(), request(now).ResolveRequest) + resolved, err := kernel.Resolve(context.Background(), request(t, now).ResolveRequest) if err != nil { t.Fatal(err) } @@ -347,14 +360,14 @@ func TestApplyCrossesAdmissionEffectVerificationAndReceiptBoundary(t *testing.T) if err != nil { t.Fatal(err) } - result, err := kernel.Apply(context.Background(), request(now)) + result, err := kernel.Apply(context.Background(), request(t, now)) if err != nil { t.Fatal(err) } if effects.executions != 1 || effects.rollbacks != 0 || journal.committed != 1 || journal.aborted != 0 || len(receipts.values) != 1 || result.Receipt.ID == "" || !lock.released { t.Fatalf("unexpected boundary evidence: effects=%+v journal=%+v receipts=%d receipt=%q released=%v", effects, journal, len(receipts.values), result.Receipt.ID, lock.released) } - retry := request(now) + retry := request(t, now) retry.IdempotencyKey = result.Admission.IdempotencyKey replayed, err := kernel.Apply(context.Background(), retry) if err != nil { @@ -449,11 +462,11 @@ func TestIdempotencyReceiptCannotHideUncommittedRecoveryJournal(t *testing.T) { if err != nil { t.Fatal(err) } - completed, err := kernel.Apply(context.Background(), request(now)) + completed, err := kernel.Apply(context.Background(), request(t, now)) if err != nil { t.Fatal(err) } - retry := request(now) + retry := request(t, now) retry.IdempotencyKey = completed.Admission.IdempotencyKey _, err = kernel.Apply(context.Background(), retry) var recovery ReplayRecoveryError @@ -471,8 +484,8 @@ func TestApplyRejectsSnapshotDriftBeforeEffect(t *testing.T) { observer := &sequenceObserver{items: []model.Observation{observation(model.PhaseObserved, "source"), observation(model.PhaseObserved, "drifted")}} journal, effects, receipts, lock := &fakeJournal{}, &fakeEffects{}, &memoryReceipts{}, &fakeLock{} kernel, _ := New(testRegistry(t), syntheticGoalContracts(t), syntheticProgramFingerprint, observer, fixedClock{now}, fakeLocker{lock}, journal, effects, receipts) - _, err := kernel.Apply(context.Background(), request(now)) - var stale StaleAdmissionError + _, err := kernel.Apply(context.Background(), request(t, now)) + var stale StalePrescriptionError if !errors.As(err, &stale) { t.Fatalf("error = %v, want StaleAdmissionError", err) } @@ -481,13 +494,31 @@ func TestApplyRejectsSnapshotDriftBeforeEffect(t *testing.T) { } } +func TestApplyRejectsHumanRevisionAdvanceBeforeEffect(t *testing.T) { + // control-law: a later human commit invalidates an older agent prescription + now := time.Unix(30, 0).UTC() + advanced := observation(model.PhaseObserved, "source") + advanced.StateRevision = 2 + observer := &sequenceObserver{items: []model.Observation{observation(model.PhaseObserved, "source"), advanced}} + journal, effects, receipts, lock := &fakeJournal{}, &fakeEffects{}, &memoryReceipts{}, &fakeLock{} + kernel, _ := New(testRegistry(t), syntheticGoalContracts(t), syntheticProgramFingerprint, observer, fixedClock{now}, fakeLocker{lock}, journal, effects, receipts) + _, err := kernel.Apply(context.Background(), request(t, now)) + var stale StalePrescriptionError + if !errors.As(err, &stale) || stale.ExpectedStateRevision != 1 || stale.ObservedStateRevision != 2 { + t.Fatalf("error = %v, want state-revision stale prescription", err) + } + if effects.executions != 0 || journal.begun != 0 || len(receipts.values) != 0 { + t.Fatalf("stale revision crossed effect boundary: effects=%d journals=%d receipts=%d", effects.executions, journal.begun, len(receipts.values)) + } +} + func TestApplyRollsBackFailedPostconditionAndDoesNotReceipt(t *testing.T) { // control-law: successful-effect-call-is-not-transition-success now := time.Unix(30, 0).UTC() observer := &sequenceObserver{items: []model.Observation{observation(model.PhaseObserved, "source"), observation(model.PhaseObserved, "source"), observation(model.PhaseObserved, "unchanged")}} journal, effects, receipts, lock := &fakeJournal{}, &fakeEffects{result: ports.EffectResult{Settlement: ports.EffectSettled}}, &memoryReceipts{}, &fakeLock{} kernel, _ := New(testRegistry(t), syntheticGoalContracts(t), syntheticProgramFingerprint, observer, fixedClock{now}, fakeLocker{lock}, journal, effects, receipts) - _, err := kernel.Apply(context.Background(), request(now)) + _, err := kernel.Apply(context.Background(), request(t, now)) var postcondition PostconditionError if !errors.As(err, &postcondition) { t.Fatalf("error = %v, want PostconditionError", err) @@ -504,7 +535,7 @@ func TestApplyRequiresRecoveryWhenJournalFailsAfterEffect(t *testing.T) { journal := &fakeJournal{failMark: "verifying"} effects, receipts, lock := &fakeEffects{result: ports.EffectResult{Settlement: ports.EffectSettled}}, &memoryReceipts{}, &fakeLock{} kernel, _ := New(testRegistry(t), syntheticGoalContracts(t), syntheticProgramFingerprint, observer, fixedClock{now}, fakeLocker{lock}, journal, effects, receipts) - _, err := kernel.Apply(context.Background(), request(now)) + _, err := kernel.Apply(context.Background(), request(t, now)) if err == nil || !strings.Contains(err.Error(), "injected journal mark failure") { t.Fatalf("error=%v, want injected post-effect journal failure", err) } @@ -519,7 +550,7 @@ func TestApplyPreservesUnknownExternalOutcomeForReconciliation(t *testing.T) { observer := &sequenceObserver{items: []model.Observation{observation(model.PhaseObserved, "source"), observation(model.PhaseObserved, "source"), observation(model.PhaseObserved, "unchanged")}} journal, effects, receipts, lock := &fakeJournal{}, &fakeEffects{result: ports.EffectResult{Settlement: ports.EffectUnknown}}, &memoryReceipts{}, &fakeLock{} kernel, _ := New(testRegistry(t), syntheticGoalContracts(t), syntheticProgramFingerprint, observer, fixedClock{now}, fakeLocker{lock}, journal, effects, receipts) - _, err := kernel.Apply(context.Background(), request(now)) + _, err := kernel.Apply(context.Background(), request(t, now)) var unknown ExternalOutcomeUnknownError if !errors.As(err, &unknown) { t.Fatalf("error=%v, want ExternalOutcomeUnknownError", err) @@ -539,7 +570,7 @@ func TestOwnedExternalExecutionErrorRequiresRecoveryWithoutRollback(t *testing.T if err != nil { t.Fatal(err) } - apply := request(now) + apply := request(t, now) apply.Authority.Receipts[0].Class = catalog.AuthorityHuman _, err = kernel.Apply(context.Background(), apply) var unknown ExternalOutcomeUnknownError diff --git a/boatstack/internal/kernel/model/state.go b/boatstack/internal/kernel/model/state.go index 4f50305..0db00c1 100644 --- a/boatstack/internal/kernel/model/state.go +++ b/boatstack/internal/kernel/model/state.go @@ -9,7 +9,7 @@ import ( "time" ) -const SnapshotSchemaVersion = 2 +const SnapshotSchemaVersion = 3 type ProtocolPhase string @@ -389,6 +389,7 @@ func (s ProgramState) Valid() bool { // fingerprinting. type Observation struct { SchemaVersion int `json:"schema_version"` + StateRevision uint64 `json:"state_revision"` ProgramFingerprint string `json:"program_fingerprint,omitempty"` RecordedProgramFingerprint string `json:"recorded_program_fingerprint,omitempty"` Invocation InvocationContext `json:"invocation"` @@ -448,6 +449,9 @@ func Canonicalize(observation Observation) (Snapshot, error) { if observation.SchemaVersion != SnapshotSchemaVersion { return Snapshot{}, fmt.Errorf("snapshot: schema version %d, want %d", observation.SchemaVersion, SnapshotSchemaVersion) } + if observation.StateRevision == 0 { + return Snapshot{}, fmt.Errorf("snapshot: durable state revision is required") + } if observation.Program.Status == "" && observation.ProgramFingerprint == "" { observation.Program = Known(ProgramUnbound, Evidence{Source: "control-program:unbound", Fingerprint: "unbound", ObservedAt: observation.ObservedAt}) } diff --git a/boatstack/internal/kernel/model/state_test.go b/boatstack/internal/kernel/model/state_test.go index 9cceb61..949a639 100644 --- a/boatstack/internal/kernel/model/state_test.go +++ b/boatstack/internal/kernel/model/state_test.go @@ -21,7 +21,7 @@ func testEvidence() Evidence { func testObservation(phase ProtocolPhase) Observation { evidence := testEvidence() return Observation{ - SchemaVersion: SnapshotSchemaVersion, + SchemaVersion: SnapshotSchemaVersion, StateRevision: 1, Invocation: InvocationContext{ RepositoryID: "repo-1", GitCommonID: "git-1", WorktreeID: "worktree-1", Ref: "refs/heads/feature", ControllerID: "controller-1", InvokingPath: testAbsolutePath("test-fixture", "repo"), RuntimeVersion: "runtime-version", RuntimePath: testAbsolutePath("test-fixture", "runtime", "boatstack"), RuntimeFingerprint: "runtime-fingerprint", diff --git a/boatstack/internal/kernel/protocol/admission.go b/boatstack/internal/kernel/protocol/admission.go index d1ce452..ecd777b 100644 --- a/boatstack/internal/kernel/protocol/admission.go +++ b/boatstack/internal/kernel/protocol/admission.go @@ -9,33 +9,35 @@ import ( "github.com/operatorstack/boatstack/boatstack/internal/kernel/model" ) -const AdmissionSchemaVersion = 2 +const AdmissionSchemaVersion = 3 type Admission struct { - SchemaVersion int `json:"schema_version"` - ID string `json:"id"` - TransitionID catalog.TransitionID `json:"transition_id"` - TransitionVersion int `json:"transition_version"` - ProgramFingerprint string `json:"program_fingerprint"` - PriorProgramFingerprint string `json:"prior_program_fingerprint,omitempty"` - ProgramDeltaFingerprint string `json:"program_delta_fingerprint,omitempty"` - SnapshotFingerprint string `json:"snapshot_fingerprint"` - SourceRevision string `json:"source_revision,omitempty"` - WorktreeFingerprint string `json:"worktree_fingerprint,omitempty"` - SourcePhase model.ProtocolPhase `json:"source_phase"` - Invocation model.InvocationContext `json:"invocation"` - Goal model.Goal `json:"goal"` - GoalScope catalog.GoalScope `json:"goal_scope,omitempty"` - GoalStatus model.FactStatus `json:"goal_status,omitempty"` - Authority AuthorityBundle `json:"authority"` - Parameters Parameters `json:"parameters,omitempty"` - Evidence []string `json:"evidence"` - IdempotencyKey string `json:"idempotency_key"` - IssuedAt time.Time `json:"issued_at"` - ExpiresAt time.Time `json:"expires_at"` + SchemaVersion int `json:"schema_version"` + ID string `json:"id"` + PrescriptionID string `json:"prescription_id"` + TransitionID catalog.TransitionID `json:"transition_id"` + TransitionVersion int `json:"transition_version"` + ExpectedStateRevision uint64 `json:"expected_state_revision"` + ExpectedProgramFingerprint string `json:"expected_program_fingerprint"` + PriorProgramFingerprint string `json:"prior_program_fingerprint,omitempty"` + ProgramDeltaFingerprint string `json:"program_delta_fingerprint,omitempty"` + ExpectedSnapshotFingerprint string `json:"expected_snapshot_fingerprint"` + SourceRevision string `json:"source_revision,omitempty"` + WorktreeFingerprint string `json:"worktree_fingerprint,omitempty"` + SourcePhase model.ProtocolPhase `json:"source_phase"` + Invocation model.InvocationContext `json:"invocation"` + Goal model.Goal `json:"goal"` + GoalScope catalog.GoalScope `json:"goal_scope,omitempty"` + GoalStatus model.FactStatus `json:"goal_status,omitempty"` + Authority AuthorityBundle `json:"authority"` + Parameters Parameters `json:"parameters,omitempty"` + Evidence []string `json:"evidence"` + IdempotencyKey string `json:"idempotency_key"` + IssuedAt time.Time `json:"issued_at"` + ExpiresAt time.Time `json:"expires_at"` } -func NewAdmission(snapshot model.Snapshot, goal model.Goal, transition catalog.Transition, authority AuthorityBundle, parameters Parameters, now time.Time, lifetime time.Duration) (Admission, error) { +func NewAdmission(snapshot model.Snapshot, goal model.Goal, transition catalog.Transition, prescription Prescription, authority AuthorityBundle, parameters Parameters, now time.Time, lifetime time.Duration) (Admission, error) { var err error goal, err = GoalForTransition(snapshot, goal, transition) if err != nil { @@ -47,10 +49,14 @@ func NewAdmission(snapshot model.Snapshot, goal model.Goal, transition catalog.T if lifetime <= 0 { return Admission{}, fmt.Errorf("admission lifetime must be positive") } + if err := prescription.ValidateCurrent(snapshot, transition); err != nil { + return Admission{}, err + } sourceRevision, worktreeFingerprint := gitBinding(snapshot) a := Admission{ - SchemaVersion: AdmissionSchemaVersion, TransitionID: transition.ID, TransitionVersion: transition.Version, - ProgramFingerprint: snapshot.ProgramFingerprint, SnapshotFingerprint: snapshot.Fingerprint, SourceRevision: sourceRevision, WorktreeFingerprint: worktreeFingerprint, + SchemaVersion: AdmissionSchemaVersion, PrescriptionID: prescription.ID, TransitionID: transition.ID, TransitionVersion: transition.Version, + ExpectedStateRevision: prescription.ExpectedStateRevision, ExpectedProgramFingerprint: prescription.ExpectedProgramFingerprint, + ExpectedSnapshotFingerprint: prescription.ExpectedSnapshotFingerprint, SourceRevision: sourceRevision, WorktreeFingerprint: worktreeFingerprint, SourcePhase: snapshot.Phase.Value, Invocation: snapshot.Invocation, Goal: goal, GoalScope: transition.Policy.GoalScope, Authority: authority.canonical(), Evidence: append([]string(nil), transition.RequiredEvidence...), Parameters: parameters.Canonical(), IssuedAt: now.UTC(), ExpiresAt: now.Add(lifetime).UTC(), } @@ -59,7 +65,7 @@ func NewAdmission(snapshot model.Snapshot, goal model.Goal, transition catalog.T } if snapshot.RecordedProgramFingerprint != "" && snapshot.RecordedProgramFingerprint != snapshot.ProgramFingerprint { a.PriorProgramFingerprint = snapshot.RecordedProgramFingerprint - delta, err := ProgramDeltaFingerprint(snapshot.RecordedProgramFingerprint, snapshot.ProgramFingerprint) + delta, err := ProgramDeltaFingerprint(snapshot.RecordedProgramFingerprint, prescription.ExpectedProgramFingerprint) if err != nil { return Admission{}, err } @@ -172,10 +178,13 @@ func (a Admission) ValidateCurrent(snapshot model.Snapshot, goal model.Goal, tra if a.GoalScope != transition.Policy.GoalScope { return fmt.Errorf("admission %q is bound to a different product-goal scope", a.ID) } - if a.SnapshotFingerprint != snapshot.Fingerprint { + if a.ExpectedStateRevision != snapshot.StateRevision { + return fmt.Errorf("admission %q is stale: state revision changed", a.ID) + } + if a.ExpectedSnapshotFingerprint != snapshot.Fingerprint { return fmt.Errorf("admission %q is stale: snapshot changed", a.ID) } - if a.ProgramFingerprint != snapshot.ProgramFingerprint { + if a.ExpectedProgramFingerprint != snapshot.ProgramFingerprint { return fmt.Errorf("admission %q is bound to a different control program", a.ID) } expectedPrior := "" @@ -186,7 +195,7 @@ func (a Admission) ValidateCurrent(snapshot model.Snapshot, goal model.Goal, tra return fmt.Errorf("admission %q is bound to a different prior control program", a.ID) } if a.PriorProgramFingerprint != "" { - delta, err := ProgramDeltaFingerprint(a.PriorProgramFingerprint, a.ProgramFingerprint) + delta, err := ProgramDeltaFingerprint(a.PriorProgramFingerprint, a.ExpectedProgramFingerprint) if err != nil || delta != a.ProgramDeltaFingerprint { return fmt.Errorf("admission %q has an invalid program delta binding", a.ID) } @@ -335,14 +344,14 @@ func validateAuthorityEvidence(snapshot model.Snapshot, authority AuthorityBundl } func (a Admission) ValidateIdentity() error { - if a.SchemaVersion != AdmissionSchemaVersion || a.ID == "" || a.TransitionID == "" || a.TransitionVersion < 1 || len(a.ProgramFingerprint) != 64 || a.SnapshotFingerprint == "" || !a.SourcePhase.Valid() || a.IdempotencyKey == "" || a.IssuedAt.IsZero() || a.ExpiresAt.Before(a.IssuedAt) { + if a.SchemaVersion != AdmissionSchemaVersion || a.ID == "" || a.PrescriptionID == "" || a.TransitionID == "" || a.TransitionVersion < 1 || a.ExpectedStateRevision == 0 || len(a.ExpectedProgramFingerprint) != 64 || len(a.ExpectedSnapshotFingerprint) != 64 || !a.SourcePhase.Valid() || a.IdempotencyKey == "" || a.IssuedAt.IsZero() || a.ExpiresAt.Before(a.IssuedAt) { return fmt.Errorf("admission: invalid schema, identity, source, or lifetime") } if (a.PriorProgramFingerprint == "") != (a.ProgramDeltaFingerprint == "") { return fmt.Errorf("admission has incomplete program delta identity") } if a.PriorProgramFingerprint != "" { - delta, err := ProgramDeltaFingerprint(a.PriorProgramFingerprint, a.ProgramFingerprint) + delta, err := ProgramDeltaFingerprint(a.PriorProgramFingerprint, a.ExpectedProgramFingerprint) if err != nil || delta != a.ProgramDeltaFingerprint { return fmt.Errorf("admission has invalid program delta identity") } diff --git a/boatstack/internal/kernel/protocol/journal.go b/boatstack/internal/kernel/protocol/journal.go new file mode 100644 index 0000000..e0ed121 --- /dev/null +++ b/boatstack/internal/kernel/protocol/journal.go @@ -0,0 +1,5 @@ +package protocol + +// JournalSchemaVersion identifies the transaction record that embeds an exact +// prescription-bound admission. +const JournalSchemaVersion = 3 diff --git a/boatstack/internal/kernel/protocol/prescription.go b/boatstack/internal/kernel/protocol/prescription.go new file mode 100644 index 0000000..e40953c --- /dev/null +++ b/boatstack/internal/kernel/protocol/prescription.go @@ -0,0 +1,86 @@ +package protocol + +import ( + "fmt" + + "github.com/operatorstack/boatstack/boatstack/internal/kernel/catalog" + "github.com/operatorstack/boatstack/boatstack/internal/kernel/model" +) + +const PrescriptionSchemaVersion = 1 + +// Prescription is the immutable compare-and-swap binding emitted by +// resolution and required by apply. It carries no authority. +type Prescription struct { + SchemaVersion int `json:"schema_version"` + ID string `json:"id"` + TransitionID catalog.TransitionID `json:"transition_id"` + ExpectedStateRevision uint64 `json:"expected_state_revision"` + ExpectedProgramFingerprint string `json:"expected_program_fingerprint"` + ExpectedSnapshotFingerprint string `json:"expected_snapshot_fingerprint"` +} + +func NewPrescription(snapshot model.Snapshot, transition catalog.Transition) (Prescription, error) { + prescription := Prescription{ + SchemaVersion: PrescriptionSchemaVersion, + TransitionID: transition.ID, + ExpectedStateRevision: snapshot.StateRevision, + ExpectedProgramFingerprint: snapshot.ProgramFingerprint, + ExpectedSnapshotFingerprint: snapshot.Fingerprint, + } + if err := prescription.validateFields(); err != nil { + return Prescription{}, err + } + identity := prescription + identity.ID = "" + var err error + prescription.ID, err = contentID("prx-", identity) + if err != nil { + return Prescription{}, err + } + return prescription, nil +} + +func (p Prescription) Validate() error { + if err := p.validateFields(); err != nil { + return err + } + identity := p + want := identity.ID + identity.ID = "" + got, err := contentID("prx-", identity) + if err != nil { + return err + } + if want == "" || got != want { + return fmt.Errorf("prescription failed content identity verification") + } + return nil +} + +func (p Prescription) validateFields() error { + if p.SchemaVersion != PrescriptionSchemaVersion || p.TransitionID == "" || p.ExpectedStateRevision == 0 || + len(p.ExpectedProgramFingerprint) != 64 || len(p.ExpectedSnapshotFingerprint) != 64 { + return fmt.Errorf("prescription has invalid schema, transition, state revision, program, or snapshot identity") + } + return nil +} + +func (p Prescription) ValidateCurrent(snapshot model.Snapshot, transition catalog.Transition) error { + if err := p.Validate(); err != nil { + return err + } + if p.TransitionID != transition.ID { + return fmt.Errorf("prescription %q is bound to transition %q, not %q", p.ID, p.TransitionID, transition.ID) + } + if p.ExpectedStateRevision != snapshot.StateRevision { + return fmt.Errorf("prescription %q expected state revision %d, observed %d", p.ID, p.ExpectedStateRevision, snapshot.StateRevision) + } + if p.ExpectedProgramFingerprint != snapshot.ProgramFingerprint { + return fmt.Errorf("prescription %q expected control program %s, observed %s", p.ID, p.ExpectedProgramFingerprint, snapshot.ProgramFingerprint) + } + if p.ExpectedSnapshotFingerprint != snapshot.Fingerprint { + return fmt.Errorf("prescription %q expected snapshot %s, observed %s", p.ID, p.ExpectedSnapshotFingerprint, snapshot.Fingerprint) + } + return nil +} diff --git a/boatstack/internal/kernel/protocol/prescription_test.go b/boatstack/internal/kernel/protocol/prescription_test.go new file mode 100644 index 0000000..2de5bfc --- /dev/null +++ b/boatstack/internal/kernel/protocol/prescription_test.go @@ -0,0 +1,51 @@ +package protocol + +import ( + "strings" + "testing" + + "github.com/operatorstack/boatstack/boatstack/internal/kernel/catalog" + "github.com/operatorstack/boatstack/boatstack/internal/kernel/model" +) + +func TestPrescriptionContentIdentityBindsTransitionStateProgramAndSnapshot(t *testing.T) { + // control-law: resolution emits one immutable state-program CAS identity + base := model.Snapshot{ + Observation: model.Observation{StateRevision: 41, ProgramFingerprint: strings.Repeat("a", 64)}, + Fingerprint: strings.Repeat("b", 64), + } + transition := catalog.Transition{ID: "program/advance"} + one, err := NewPrescription(base, transition) + if err != nil { + t.Fatal(err) + } + if err := one.ValidateCurrent(base, transition); err != nil { + t.Fatal(err) + } + mutations := []struct { + name string + snapshot model.Snapshot + transition catalog.Transition + }{ + {name: "state", snapshot: model.Snapshot{Observation: model.Observation{StateRevision: 42, ProgramFingerprint: strings.Repeat("a", 64)}, Fingerprint: strings.Repeat("b", 64)}, transition: transition}, + {name: "program", snapshot: model.Snapshot{Observation: model.Observation{StateRevision: 41, ProgramFingerprint: strings.Repeat("c", 64)}, Fingerprint: strings.Repeat("b", 64)}, transition: transition}, + {name: "snapshot", snapshot: model.Snapshot{Observation: model.Observation{StateRevision: 41, ProgramFingerprint: strings.Repeat("a", 64)}, Fingerprint: strings.Repeat("d", 64)}, transition: transition}, + {name: "transition", snapshot: base, transition: catalog.Transition{ID: "program/other"}}, + } + for _, mutation := range mutations { + t.Run(mutation.name, func(t *testing.T) { + other, err := NewPrescription(mutation.snapshot, mutation.transition) + if err != nil { + t.Fatal(err) + } + if other.ID == one.ID { + t.Fatal("load-bearing prescription change preserved content identity") + } + }) + } + tampered := one + tampered.ExpectedStateRevision++ + if err := tampered.Validate(); err == nil { + t.Fatal("tampered prescription retained validity") + } +} diff --git a/boatstack/internal/kernel/protocol/receipt.go b/boatstack/internal/kernel/protocol/receipt.go index f7f6fde..26a1ac6 100644 --- a/boatstack/internal/kernel/protocol/receipt.go +++ b/boatstack/internal/kernel/protocol/receipt.go @@ -8,7 +8,7 @@ import ( "github.com/operatorstack/boatstack/boatstack/internal/kernel/model" ) -const ReceiptSchemaVersion = 3 +const ReceiptSchemaVersion = 4 type Outcome string @@ -33,7 +33,10 @@ type TransitionReceipt struct { RuntimeVersion string `json:"runtime_version,omitempty"` RuntimeFingerprint string `json:"runtime_fingerprint,omitempty"` RuntimeSourceRevision string `json:"runtime_source_revision,omitempty"` + PrescriptionID string `json:"prescription_id"` AdmissionID string `json:"admission_id"` + PriorStateRevision uint64 `json:"prior_state_revision"` + ResultingStateRevision uint64 `json:"resulting_state_revision"` GoalID string `json:"goal_id"` GoalKind model.GoalKind `json:"goal_kind"` DeliveryID string `json:"delivery_id"` @@ -60,6 +63,9 @@ func NewReceipt(flowID string, sequence uint64, admission Admission, transition if completedAt.Before(startedAt) { return TransitionReceipt{}, fmt.Errorf("receipt completion precedes start") } + if admission.ExpectedStateRevision == ^uint64(0) || target.StateRevision != admission.ExpectedStateRevision+1 { + return TransitionReceipt{}, fmt.Errorf("receipt target revision must advance exactly once from the prescribed revision") + } classes := make([]string, 0, len(admission.Authority.Receipts)) for _, authority := range admission.Authority.Receipts { classes = append(classes, string(authority.Class)) @@ -70,9 +76,12 @@ func NewReceipt(flowID string, sequence uint64, admission Admission, transition } receipt := TransitionReceipt{ SchemaVersion: ReceiptSchemaVersion, FlowID: flowID, Sequence: sequence, TransitionID: transition.ID, - TransitionVersion: transition.Version, ProgramFingerprint: admission.ProgramFingerprint, AdmissionID: admission.ID, GoalID: admission.Goal.ID, GoalKind: admission.Goal.Kind, DeliveryID: admission.Goal.DeliveryID, + TransitionVersion: transition.Version, ProgramFingerprint: admission.ExpectedProgramFingerprint, + PrescriptionID: admission.PrescriptionID, AdmissionID: admission.ID, + PriorStateRevision: admission.ExpectedStateRevision, ResultingStateRevision: target.StateRevision, + GoalID: admission.Goal.ID, GoalKind: admission.Goal.Kind, DeliveryID: admission.Goal.DeliveryID, GoalScope: admission.GoalScope, GoalStatus: admission.GoalStatus, - SourceFingerprint: admission.SnapshotFingerprint, TargetFingerprint: target.Fingerprint, + SourceFingerprint: admission.ExpectedSnapshotFingerprint, TargetFingerprint: target.Fingerprint, AuthorityClasses: classes, IdempotencyKey: admission.IdempotencyKey, Verifier: transition.Verifier, Outcome: outcome, Recovery: transition.Interruption.Recovery, Terminal: terminal, StartedAt: startedAt.UTC(), CompletedAt: completedAt.UTC(), DurationNanoseconds: completedAt.Sub(startedAt).Nanoseconds(), @@ -102,7 +111,7 @@ func NewReceipt(flowID string, sequence uint64, admission Admission, transition } func (r TransitionReceipt) Validate() error { - if r.SchemaVersion != ReceiptSchemaVersion || r.ID == "" || r.FlowID == "" || r.Sequence == 0 || r.TransitionID == "" || r.TransitionVersion < 1 || len(r.ProgramFingerprint) != 64 || r.AdmissionID == "" || r.SourceFingerprint == "" || r.TargetFingerprint == "" || r.IdempotencyKey == "" || r.Verifier == "" { + if r.SchemaVersion != ReceiptSchemaVersion || r.ID == "" || r.FlowID == "" || r.Sequence == 0 || r.TransitionID == "" || r.TransitionVersion < 1 || len(r.ProgramFingerprint) != 64 || r.PrescriptionID == "" || r.AdmissionID == "" || r.PriorStateRevision == 0 || r.PriorStateRevision == ^uint64(0) || r.ResultingStateRevision == 0 || r.ResultingStateRevision != r.PriorStateRevision+1 || r.SourceFingerprint == "" || r.TargetFingerprint == "" || r.IdempotencyKey == "" || r.Verifier == "" { return fmt.Errorf("receipt has incomplete identity or evidence") } if !r.GoalScope.Valid() { diff --git a/boatstack/internal/plant/observer.go b/boatstack/internal/plant/observer.go index fc84ee2..86f93b5 100644 --- a/boatstack/internal/plant/observer.go +++ b/boatstack/internal/plant/observer.go @@ -259,7 +259,7 @@ func (o Observer) Observe(ctx context.Context, request ports.ObservationRequest) goalFact = model.Fact[model.Goal]{Status: model.FactKnown, Value: state.Goal, Evidence: stateEvidence} } return model.Observation{ - SchemaVersion: model.SnapshotSchemaVersion, RecordedProgramFingerprint: recordedProgramFingerprint, Invocation: current, + SchemaVersion: model.SnapshotSchemaVersion, StateRevision: state.Revision, RecordedProgramFingerprint: recordedProgramFingerprint, Invocation: current, Phase: model.Fact[model.ProtocolPhase]{Status: model.FactKnown, Value: phase, Evidence: stateEvidence}, Engagement: model.Fact[model.EngagementState]{Status: model.FactKnown, Value: state.Engagement, Evidence: stateEvidence}, Delivery: model.Fact[model.DeliveryState]{Status: model.FactKnown, Value: delivery, Evidence: deliveryEvidence}, @@ -614,11 +614,11 @@ type pendingJournalHeader struct { Status string `json:"status"` Reason string `json:"reason"` Admission struct { - ID string `json:"id"` - ProgramFingerprint string `json:"program_fingerprint"` - SourcePhase model.ProtocolPhase `json:"source_phase"` - Invocation model.InvocationContext `json:"invocation"` - Parameters protocol.Parameters `json:"parameters"` + ID string `json:"id"` + ExpectedProgramFingerprint string `json:"expected_program_fingerprint"` + SourcePhase model.ProtocolPhase `json:"source_phase"` + Invocation model.InvocationContext `json:"invocation"` + Parameters protocol.Parameters `json:"parameters"` } `json:"admission"` Mutations []struct { Path string `json:"path"` @@ -661,7 +661,7 @@ func pendingJournalEvidence(root, ignoreAdmissionID string, now time.Time) (pend return pendingJournalSet{}, readErr } var header pendingJournalHeader - if json.Unmarshal(raw, &header) != nil || header.SchemaVersion != 2 || header.TransitionClass != string(catalog.EventRecovery) { + if json.Unmarshal(raw, &header) != nil || header.SchemaVersion != protocol.JournalSchemaVersion || header.TransitionClass != string(catalog.EventRecovery) { continue } if transactionID, ok := header.Admission.Parameters.Get("transaction_id"); ok { @@ -682,7 +682,7 @@ func pendingJournalEvidence(root, ignoreAdmissionID string, now time.Time) (pend return pendingJournalSet{}, err } class := catalog.EventClass(header.TransitionClass) - if header.SchemaVersion != 2 || header.Admission.ID == "" || len(header.Admission.ProgramFingerprint) != 64 || entry.Name() != header.Admission.ID+".pending" || header.TransitionID == "" || header.Status == "" || !class.Valid() || !class.Controllable() { + if header.SchemaVersion != protocol.JournalSchemaVersion || header.Admission.ID == "" || len(header.Admission.ExpectedProgramFingerprint) != 64 || entry.Name() != header.Admission.ID+".pending" || header.TransitionID == "" || header.Status == "" || !class.Valid() || !class.Controllable() { return pendingJournalSet{}, fmt.Errorf("invalid pending transaction journal %s", path) } if header.Admission.ID == ignoreAdmissionID { @@ -716,7 +716,7 @@ func pendingJournalEvidence(root, ignoreAdmissionID string, now time.Time) (pend } set := pendingJournalSet{ Found: true, Evidence: []model.Evidence{evidence}, TransactionState: transactionState, - ProgramFingerprint: header.Admission.ProgramFingerprint, + ProgramFingerprint: header.Admission.ExpectedProgramFingerprint, ReconcilesProgram: header.ReconcilesProgram, Recovery: model.RecoveryContext{TransactionID: header.Admission.ID, Cause: cause, SourcePhase: header.Admission.SourcePhase, Permitted: permitted, BudgetRemaining: budget, Resumption: header.Admission.SourcePhase}, Transaction: model.TransactionContext{ID: header.Admission.ID, TransitionID: header.TransitionID, Status: header.Status, ResourceDigests: resourceDigests, ExternalPossible: external}, diff --git a/boatstack/internal/plant/observer_test.go b/boatstack/internal/plant/observer_test.go index 44bf959..5bb8808 100644 --- a/boatstack/internal/plant/observer_test.go +++ b/boatstack/internal/plant/observer_test.go @@ -13,6 +13,7 @@ import ( "github.com/operatorstack/boatstack/boatstack/internal/kernel/durable" "github.com/operatorstack/boatstack/boatstack/internal/kernel/model" "github.com/operatorstack/boatstack/boatstack/internal/kernel/ports" + "github.com/operatorstack/boatstack/boatstack/internal/kernel/protocol" boatstackruntime "github.com/operatorstack/boatstack/boatstack/internal/runtime" ) @@ -283,16 +284,16 @@ func TestRecoveryAttemptsExhaustToEscalationOnly(t *testing.T) { root := t.TempDir() originalID := "adm-interrupted" pending := map[string]any{ - "schema_version": 2, + "schema_version": protocol.JournalSchemaVersion, "transition_id": "plan.create", "transition_class": "owned-local", "status": "recovery-required", "reason": "simulated interruption", "admission": map[string]any{ - "id": originalID, - "program_fingerprint": strings.Repeat("a", 64), - "source_phase": "ACTIVE", - "invocation": map[string]any{"correlation_id": "prior-process"}, + "id": originalID, + "expected_program_fingerprint": strings.Repeat("a", 64), + "source_phase": "ACTIVE", + "invocation": map[string]any{"correlation_id": "prior-process"}, }, } writeJSON := func(name string, value any) { @@ -308,7 +309,7 @@ func TestRecoveryAttemptsExhaustToEscalationOnly(t *testing.T) { writeJSON(originalID+".pending", pending) for attempt := 1; attempt <= 3; attempt++ { aborted := map[string]any{ - "schema_version": 2, + "schema_version": protocol.JournalSchemaVersion, "transition_id": "recovery.rollback", "transition_class": "recovery", "status": "aborted", @@ -365,13 +366,13 @@ func TestInterruptedRecoveryAttemptCollapsesToEscalatableTransactionGroup(t *tes } originalID := "adm-original" write(originalID+".pending", map[string]any{ - "schema_version": 2, "transition_id": "plan.create", "transition_class": "owned-local", "status": "recovery-required", - "admission": map[string]any{"id": originalID, "program_fingerprint": strings.Repeat("a", 64), "source_phase": "ACTIVE", "invocation": map[string]any{"correlation_id": "old-process"}}, + "schema_version": protocol.JournalSchemaVersion, "transition_id": "plan.create", "transition_class": "owned-local", "status": "recovery-required", + "admission": map[string]any{"id": originalID, "expected_program_fingerprint": strings.Repeat("a", 64), "source_phase": "ACTIVE", "invocation": map[string]any{"correlation_id": "old-process"}}, }) write("adm-nested.pending", map[string]any{ - "schema_version": 2, "transition_id": "recovery.rollback", "transition_class": "recovery", "status": "verifying", + "schema_version": protocol.JournalSchemaVersion, "transition_id": "recovery.rollback", "transition_class": "recovery", "status": "verifying", "admission": map[string]any{ - "id": "adm-nested", "program_fingerprint": strings.Repeat("a", 64), "source_phase": "RECOVERY", "invocation": map[string]any{"correlation_id": "old-process"}, + "id": "adm-nested", "expected_program_fingerprint": strings.Repeat("a", 64), "source_phase": "RECOVERY", "invocation": map[string]any{"correlation_id": "old-process"}, "parameters": []map[string]string{{"name": "transaction_id", "value": originalID}}, }, }) diff --git a/boatstack/internal/surfaces/protocol.go b/boatstack/internal/surfaces/protocol.go index 1003d09..c91e1a9 100644 --- a/boatstack/internal/surfaces/protocol.go +++ b/boatstack/internal/surfaces/protocol.go @@ -11,7 +11,7 @@ import ( "github.com/operatorstack/boatstack/boatstack/internal/kernel/supervisor" ) -const SchemaVersion = 2 +const SchemaVersion = 3 type Operation string @@ -43,6 +43,7 @@ type Request struct { FlowID string `json:"flow_id,omitempty"` Goal model.Goal `json:"goal,omitempty"` TransitionID catalog.TransitionID `json:"transition_id,omitempty"` + Prescription protocol.Prescription `json:"prescription,omitempty"` Authority protocol.AuthorityBundle `json:"authority,omitempty"` RepositoryAuthority bool `json:"repository_authority,omitempty"` Parameters protocol.Parameters `json:"parameters,omitempty"` @@ -73,6 +74,12 @@ func (r Request) Validate(now time.Time) error { if r.FlowID == "" || r.TransitionID == "" { return fmt.Errorf("apply/recover request requires flow and transition identity") } + if err := r.Prescription.Validate(); err != nil { + return fmt.Errorf("apply/recover request requires an exact resolution prescription: %w", err) + } + if r.Prescription.TransitionID != r.TransitionID { + return fmt.Errorf("apply/recover transition does not match prescription") + } } if r.Operation == OperationGuard && (strings.TrimSpace(r.Command) == "" || len(r.Command) > 1<<20) { return fmt.Errorf("guard operation requires a bounded command") @@ -121,6 +128,7 @@ type Response struct { Goal model.Goal `json:"goal,omitempty"` Snapshot *model.Snapshot `json:"snapshot,omitempty"` Decision *supervisor.Decision `json:"decision,omitempty"` + Prescription *protocol.Prescription `json:"prescription,omitempty"` Admission *protocol.Admission `json:"admission,omitempty"` Receipt *protocol.TransitionReceipt `json:"receipt,omitempty"` Replayed bool `json:"replayed,omitempty"` diff --git a/boatstack/internal/surfaces/protocol_test.go b/boatstack/internal/surfaces/protocol_test.go new file mode 100644 index 0000000..6c8dc56 --- /dev/null +++ b/boatstack/internal/surfaces/protocol_test.go @@ -0,0 +1,29 @@ +package surfaces + +import ( + "testing" + "time" +) + +func TestSurfaceSchemaIsFlagDayAndApplyRequiresPrescription(t *testing.T) { + // control-law: every mutating surface reaches the exact prescription boundary + base := Request{ + SchemaVersion: SchemaVersion, + Operation: OperationResolve, + Repository: "/repository", + Host: "cli", + CorrelationID: "correlation", + } + old := base + old.SchemaVersion-- + if err := old.Validate(time.Now()); err == nil { + t.Fatal("older surface schema was accepted") + } + apply := base + apply.Operation = OperationApply + apply.FlowID = "flow" + apply.TransitionID = "engagement.begin" + if err := apply.Validate(time.Now()); err == nil { + t.Fatal("apply without an exact prescription was accepted") + } +} diff --git a/boatstack/internal/surfaces/render.go b/boatstack/internal/surfaces/render.go index 409ffc4..1827f04 100644 --- a/boatstack/internal/surfaces/render.go +++ b/boatstack/internal/surfaces/render.go @@ -2,6 +2,7 @@ package surfaces import ( "fmt" + "strconv" "strings" "github.com/operatorstack/boatstack/boatstack/internal/kernel/catalog" @@ -24,8 +25,12 @@ type CommandAST struct { Arguments []string } -func PrescriptionCommand(transition catalog.Transition, repository string, goal model.Goal, flowID string, parameters protocol.Parameters) CommandAST { - arguments := []string{"apply", "--repo", repository, "--transition", string(transition.ID), "--flow", flowID} +func PrescriptionCommand(transition catalog.Transition, prescription protocol.Prescription, correlation, repository string, goal model.Goal, flowID string, parameters protocol.Parameters) CommandAST { + arguments := []string{"apply", "--repo", repository, "--transition", string(transition.ID), "--flow", flowID, + "--correlation", correlation, "--prescription-id", prescription.ID, + "--expected-state-revision", strconv.FormatUint(prescription.ExpectedStateRevision, 10), + "--expected-program-fingerprint", prescription.ExpectedProgramFingerprint, + "--expected-snapshot-fingerprint", prescription.ExpectedSnapshotFingerprint} if goal.Validate() == nil { arguments = append(arguments, "--goal-kind", string(goal.Kind), "--delivery", goal.DeliveryID, "--goal-id", goal.ID) } @@ -63,7 +68,7 @@ type HostPrescription struct { // ProjectHostPrescription changes host capability metadata only. Every host // consumes the same semantic command, authority prompt, and postcondition. -func ProjectHostPrescription(host string, transition catalog.Transition, repository string, goal model.Goal, flowID string, parameters protocol.Parameters) (HostPrescription, error) { +func ProjectHostPrescription(host string, transition catalog.Transition, prescription protocol.Prescription, correlation, repository string, goal model.Goal, flowID string, parameters protocol.Parameters) (HostPrescription, error) { known := false for _, candidate := range CanonicalHostNames() { if host == candidate { @@ -75,7 +80,7 @@ func ProjectHostPrescription(host string, transition catalog.Transition, reposit return HostPrescription{}, fmt.Errorf("unsupported host %q", host) } return HostPrescription{ - Host: host, TransitionID: string(transition.ID), Command: PrescriptionCommand(transition, repository, goal, flowID, parameters), + Host: host, TransitionID: string(transition.ID), Command: PrescriptionCommand(transition, prescription, correlation, repository, goal, flowID, parameters), AuthorityPrompt: transition.Prescription.AuthorityPrompt, ExpectedPostcondition: transition.Prescription.ExpectedPostcondition, }, nil } diff --git a/boatstack/internal/surfaces/render_test.go b/boatstack/internal/surfaces/render_test.go index 53fa722..b58eccd 100644 --- a/boatstack/internal/surfaces/render_test.go +++ b/boatstack/internal/surfaces/render_test.go @@ -21,7 +21,14 @@ func TestShellRenderersConsumeOneCommandAST(t *testing.T) { } goal := model.Goal{ID: "goal", Kind: model.GoalVerified, DeliveryID: "delivery"} parameters := protocol.Parameters{{Name: "source_path", Value: "/tmp/O'Brien plan.md"}, {Name: "delivery_id", Value: "delivery"}} - command := PrescriptionCommand(transition, "/repo with space", goal, "flow", parameters) + prescription := protocol.Prescription{ID: "prx-fixture", ExpectedStateRevision: 41, ExpectedProgramFingerprint: strings.Repeat("a", 64), ExpectedSnapshotFingerprint: strings.Repeat("b", 64)} + command := PrescriptionCommand(transition, prescription, "corr-1", "/repo with space", goal, "flow", parameters) + joined := strings.Join(command.Arguments, " ") + for _, binding := range []string{"--correlation corr-1", "--prescription-id prx-fixture", "--expected-state-revision 41", "--expected-program-fingerprint", "--expected-snapshot-fingerprint"} { + if !strings.Contains(joined, binding) { + t.Fatalf("prescription command omitted CAS binding %q: %s", binding, joined) + } + } before := append([]string(nil), command.Arguments...) posix, err := RenderCommand(command, ShellPOSIX) if err != nil { @@ -104,6 +111,7 @@ func TestLocusModelsAreGeneratedFromEveryRuntimeTransition(t *testing.T) { func TestEveryHostConsumesOneSemanticPrescription(t *testing.T) { goal := model.Goal{ID: "goal", Kind: model.GoalVerified, DeliveryID: "delivery"} + prescription := protocol.Prescription{ID: "prx-fixture", ExpectedStateRevision: 41, ExpectedProgramFingerprint: strings.Repeat("a", 64), ExpectedSnapshotFingerprint: strings.Repeat("b", 64)} for _, transition := range testprogram.StandardRegistry().All() { if !transition.Controllable() { continue @@ -114,7 +122,7 @@ func TestEveryHostConsumesOneSemanticPrescription(t *testing.T) { } var canonical HostPrescription for index, host := range CanonicalHostNames() { - projection, err := ProjectHostPrescription(host, transition, "/repo", goal, "flow", parameters) + projection, err := ProjectHostPrescription(host, transition, prescription, "corr-1", "/repo", goal, "flow", parameters) if err != nil { t.Fatal(err) } diff --git a/boatstack/kernel.go b/boatstack/kernel.go index bd3d392..e4d9770 100644 --- a/boatstack/kernel.go +++ b/boatstack/kernel.go @@ -69,7 +69,7 @@ func NewKernel(externalStateRoot string, program control.ControlProgram) (Kernel if err != nil { return Kernel{}, err } - driver := programEffectDriver{base: baseDriver, program: program} + driver := programEffectDriver{base: baseDriver, program: program, resolver: resolver, clock: clock} registry := program.RuntimeRegistry() runtimeEngine, err := engine.New(registry, program.RuntimeGoalContracts(), program.Fingerprint(), observer, clock, locker, journal, driver, receipts) if err != nil { @@ -112,6 +112,10 @@ func (k Kernel) Handle(ctx context.Context, request surfaces.Request) (surfaces. case surfaces.OperationResolve: resolution, resolveErr := k.engine.Resolve(ctx, engine.ResolveRequest{Invocation: invocation, Goal: request.Goal, Authority: request.Authority, Parameters: request.Parameters, Requested: request.TransitionID}) response.Goal, response.Decision = resolution.Goal, &resolution.Decision + if resolution.Prescription.ID != "" { + response.Prescription = &resolution.Prescription + response.Admission = &resolution.Admission + } if resolution.Snapshot.Fingerprint != "" { response.Snapshot = &resolution.Snapshot } @@ -124,8 +128,9 @@ func (k Kernel) Handle(ctx context.Context, request surfaces.Request) (surfaces. case surfaces.OperationApply, surfaces.OperationRecover: result, applyErr := k.engine.Apply(ctx, engine.ApplyRequest{ ResolveRequest: engine.ResolveRequest{Invocation: invocation, Goal: request.Goal, Authority: request.Authority, Requested: request.TransitionID}, - FlowID: request.FlowID, Parameters: request.Parameters, IdempotencyKey: request.IdempotencyKey, AdmissionLifetime: 2 * time.Minute, + FlowID: request.FlowID, Prescription: request.Prescription, Parameters: request.Parameters, IdempotencyKey: request.IdempotencyKey, AdmissionLifetime: 2 * time.Minute, }) + response.Prescription = &request.Prescription response.Goal = result.Goal if result.Target.Fingerprint != "" { response.Snapshot = &result.Target diff --git a/boatstack/kernel_test.go b/boatstack/kernel_test.go index 5645ba1..720e8b0 100644 --- a/boatstack/kernel_test.go +++ b/boatstack/kernel_test.go @@ -8,6 +8,9 @@ import ( boatstack "github.com/operatorstack/boatstack/boatstack" "github.com/operatorstack/boatstack/boatstack/distribution" + "github.com/operatorstack/boatstack/boatstack/internal/kernel/catalog" + "github.com/operatorstack/boatstack/boatstack/internal/kernel/model" + "github.com/operatorstack/boatstack/boatstack/internal/kernel/protocol" "github.com/operatorstack/boatstack/boatstack/internal/surfaces" ) @@ -26,12 +29,21 @@ func TestRecoverSurfaceConsumesCompiledRegistryInsteadOfFixedProgramIDs(t *testi Repository: "/repository-is-not-consulted", Host: "cli", CorrelationID: "compiled-recovery", FlowID: "flow", TransitionID: "plan.create", } + prescriptionSnapshot := model.Snapshot{Observation: model.Observation{StateRevision: 1, ProgramFingerprint: program.Fingerprint()}, Fingerprint: strings.Repeat("a", 64)} + request.Prescription, err = protocol.NewPrescription(prescriptionSnapshot, catalog.Transition{ID: request.TransitionID}) + if err != nil { + t.Fatal(err) + } response, err := kernel.Handle(context.Background(), request) if err == nil || !strings.Contains(err.Error(), "compiled control program") || response.Error == "" { t.Fatalf("non-recovery transition crossed recover surface: response=%+v error=%v", response, err) } request.TransitionID = "example.extension.recover" + request.Prescription, err = protocol.NewPrescription(prescriptionSnapshot, catalog.Transition{ID: request.TransitionID}) + if err != nil { + t.Fatal(err) + } if err := request.Validate(time.Now()); err != nil { t.Fatalf("surface schema rejected a recovery ID before compiled-registry validation: %v", err) } diff --git a/boatstack/program_effects.go b/boatstack/program_effects.go index 1422760..f53c925 100644 --- a/boatstack/program_effects.go +++ b/boatstack/program_effects.go @@ -15,8 +15,10 @@ import ( ) type programEffectDriver struct { - base ports.EffectDriver - program control.ControlProgram + base ports.EffectDriver + program control.ControlProgram + resolver ports.InvocationResolver + clock ports.Clock } func (d programEffectDriver) Prepare(ctx context.Context, admission protocol.Admission, transition catalog.Transition) (ports.PreparedEffect, error) { @@ -37,11 +39,11 @@ func (d programEffectDriver) Prepare(ctx context.Context, admission protocol.Adm } request := control.ProgramRuntimeRequest{ ProtocolVersion: control.ProgramRuntimeProtocolVersion, ProgramID: flow.Identity.ID, ProgramVersion: flow.Identity.Version, - ProgramFingerprint: admission.ProgramFingerprint, CorrelationID: admission.Invocation.Correlation, + ProgramFingerprint: admission.ExpectedProgramFingerprint, CorrelationID: admission.Invocation.Correlation, RepositoryRoot: admission.Invocation.InvokingPath, TransitionID: transition.ID, Parameters: parameters, Settings: flow.Manifest.Settings, } if transition.Class == catalog.EventOwnedExternal { - return effects.NewExtensionExternalPrepared(func(executionContext context.Context) (ports.EffectResult, error) { + prepared, err := effects.NewExtensionExternalPrepared(func(executionContext context.Context) (ports.EffectResult, error) { request.Operation = control.ProgramExecuteExternalOperation response, invokeErr := flow.Runtime.InvokeProgram(executionContext, request) if invokeErr != nil { @@ -52,6 +54,10 @@ func (d programEffectDriver) Prepare(ctx context.Context, admission protocol.Adm } return decodeExtensionSettlement(flow.Identity.ID, response.ExternalResult) }) + if err != nil { + return nil, err + } + return effects.BindStateRevision(ctx, prepared, d.resolver, d.clock, admission, transition) } operation := control.ProgramPlanLocalEffectOperation if transition.Class == catalog.EventRecovery { @@ -68,7 +74,11 @@ func (d programEffectDriver) Prepare(ctx context.Context, admission protocol.Adm if err := validateProgramWrites(d.program, transition, flow.Identity.ID, response.Writes); err != nil { return nil, err } - return effects.NewFlowLocalPrepared(admission.Invocation.InvokingPath, flow.Identity.ID, response.Writes) + prepared, err := effects.NewFlowLocalPrepared(admission.Invocation.InvokingPath, flow.Identity.ID, response.Writes) + if err != nil { + return nil, err + } + return effects.BindStateRevision(ctx, prepared, d.resolver, d.clock, admission, transition) } extension, ok := d.program.ExtensionByID(transition.Origin.ID) if !ok || extension.Runtime == nil { @@ -80,11 +90,11 @@ func (d programEffectDriver) Prepare(ctx context.Context, admission protocol.Adm } baseRequest := control.ExtensionRequest{ ProtocolVersion: control.ExtensionProtocolVersion, ExtensionID: extension.Identity.ID, ExtensionVersion: extension.Identity.Version, - ProgramFingerprint: admission.ProgramFingerprint, CorrelationID: admission.Invocation.Correlation, + ProgramFingerprint: admission.ExpectedProgramFingerprint, CorrelationID: admission.Invocation.Correlation, RepositoryRoot: admission.Invocation.InvokingPath, TransitionID: transition.ID, Parameters: parameters, Settings: extension.Manifest.Settings, } if transition.Class == catalog.EventOwnedExternal { - return effects.NewExtensionExternalPrepared(func(executionContext context.Context) (ports.EffectResult, error) { + prepared, err := effects.NewExtensionExternalPrepared(func(executionContext context.Context) (ports.EffectResult, error) { request := baseRequest request.Operation = control.ExtensionExecuteExternalOperation response, invokeErr := extension.Runtime.Invoke(executionContext, request) @@ -96,6 +106,10 @@ func (d programEffectDriver) Prepare(ctx context.Context, admission protocol.Adm } return decodeExtensionSettlement(extension.Identity.ID, response.ExternalResult) }) + if err != nil { + return nil, err + } + return effects.BindStateRevision(ctx, prepared, d.resolver, d.clock, admission, transition) } operation := control.ExtensionPlanLocalEffectOperation if transition.Class == catalog.EventRecovery { @@ -112,7 +126,11 @@ func (d programEffectDriver) Prepare(ctx context.Context, admission protocol.Adm if err := validateProgramWrites(d.program, transition, extension.Identity.ID, response.Writes); err != nil { return nil, err } - return effects.NewExtensionLocalPrepared(admission.Invocation.InvokingPath, extension.Identity.ID, response.Writes) + prepared, err := effects.NewExtensionLocalPrepared(admission.Invocation.InvokingPath, extension.Identity.ID, response.Writes) + if err != nil { + return nil, err + } + return effects.BindStateRevision(ctx, prepared, d.resolver, d.clock, admission, transition) } func validateProgramWrites(program control.ControlProgram, transition catalog.Transition, owner string, writes []control.ResourceWrite) error { diff --git a/boatstack/references/failure-moves.md b/boatstack/references/failure-moves.md index 8bc8936..5058114 100644 --- a/boatstack/references/failure-moves.md +++ b/boatstack/references/failure-moves.md @@ -4,7 +4,7 @@ Use the failure class, not the latest symptom: | Failure class | V2 move | |---|---| -| stale snapshot or prescription | re-observe; request a new admission | +| stale snapshot or prescription | discard it; re-resolve; execute no effects | | ambiguous identity | preserve resources; supply exact invocation | | configuration drift | validate and request `configuration.mutate` | | runtime absent or wrong | install a verified candidate; request runtime update | diff --git a/docs/architecture/boatstack-v2-kernel.md b/docs/architecture/boatstack-v2-kernel.md index 4f2e30d..7826866 100644 --- a/docs/architecture/boatstack-v2-kernel.md +++ b/docs/architecture/boatstack-v2-kernel.md @@ -555,10 +555,12 @@ or recovery refusals therefore cannot first appear at apply. ## 9. Admission and authority model Knowledge, precondition evidence, authority, and proof of effect are four -separate objects. `Admission` binds the exact transition ID/version, snapshot -fingerprint, invocation identity, goal and plan lock, observation/configuration -fingerprints, source revision, branch/worktree, authority receipt, provider -preview, idempotency key, and expiry. +separate objects. A content-addressed `Prescription` binds the exact transition, +durable state revision, executable program fingerprint, and snapshot fingerprint. +`Admission` binds that prescription plus transition ID/version, invocation +identity, goal and plan lock, observation/configuration fingerprints, source +revision, branch/worktree, authority receipt, provider preview, idempotency key, +and expiry. `admission.Admit` re-observes or compares current controlling fingerprints before any writer runs. A stale prescription fails without mutation. Human approval, @@ -582,17 +584,17 @@ validation, path resolution, and safety classification are read-only. Local transitions follow one journaled protocol: -1. validate admission against the exact source snapshot; -2. acquire a partition-scoped lock keyed by repository/worktree/resources; -3. capture exact prior bytes and external preconditions; -4. stage all local writes; -5. verify staged representations; -6. install effects in declared order; -7. install the authoritative binding/state last; -8. re-observe independently; -9. verify the target predicate; -10. append the immutable receipt and commit journal; -11. release the lock. +1. validate the prescription against the observed source snapshot; +2. acquire the repository/worktree/resource lock; +3. re-observe and compare the exact state revision, program, and snapshot; +4. validate admission against that locked snapshot; +5. capture exact prior bytes and external preconditions; +6. stage all local writes; +7. verify staged representations; +8. install effects in declared order; +9. install durable state revision `N+1` last; +10. re-observe independently and verify the target predicate; +11. append the immutable receipt, commit the journal, and release the lock. Failure restores exact prior bytes where reversible. A mixed epoch is never an accepted snapshot. An irreversible or unknown external outcome produces a typed @@ -621,12 +623,13 @@ both effect completion and postcondition truth. Otherwise the engine enters the declared rollback, compensation, or recovery path and returns non-success. `TransitionReceipt` is immutable and content-addressed. It binds schema, flow -and sequence IDs, transition ID/version, admission and goal IDs, source and -target fingerprints, authority classes, idempotency key, timestamps/duration, -outcome, postcondition verifier, recovery/terminal classification, and a -privacy-safe failure class. The admission ID transitively binds invocation, -authority receipts, parameters, and expiry. A receipt never embeds arbitrary -output, source, prompts, or secrets. +and sequence IDs, transition ID/version, prescription and admission IDs, +executable program fingerprint, prior and resulting durable state revisions, +goal ID, source and target fingerprints, authority classes, idempotency key, +timestamps/duration, outcome, postcondition verifier, recovery/terminal +classification, and a privacy-safe failure class. The admission ID transitively +binds invocation, authority receipts, parameters, and expiry. A receipt never +embeds arbitrary output, source, prompts, or secrets. Receipts are the only accepted evidence that a managed transition occurred. Plan approvals, publication settlement, and terminal claims point to exact @@ -763,8 +766,9 @@ Receipts are the factual source. The facade exposes a passive JSONL reader, Telemetry is consumer-neutral and privacy-safe. Allowlisted fields are schema version, flow ID, sequence, timestamp, goal ID, -transition ID, source/target fingerprints, outcome, duration, recovery and -authority classifications, terminal status, and controlled failure class. +transition ID, program and prescription identity, prior/resulting state +revisions, source/target fingerprints, outcome, duration, recovery and authority +classifications, terminal status, and controlled failure class. Prompts, reasoning, source code, diffs, arbitrary command output, secrets, environment variables, and user documents are prohibited. diff --git a/docs/architecture/prescription-transactions.md b/docs/architecture/prescription-transactions.md new file mode 100644 index 0000000..370e4c8 --- /dev/null +++ b/docs/architecture/prescription-transactions.md @@ -0,0 +1,45 @@ +# Prescription transaction boundary + +Resolution and application form one compare-and-swap transaction over the +durable logical state and the immutable executable Control Program. + +```text +snapshot(state revision N, program fingerprint P) + -> resolve + -> prescription(transition, N, P, snapshot fingerprint) + -> repository-scoped lock + -> re-observe and compare the complete binding + -> effect and durable state commit N+1 + -> receipt(N, N+1, P, prescription, admission) +``` + +The prescription is content-addressed and carries no authority. Apply and +recover require its transition ID, durable state revision, program fingerprint, +snapshot fingerprint, and correlation unchanged. Authority remains separately +typed, scoped, and validated by admission. + +## Control law + +An effect may commit only when the current durable state revision, executable +program fingerprint, and admission-relevant snapshot exactly equal the values +observed during resolution. The comparison occurs again under the +repository-scoped kernel lock before journaling or effects. + +Every accepted logical transition, including recovery and extension-owned +effects, advances the durable revision exactly once from `N` to `N+1`. Read-only +resolution, refusal, failed preflight, rollback, and ordinary stale replay do +not advance it. Revision zero and overflow are invalid. + +A mismatch returns `STALE_PRESCRIPTION`, produces no managed effect or receipt, +and requires a new resolution. Explicit idempotent replay is valid only when +the caller supplies the original idempotency key and the current state proves +that the recorded transaction settled. + +## Interruption + +Local effects stage reversible resource mutations in the transaction journal; +the logical state mutation installs last. A failed local effect rolls back +without advancing the committed revision. Recovery of an interrupted journal is +itself a prescribed transition and commits a new revision. An external effect +whose settlement cannot be proven remains recovery-required and is never +blindly retried. diff --git a/docs/getting-started.md b/docs/getting-started.md index ecf3cd4..caa9806 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -31,11 +31,24 @@ the current, independently hashed `.boatstack/project.json`. ## Enter managed scope ```sh -boatstack apply --repo . --transition engagement.begin \ +boatstack next --repo . --transition engagement.begin \ --goal-id search-timeout --goal-kind verified-implementation \ - --delivery search-timeout --repository-authority + --delivery search-timeout --repository-authority --format json + +boatstack apply --repo . --transition engagement.begin --flow search-timeout \ + --goal-id search-timeout --goal-kind verified-implementation \ + --delivery search-timeout --repository-authority \ + --correlation --prescription-id \ + --expected-state-revision \ + --expected-program-fingerprint \ + --expected-snapshot-fingerprint ``` +Friendly transition commands resolve and consume one exact prescription in the +same invocation. Integrations that call raw `apply` or `recover` must forward +the prescription ID, state revision, program fingerprint, snapshot fingerprint, +and correlation returned by `next` without modification. + A saved plan alone never engages Boatstack. Use `status` or `next` at any time; both are read-only. diff --git a/release-notes/2026-08-12-prescription-transaction-boundary.md b/release-notes/2026-08-12-prescription-transaction-boundary.md new file mode 100644 index 0000000..a28374d --- /dev/null +++ b/release-notes/2026-08-12-prescription-transaction-boundary.md @@ -0,0 +1,3 @@ +### Bind prescriptions to exact state and program revisions + +Boatstack now rejects stale prescriptions before effects, serializes competing commits, and records the exact prior and resulting durable revisions in every successful transition receipt.