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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
119 changes: 104 additions & 15 deletions boatstack/cmd/boatstack-helper/flow_runtime.go
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,10 @@ import (

"github.com/operatorstack/boatstack/boatstack/controlprogram"
softwareflow "github.com/operatorstack/boatstack/boatstack/flow/softwaredelivery"
"github.com/operatorstack/boatstack/boatstack/internal/softwaredelivery/durable"
"github.com/operatorstack/boatstack/boatstack/internal/softwaredelivery/effects"
"github.com/operatorstack/boatstack/boatstack/internal/softwaredelivery/model"
"github.com/operatorstack/boatstack/boatstack/internal/softwaredelivery/plant"
"github.com/operatorstack/boatstack/boatstack/internal/softwaredelivery/protocol"
"github.com/operatorstack/boatstack/boatstack/internal/softwaredelivery/surfaces"
)
Expand Down Expand Up @@ -59,6 +62,7 @@ func bindFlowEntry(ctx context.Context, options commandOptions) (commandOptions,
if options.flowProgramFingerprint != "" && options.flowProgramFingerprint != compiled.Fingerprint {
return commandOptions{}, fmt.Errorf("FLOW_PROGRAM_DRIFT: run fingerprint does not match the current artifact")
}
options.flowProgramFingerprint = compiled.Fingerprint
objective, err := softwareflow.ObjectiveForEntry(ctx, compiled, resolver, options.entryID)
if err != nil {
return commandOptions{}, err
Expand All @@ -67,27 +71,35 @@ func bindFlowEntry(ctx context.Context, options commandOptions) (commandOptions,
if !ok {
return commandOptions{}, fmt.Errorf("FLOW_ENTRY_UNKNOWN: %s", options.entryID)
}
plan, deliveryID, err := resolveBoundPlan(repository, entry, options)
options, err = bindActiveFlowContext(ctx, repository, options, objective)
if err != nil {
return commandOptions{}, err
}
planRaw, err := os.ReadFile(plan)
if err != nil {
return commandOptions{}, fmt.Errorf("FLOW_INPUT_REQUIRED: read selected plan: %w", err)
}
planDigest := sha256.Sum256(planRaw)
planFingerprint := hex.EncodeToString(planDigest[:])
repositoryIdentity, err := flowRepositoryIdentity(repository)
plan, deliveryID, err := resolveBoundPlan(repository, entry, objective, options)
if err != nil {
return commandOptions{}, err
}
runID := flowRunID(repositoryIdentity, compiled.Fingerprint, options.entryID, deliveryID, planFingerprint)
if options.runID != "" && options.runID != runID {
return commandOptions{}, fmt.Errorf("FLOW_RUN_MISMATCH: run ID does not identify the selected plan and repository")
planFingerprint := ""
if plan != "" {
planRaw, readErr := os.ReadFile(plan)
if readErr != nil {
return commandOptions{}, fmt.Errorf("FLOW_INPUT_REQUIRED: read selected plan: %w", readErr)
}
planDigest := sha256.Sum256(planRaw)
planFingerprint = hex.EncodeToString(planDigest[:])
repositoryIdentity, identityErr := flowRepositoryIdentity(repository)
if identityErr != nil {
return commandOptions{}, identityErr
}
runID := flowRunID(repositoryIdentity, compiled.Fingerprint, options.entryID, deliveryID, planFingerprint)
if options.runID != "" && options.runID != runID {
return commandOptions{}, fmt.Errorf("FLOW_RUN_MISMATCH: run ID does not identify the selected plan and repository")
}
options.runID = runID
} else if options.runID == "" {
return commandOptions{}, fmt.Errorf("FLOW_ACTIVE_RUN_INVALID: active abandonment has no committed run identity")
}
options.repository = repository
options.flowProgramFingerprint = compiled.Fingerprint
options.runID = runID
if options.objectiveKind == "" {
options.objectiveKind = string(objective)
}
Expand Down Expand Up @@ -137,6 +149,80 @@ func bindFlowEntry(ctx context.Context, options commandOptions) (commandOptions,
return options, nil
}

func bindActiveFlowContext(ctx context.Context, repository string, options commandOptions, entryObjective model.ObjectiveKind) (commandOptions, error) {
if options.runID != "" && entryObjective != model.ObjectiveAbandoned {
return options, nil
}
resolver, err := plant.NewResolver("")
if err != nil {
return commandOptions{}, err
}
host := options.host
if host == "" {
host = "cli"
}
invocation, err := resolver.ResolveInvocation(ctx, repository, host, "flow-entry-resume")
if err != nil {
common, commonErr := flowRepositoryIdentity(repository)
if commonErr == nil {
if _, stateErr := os.Stat(filepath.Join(common, "boatstack", "v2")); os.IsNotExist(stateErr) {
return options, nil
}
}
if _, stateErr := os.Stat(filepath.Join(repository, ".git", "boatstack")); stateErr != nil {
return options, nil
}
return commandOptions{}, err
}
layout, _, err := resolver.ResolveLayout(ctx, invocation)
if err != nil {
return commandOptions{}, err
}
raw, err := os.ReadFile(layout.StatePath)
if os.IsNotExist(err) {
return options, nil
}
if err != nil {
return commandOptions{}, fmt.Errorf("FLOW_ACTIVE_RUN_INVALID: read durable state: %w", err)
}
state, err := durable.DecodeState(raw)
if err != nil {
return commandOptions{}, fmt.Errorf("FLOW_ACTIVE_RUN_INVALID: decode durable state: %w", err)
}
active, ok := state.ActiveObjective()
if !ok {
return options, nil
}
prefix := "objective-" + options.programID + "-" + options.entryID + "-"
receipt, found, findErr := effects.FindLatestCommittedFlowForObjective(layout, invocation, active, state.Revision)
if findErr != nil {
return commandOptions{}, fmt.Errorf("FLOW_ACTIVE_RUN_INVALID: inspect committed flow receipts: %w", findErr)
}
if !found || !strings.HasPrefix(receipt.FlowID, "run-") {
return commandOptions{}, fmt.Errorf("FLOW_ACTIVE_RUN_INVALID: active objective has no committed run identity")
}
if active.Kind == entryObjective && strings.HasPrefix(active.ID, prefix) {
options.runID, options.deliveryID = receipt.FlowID, active.DeliveryID
options.objectiveID, options.objectiveKind = active.ID, string(active.Kind)
options.activeFlowBound = true
return options, nil
}
if entryObjective == model.ObjectiveAbandoned {
repositoryIdentity, identityErr := flowRepositoryIdentity(repository)
if identityErr != nil {
return commandOptions{}, identityErr
}
expectedRunID := flowRunID(repositoryIdentity, options.flowProgramFingerprint, options.entryID, active.DeliveryID, "active-run:"+receipt.FlowID)
if options.runID != "" && options.runID != expectedRunID {
return commandOptions{}, fmt.Errorf("FLOW_RUN_MISMATCH: run ID does not identify the active delivery")
}
options.runID = expectedRunID
options.deliveryID, options.activeFlowBound = active.DeliveryID, true
return options, nil
}
return commandOptions{}, fmt.Errorf("FLOW_ACTIVE_RUN_CONFLICT: delivery %q is active under objective %q; abandon it before selecting another inbox plan", active.DeliveryID, active.ID)
}

func validateResolvedParameter(parameters protocol.Parameters, name, expected string) error {
if actual, exists := parameters.Get(name); exists && actual != expected {
return fmt.Errorf("FLOW_INPUT_MISMATCH: parameter %s conflicts with the entry-resolved value", name)
Expand Down Expand Up @@ -186,8 +272,11 @@ func bindRPCFlowEntry(ctx context.Context, request surfaces.Request) (surfaces.R
return request, nil
}

func resolveBoundPlan(repository string, entry controlprogram.Entry, options commandOptions) (string, string, error) {
if options.runID == "" {
func resolveBoundPlan(repository string, entry controlprogram.Entry, entryObjective model.ObjectiveKind, options commandOptions) (string, string, error) {
if options.activeFlowBound && entryObjective == model.ObjectiveAbandoned {
return "", options.deliveryID, nil
}
if options.runID == "" && options.deliveryID == "" {
return resolvePlanInput(repository, entry)
}
if !flowSegment.MatchString(options.deliveryID) {
Expand Down
13 changes: 13 additions & 0 deletions boatstack/cmd/boatstack-helper/flow_runtime_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -820,6 +820,19 @@ func TestFlowEntryBindsStableRunAndResumesManagedPlan(t *testing.T) {
}
}

func TestRepositoryNamedAbandonmentEntryUsesCompiledObjective(t *testing.T) {
entry := controlprogram.Entry{ID: "cancel", Target: "safely-abandoned"}
plan, delivery, err := resolveBoundPlan(t.TempDir(), entry, model.ObjectiveAbandoned, commandOptions{
entryID: "cancel", activeFlowBound: true, deliveryID: "delivery-one",
})
if err != nil {
t.Fatal(err)
}
if plan != "" || delivery != "delivery-one" {
t.Fatalf("repository-named abandonment resolved plan=%q delivery=%q", plan, delivery)
}
}

func TestFlowEntryRejectsSelectedPlanContentSubstitution(t *testing.T) {
// control-law: one-flow-run-binds-the-exact-selected-plan-bytes
repository := flowRepository(t)
Expand Down
1 change: 1 addition & 0 deletions boatstack/cmd/boatstack-helper/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ type commandOptions struct {
deliveryID string
programID string
flowProgramFingerprint string
activeFlowBound bool
entryID string
runID string
transitionID string
Expand Down
6 changes: 6 additions & 0 deletions boatstack/flow/softwaredelivery/definition.go
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,12 @@ func (d Definition) RuntimeManifest(ctx context.Context) (delivery.ProgramRuntim
if len(transition.ObjectiveKinds) == 0 {
return delivery.ProgramRuntimeManifest{}, fmt.Errorf("transition %q supports none of the declared entry objectives", declaration.ID)
}
if transition.ID == "plan.abandon" && objectives[model.ObjectiveAbandoned] {
// A repository Flow that explicitly exposes a safely-abandoned entry
// makes abandonment progress for that objective only. Human authority
// remains mandatory and other objectives cannot select this transition.
transition.SelectionClass = delivery.SelectionProgramProgress
}
sort.Slice(transition.ObjectiveKinds, func(i, j int) bool { return transition.ObjectiveKinds[i] < transition.ObjectiveKinds[j] })
selected = append(selected, transition)
}
Expand Down
54 changes: 54 additions & 0 deletions boatstack/flow/softwaredelivery/definition_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -151,6 +151,60 @@ func TestRepositoryTransitionCannotWidenTrustedObjectiveKinds(t *testing.T) {
}
}

func TestAbandonmentEntryMakesTrustedAbandonmentObjectiveProgress(t *testing.T) {
truth := true
resolver, err := softwareflow.NewResolver(context.Background())
if err != nil {
t.Fatal(err)
}
document := controlprogram.Document{
SchemaVersion: controlprogram.SchemaVersion,
Program: controlprogram.Program{ID: "product-delivery", Version: "1"},
Facets: []controlprogram.Facet{
{ID: "publication", Kind: "string"}, {ID: "verification", Kind: "string"},
{ID: "configuration", Kind: "string"}, {ID: "runtime", Kind: "string"},
{ID: "delivery", Kind: "string"}, {ID: "workspace", Kind: "string"},
},
Operators: []controlprogram.Operator{
{ID: "publication.observe", Binding: &controlprogram.OperatorBinding{Reference: "software-delivery/publication.observe", Version: "1"}},
{ID: "plan.abandon", Binding: &controlprogram.OperatorBinding{Reference: "software-delivery/plan.abandon", Version: "1"}},
},
Transitions: []controlprogram.Transition{
{ID: "publication.observe", Operator: "publication.observe", Guard: controlprogram.Predicate{True: &truth}, Target: controlprogram.Predicate{True: &truth}, Priority: 77},
{ID: "plan.abandon", Operator: "plan.abandon", Guard: controlprogram.Predicate{True: &truth}, Target: controlprogram.Predicate{True: &truth}, Priority: 31},
},
Targets: []controlprogram.Target{
{ID: "published-pr", Predicate: controlprogram.Predicate{All: []controlprogram.Predicate{fact("verification", "current"), fact("configuration", "verified"), fact("runtime", "verified"), fact("publication", "open")}}},
{ID: "safely-abandoned", Predicate: controlprogram.Predicate{All: []controlprogram.Predicate{fact("delivery", "discarded"), {Fact: &controlprogram.FactPredicate{Facet: "workspace", Statuses: []string{"known"}, Values: []string{"abandoned", "absent"}}}}}},
},
Entries: []controlprogram.Entry{{ID: "run", Target: "published-pr"}, {ID: "abandon", Target: "safely-abandoned"}},
}
compiled, err := controlprogram.Compile(document, resolver)
if err != nil {
t.Fatal(err)
}
definition, err := softwareflow.NewDefinition(compiled, resolver)
if err != nil {
t.Fatal(err)
}
manifest, err := definition.RuntimeManifest(context.Background())
if err != nil {
t.Fatal(err)
}
for _, transition := range manifest.Transitions {
if transition.ID == "plan.abandon" {
if transition.SelectionClass != delivery.SelectionProgramProgress || len(transition.ObjectiveKinds) != 1 || transition.ObjectiveKinds[0] != delivery.ObjectiveAbandoned {
t.Fatalf("abandonment transition = %#v", transition)
}
if transition.Priority != 31 {
t.Fatalf("priority = %d, want 31", transition.Priority)
}
return
}
}
t.Fatal("trusted plan.abandon transition was not selected")
}

func TestCompiledBindingDriftFailsClosed(t *testing.T) {
truth := true
compiled, resolver := compiledFlow(t, controlprogram.Predicate{True: &truth})
Expand Down
25 changes: 24 additions & 1 deletion boatstack/flow/softwaredelivery/skills.go
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,19 @@ func renderSkill(compiled controlprogram.Compiled, entry controlprogram.Entry, s
description = "Run repository Flow entry " + entry.ID + " to target " + entry.Target + "."
}
description += " Use only when the user explicitly selects this repository Flow entry."
supersession := ""
if entry.Target == "published-pr" {
abandonmentSkill, ok := targetEntrySkill(compiled.Document.Program.ID, compiled.Document.Entries, "safely-abandoned")
if ok {
supersession = fmt.Sprintf(`
If the user requests different work, never retarget this run. When no objective
binding receipt exists, stop this unbound attempt and allow the inbox plan to be
replaced. Once the objective is bound, require explicit use of $%s for
the same delivery and wait for its abandonment receipt before selecting a new
plan and starting a new run.
`, abandonmentSkill)
}
}
return []byte(fmt.Sprintf(`---
name: %s
description: %q
Expand All @@ -67,11 +80,21 @@ Apply only the exact immediately preceding prescription and its declared
parameters. A question suspends this run: ask the user, submit only the typed
answer evidence, and resume the same run ID. Nothing continues in the
background while input is missing. Never synthesize authority.
%s

Stop only when Boatstack reports the marked target, a typed blocker, refusal,
unresolved recovery, or missing authority. This entry grants no merge or deploy
authority.
`, slug, description, title(slug), compiled.Document.Program.ID, entry.ID, entry.Target, compiled.Document.Program.ID, entry.ID, host))
`, slug, description, title(slug), compiled.Document.Program.ID, entry.ID, entry.Target, compiled.Document.Program.ID, entry.ID, host, supersession))
}

func targetEntrySkill(programID string, entries []controlprogram.Entry, target string) (string, bool) {
for _, entry := range entries {
if entry.Target == target {
return flowSkillSlug(programID, entry.ID), true
}
}
return "", false
}

func title(value string) string {
Expand Down
26 changes: 26 additions & 0 deletions boatstack/flow/softwaredelivery/skills_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,32 @@ func TestGeneratedSkillDescriptionIsQuotedYAML(t *testing.T) {
}
}

func TestGeneratedRunSkillRequiresExplicitAbandonmentBeforeReplacement(t *testing.T) {
compiled := controlprogram.Compiled{Document: controlprogram.Document{
Program: controlprogram.Program{ID: "product-delivery"},
Entries: []controlprogram.Entry{
{ID: "run", Target: "published-pr"},
{ID: "cancel", Target: "safely-abandoned"},
},
}}
files, err := softwareflow.GenerateSkills(compiled, []string{"codex", "claude"})
if err != nil {
t.Fatal(err)
}
if len(files) != 6 {
t.Fatalf("generated file count = %d, want 6", len(files))
}
run := string(files[".agents/skills/product-delivery-run/SKILL.md"])
for _, contract := range []string{"never retarget this run", "$product-delivery-cancel", "abandonment receipt", "starting a new run"} {
if !strings.Contains(run, contract) {
t.Fatalf("generated run skill lacks %q", contract)
}
}
if _, ok := files[".agents/skills/product-delivery-cancel/SKILL.md"]; !ok {
t.Fatal("abandonment entry skill was not generated")
}
}

func TestGeneratedSkillsRejectKernelMaintenanceIdentity(t *testing.T) {
compiled := controlprogram.Compiled{Document: controlprogram.Document{
Program: controlprogram.Program{ID: "boatstack"},
Expand Down
10 changes: 7 additions & 3 deletions boatstack/flow/standard/supervisor_parity_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -114,10 +114,14 @@ func TestExplicitPostTerminalCleanupRemainsAdmissible(t *testing.T) {
func TestTerminalEvidenceForOldObjectiveDoesNotTerminateNewObjective(t *testing.T) {
// control-law: terminal-evidence-is-bound-to-exact-objective-not-local-phase
s := New(testprogram.StandardRegistry(), testObjectiveContracts())
newObjective := model.Objective{ID: "next-objective", Kind: model.ObjectiveOpenPR, DeliveryID: "delivery"}
decision := s.Resolve(snapshotFor(t, model.PhaseTerminal, model.TerminalEstablished), newObjective, catalog.AuthoritySet{catalog.AuthorityHuman: true}, "objective.bind")
snapshot := snapshotFor(t, model.PhaseTerminal, model.TerminalEstablished)
snapshot.Workspace = model.Known(model.WorkspacePublished, snapshot.Workspace.Evidence[0])
snapshot.Publication = model.Known(model.PublicationOpen, snapshot.Publication.Evidence[0])
snapshot = recanonicalize(t, snapshot)
newObjective := model.Objective{ID: "next-objective", Kind: model.ObjectiveOpenPR, DeliveryID: "next-delivery"}
decision := s.Resolve(snapshot, newObjective, catalog.AuthoritySet{catalog.AuthorityHuman: true}, "")
if decision.Kind != DecisionPrescribed || decision.Transition == nil || decision.Transition.ID != "objective.bind" {
t.Fatalf("decision=%#v, want exact new-objective configuration", decision)
t.Fatalf("untargeted terminal replacement decision=%#v, want exact new-objective configuration", decision)
}
}

Expand Down
4 changes: 4 additions & 0 deletions boatstack/internal/softwaredelivery/durable/state.go
Original file line number Diff line number Diff line change
Expand Up @@ -163,6 +163,10 @@ func (s State) ConfigurationPolicy() model.ConfigurationPolicy {
}.Canonical()
}

func (s State) ActiveObjective() (model.Objective, bool) {
return s.Objective, s.Objective.ID != "" && s.Terminal == model.TerminalNonterminal
Comment on lines +166 to +167

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Preserve abandonment run identity after terminal commit

Invariant: a committed Flow transition must remain replayable under its original run identity after reaching its marked terminal. Abandonment uses a synthetic run ID derived from active-run:<original-run>, but after plan.abandon commits, ActiveObjective returns false. A retry after a lost response therefore resolves the plan normally and derives a different run ID from its file digest, failing with FLOW_RUN_MISMATCH before idempotency lookup or marked-state observation. The durable abandonment succeeds, but the client cannot recover its receipt through the Flow path. Add an end-to-end test that discards the first successful abandonment response and retries the exact request, expecting the committed receipt or marked result.

Confidence: 0.98

}

func EncodeState(state State) ([]byte, error) {
state = state.Canonical()
if err := state.Validate(); err != nil {
Expand Down
Loading
Loading