diff --git a/labs/12-product-engineering-loop/boatstack-distribution/release-notes/2026-07-28-flow-watch-loop.md b/labs/12-product-engineering-loop/boatstack-distribution/release-notes/2026-07-28-flow-watch-loop.md new file mode 100644 index 000000000..90a73bb7c --- /dev/null +++ b/labs/12-product-engineering-loop/boatstack-distribution/release-notes/2026-07-28-flow-watch-loop.md @@ -0,0 +1,5 @@ +### You can now wait for a pull request to move without polling it yourself + +`flow watch` observes your delivery frontier on an interval and exits the moment something changes: checks finish or fail, a review lands, a merge happens. It also exits immediately when nothing can move, and with a distinct exit code when its timeout passes with no change, so a script or an agent loop can tell "something happened" from "still waiting". Defaults are a 30-second interval and a 30-minute timeout, both adjustable. + +The watch only observes: it performs no writes and never runs an operation on your behalf. When it exits, run `next-status` and continue from the fresh state. Before this, waiting on CI meant either re-running status by hand or asking your agent to poll GitHub in prose. diff --git a/labs/12-product-engineering-loop/product-engineering-loop/SKILL.md b/labs/12-product-engineering-loop/product-engineering-loop/SKILL.md index 87bbe5b1b..2a66ff302 100644 --- a/labs/12-product-engineering-loop/product-engineering-loop/SKILL.md +++ b/labs/12-product-engineering-loop/product-engineering-loop/SKILL.md @@ -32,6 +32,8 @@ For the full state machine, read [workflow.md](references/workflow.md). For arti Run the project-local helper's read-only `next-status --repo . --json` inspection. Repository artifacts, managed delivery state, gate receipts, and the recorded PR identity are evidence; conversation, terminal, worktree, and process observations are context only. Never run the returned operation automatically. `NOT_STARTED` points to `auto-plan` (run it with the plan path via `--plan`); `PUBLISHED` means a PR exists but is not a verified merge; only `FEATURE_COMPLETE` requires no action. If state is ambiguous, stale, or invalid, name the blocker instead of choosing by recency or clearing artifacts. When an `AMBIGUOUS` block names only past deliveries the user no longer cares about, name the ignorable delivery slug(s) and offer to exclude them from ambiguity resolution; only after explicit user confirmation, add each slug with `.product-loop/bin/boatstack-helper ignore-delivery --repo . --feature ` (a bounded, provenance-safe write to `workflow.ignored_deliveries` — never hand-edit config or delivery state). Any new, unlisted ambiguous delivery still pauses the workflow. +To see every feature at once, run the read-only `.product-loop/bin/boatstack-helper flow frontier --repo .`. It lists each delivery, its observed position, and who owes the next step. To wait for a published PR to move (checks finish, a review lands, a merge happens), run the read-only `.product-loop/bin/boatstack-helper flow watch --repo .`. The watch observes on an interval and exits when the frontier changes, when nothing can move, or at its timeout. It never acts on what it sees. When it exits, run `next-status` again and continue from the fresh state. + ## Run through ship For `$boatstack run`, `/boatstack-run`, or natural language such as “run Boatstack through ship,” first run the read-only `next-status --repo . --json` and `operation-status --repo . --json`. Wait for an executing operation and reconcile unknown completion before retrying. When the host supplies the plan path, enter `auto-plan` with `--plan `; when no plan path is supplied, stop and ask the user for the plan to build. Return **Feature complete** only for a verified completed feature, and stop on unverified, ambiguous, stale, or invalid state. Before the first delivery-stage operation (`build`, `repair`, `test-gate`, `review-gate`, or `ship-gate`), run `run-preflight --repo . --json`. Planning and approval do not require a remote fetch. The preflight fetches `origin` and verifies the current named branch contains the fetched delivery base and is not behind or diverged from its upstream. A failed fetch, missing remote/base, stale base, upstream drift, or constrained branch mismatch blocks before delivery mutation. Never repair freshness by merging, rebasing, switching or creating a constrained delivery branch, discarding changes, force-pushing, or broadening permissions. diff --git a/labs/12-product-engineering-loop/product-engineering-loop/cmd/boatstack-helper/flow.go b/labs/12-product-engineering-loop/product-engineering-loop/cmd/boatstack-helper/flow.go index 7b5874cbf..f0eb6733c 100644 --- a/labs/12-product-engineering-loop/product-engineering-loop/cmd/boatstack-helper/flow.go +++ b/labs/12-product-engineering-loop/product-engineering-loop/cmd/boatstack-helper/flow.go @@ -4,6 +4,7 @@ import ( "flag" "fmt" "os" + "time" boatstack "github.com/operatorstack/boatstack/boatstack" ) @@ -14,7 +15,7 @@ import ( // gate, authority, or exit code. func flowCommand(arguments []string) int { if len(arguments) == 0 { - fmt.Fprintln(os.Stderr, "usage: boatstack-helper flow ") + fmt.Fprintln(os.Stderr, "usage: boatstack-helper flow ") return 2 } switch arguments[0] { @@ -26,6 +27,8 @@ func flowCommand(arguments []string) int { return flowTasksCommand(arguments[1:]) case "frontier": return flowFrontierCommand(arguments[1:]) + case "watch": + return flowWatchCommand(arguments[1:]) case "report": return flowReportCommand(arguments[1:]) default: @@ -177,6 +180,41 @@ func flowFrontierCommand(arguments []string) int { return 0 } +// flowWatchCommand runs the bounded observe-compare loop: re-observe the +// frontier on an interval, exit 0 the moment it changes (or when nothing can +// move), exit 1 when the timeout passes with no change. It observes and +// exits; it never acts on what it sees. +// control-law: watch-observes-and-exits-never-acts +func flowWatchCommand(arguments []string) int { + flags := flag.NewFlagSet("flow watch", flag.ContinueOnError) + repo := flags.String("repo", ".", "repository whose delivery frontier should be watched") + interval := flags.Duration("interval", 30*time.Second, "time between frontier observations") + timeout := flags.Duration("timeout", 30*time.Minute, "maximum time to wait for a frontier change") + jsonOutput := flags.Bool("json", false, "print the structured watch result") + if err := flags.Parse(arguments); err != nil { + return 2 + } + result, err := boatstack.WatchFrontier(boatstack.FlowWatchOptions{ + Repo: *repo, Interval: *interval, Timeout: *timeout, + }) + if err != nil { + return fail(err) + } + if *jsonOutput { + value, marshalErr := boatstack.MarshalJSON(result) + if marshalErr != nil { + return fail(marshalErr) + } + fmt.Print(string(value)) + } else { + fmt.Print(boatstack.FormatFlowWatch(result)) + } + if result.Outcome == boatstack.WatchOutcomeTimeout { + return 1 + } + return 0 +} + // flowTasksCommand renders the active delivery slice's sub-actions from the // compiled plan task DAG, in dependency order, with the one to start pointed at. // It is read-only and never fails on flow position — an unresolved slice or an diff --git a/labs/12-product-engineering-loop/product-engineering-loop/flow_watch.go b/labs/12-product-engineering-loop/product-engineering-loop/flow_watch.go new file mode 100644 index 000000000..df86bcafd --- /dev/null +++ b/labs/12-product-engineering-loop/product-engineering-loop/flow_watch.go @@ -0,0 +1,169 @@ +package boatstack + +import ( + "fmt" + "sort" + "strings" + "time" +) + +// `flow watch` is the bounded waiting primitive for the asynchronous world a +// published PR lives in (CI runs, reviews land, merges happen). Each tick it +// re-runs the same read-only frontier observation and compares a stable +// signature of every row; it EXITS on the first change, on an all-terminal +// frontier, or at the deadline — it never acts on what it sees. Boatstack +// stays a synchronous oracle: the loop here only decides when to ask the +// oracle again, and hands control back the moment the answer differs. No +// daemon, no writes, no transition execution path is reachable from it. +// control-law: watch-observes-and-exits-never-acts +const flowWatchSchemaVersion = 1 + +const ( + WatchOutcomeChanged = "changed" + WatchOutcomeTerminal = "terminal" + WatchOutcomeTimeout = "timeout" +) + +// Seams for tests: the watcher must be provable without real waiting. +var ( + flowWatchNow = time.Now + flowWatchSleep = time.Sleep +) + +type FlowWatchOptions struct { + Repo string + Interval time.Duration + Timeout time.Duration +} + +// FlowWatchResult reports why the watch loop returned and what it saw. Final +// always carries the last observed frontier so the caller re-orients without +// another resolution. +type FlowWatchResult struct { + SchemaVersion int `json:"schema_version"` + Outcome string `json:"outcome"` + Ticks int `json:"ticks"` + ChangedRows []string `json:"changed_rows,omitempty"` + Final FlowFrontier `json:"final"` +} + +const ( + defaultWatchInterval = 30 * time.Second + defaultWatchTimeout = 30 * time.Minute + // minimumWatchInterval keeps a mistyped interval from hammering GitHub. + minimumWatchInterval = 5 * time.Second +) + +// WatchFrontier runs the bounded observe-compare loop. It returns an error +// only for the faults ResolveFrontier itself refuses (unreadable store, +// invalid config); a failing gh observation degrades each row to an Unknown +// phase — a signature like any other — and the loop stays bounded. +func WatchFrontier(options FlowWatchOptions) (FlowWatchResult, error) { + interval := options.Interval + if interval <= 0 { + interval = defaultWatchInterval + } + if interval < minimumWatchInterval { + interval = minimumWatchInterval + } + timeout := options.Timeout + if timeout <= 0 { + timeout = defaultWatchTimeout + } + + result := FlowWatchResult{SchemaVersion: flowWatchSchemaVersion} + initial, err := ResolveFrontier(options.Repo) + if err != nil { + return result, err + } + result.Final = initial + if frontierAllTerminal(initial) { + result.Outcome = WatchOutcomeTerminal + return result, nil + } + baseline := frontierSignatures(initial) + deadline := flowWatchNow().Add(timeout) + for { + if !flowWatchNow().Before(deadline) { + result.Outcome = WatchOutcomeTimeout + return result, nil + } + flowWatchSleep(interval) + result.Ticks++ + current, err := ResolveFrontier(options.Repo) + if err != nil { + return result, err + } + result.Final = current + signatures := frontierSignatures(current) + if changed := signatureDiff(baseline, signatures); len(changed) > 0 { + result.Outcome = WatchOutcomeChanged + result.ChangedRows = changed + return result, nil + } + } +} + +// frontierAllTerminal reports whether nothing on the frontier can move: no +// rows at all, or every row terminal. Blocked and operator rows are NOT +// terminal — external state (a review, a merge, a fix landing elsewhere) can +// change them, which is exactly what a watcher waits for. +func frontierAllTerminal(frontier FlowFrontier) bool { + if !frontier.Initialized || len(frontier.Rows) == 0 { + return true + } + for _, row := range frontier.Rows { + if row.Actor != string(NextActorNone) { + return false + } + } + return true +} + +// frontierSignatures reduces each row to the stable facts a caller would act +// on: position, actor, lifecycle, and the failing-check set. Reasons and +// prescribed command text are deliberately excluded — wording changes are not +// frontier changes. +func frontierSignatures(frontier FlowFrontier) map[string]string { + signatures := map[string]string{} + for _, row := range frontier.Rows { + key := row.Feature + "/" + row.Slice + signatures[key] = strings.Join([]string{ + row.Stage, row.Lifecycle, row.PRPhase, row.Actor, row.NextOperation, + fmt.Sprintf("blocked=%t", row.Blocked), + strings.Join(row.PRFailingChecks, "|"), + }, "·") + } + return signatures +} + +func signatureDiff(before, after map[string]string) []string { + changed := []string{} + for key, value := range after { + if previous, ok := before[key]; !ok || previous != value { + changed = append(changed, key) + } + } + for key := range before { + if _, ok := after[key]; !ok { + changed = append(changed, key+" (gone)") + } + } + sort.Strings(changed) + return changed +} + +// FormatFlowWatch renders the watch outcome and the final frontier. +func FormatFlowWatch(result FlowWatchResult) string { + var b strings.Builder + switch result.Outcome { + case WatchOutcomeChanged: + fmt.Fprintf(&b, "Watch: frontier changed after %d tick(s): %s\n", result.Ticks, strings.Join(result.ChangedRows, ", ")) + case WatchOutcomeTerminal: + b.WriteString("Watch: nothing on the frontier can move; not waiting.\n") + case WatchOutcomeTimeout: + fmt.Fprintf(&b, "Watch: no frontier change within the timeout (%d tick(s)).\n", result.Ticks) + } + b.WriteString(FormatFlowFrontier(result.Final)) + return b.String() +} diff --git a/labs/12-product-engineering-loop/product-engineering-loop/flow_watch_conformance_test.go b/labs/12-product-engineering-loop/product-engineering-loop/flow_watch_conformance_test.go new file mode 100644 index 000000000..8d9b4d5ab --- /dev/null +++ b/labs/12-product-engineering-loop/product-engineering-loop/flow_watch_conformance_test.go @@ -0,0 +1,164 @@ +package boatstack + +// control-law: watch-observes-and-exits-never-acts +// +// `flow watch` is a bounded observe-compare loop over the read-only frontier: +// each tick re-observes, and the loop exits on the first signature change, on +// an all-terminal frontier, or at the deadline. It never executes a +// transition and never writes — across any number of ticks the delivery +// ledger stays byte-identical. Time is injected through seams so the law is +// provable without real waiting. +// +// Test classes: positive (an external phase change ends the wait and names +// the changed row), negative (no change → timeout outcome, frontier intact), +// bypass (zero writes across many ticks), failure-state (gh failing every +// tick degrades to Unknown rows and the loop still terminates at the +// deadline; an all-terminal frontier refuses to wait at all). + +import ( + "errors" + "os" + "sync/atomic" + "testing" + "time" +) + +// fakeWatchClock replaces the time seams: sleeping advances a virtual clock, +// so deadlines fire deterministically and instantly. +func fakeWatchClock(t *testing.T) *atomic.Int64 { + t.Helper() + var virtual atomic.Int64 + previousNow, previousSleep := flowWatchNow, flowWatchSleep + flowWatchNow = func() time.Time { return time.Unix(0, virtual.Load()) } + flowWatchSleep = func(d time.Duration) { virtual.Add(int64(d)) } + t.Cleanup(func() { flowWatchNow, flowWatchSleep = previousNow, previousSleep }) + return &virtual +} + +func watchRepoWithOpenPR(t *testing.T) string { + t.Helper() + repo := nextTestRepo(t) + writeNextDelivery(t, repo, "shipped", "PUBLISHED", 1) + updateRecoveryDelivery(t, repo, "shipped", "feat/phase", "https://example.invalid/pr/9", "") + return repo +} + +// Positive: the PR's checks finish between ticks; the watch exits with +// outcome "changed" and names the row that moved. +func TestWatchExitsWhenTheFrontierChanges(t *testing.T) { + fakeWatchClock(t) + repo := watchRepoWithOpenPR(t) + var calls atomic.Int64 + withRecoveryGh(t, func(_ string, args ...string) (string, error) { + if calls.Add(1) <= 1 { + return phaseObservationPayload("OPEN", "", "CLEAN", rollupCheckRunPending)(repo, args...) + } + return phaseObservationPayload("OPEN", "", "CLEAN", rollupCheckRunFail)(repo, args...) + }) + result, err := WatchFrontier(FlowWatchOptions{Repo: repo, Interval: time.Minute, Timeout: time.Hour}) + if err != nil { + t.Fatal(err) + } + if result.Outcome != WatchOutcomeChanged || result.Ticks != 1 { + t.Fatalf("unexpected watch result: %#v", result) + } + if len(result.ChangedRows) != 1 || result.ChangedRows[0] != "shipped/delivery" { + t.Fatalf("changed row not named: %#v", result.ChangedRows) + } + if result.Final.Rows[0].PRPhase != string(PRPhaseChecksFailing) { + t.Fatalf("final frontier does not carry the new observation: %#v", result.Final.Rows) + } +} + +// Negative: nothing changes; the watch times out with the frontier intact and +// the CLI-visible timeout outcome. +func TestWatchTimesOutWhenNothingChanges(t *testing.T) { + fakeWatchClock(t) + repo := watchRepoWithOpenPR(t) + withRecoveryGh(t, phaseObservationPayload("OPEN", "", "CLEAN", rollupCheckRunPending)) + result, err := WatchFrontier(FlowWatchOptions{Repo: repo, Interval: 10 * time.Minute, Timeout: time.Hour}) + if err != nil { + t.Fatal(err) + } + if result.Outcome != WatchOutcomeTimeout { + t.Fatalf("unexpected outcome: %#v", result) + } + if result.Ticks < 5 || result.Ticks > 7 { + t.Fatalf("unexpected tick count for 1h/10m: %d", result.Ticks) + } + if result.Final.Rows[0].PRPhase != string(PRPhaseChecksPending) { + t.Fatalf("frontier drifted without a change: %#v", result.Final.Rows) + } +} + +// Bypass: across many ticks — including a tick that observes a terminal +// MERGED lifecycle — the watch writes nothing. The change is reported, never +// recorded. +func TestWatchWritesNothingAcrossTicks(t *testing.T) { + fakeWatchClock(t) + repo := watchRepoWithOpenPR(t) + statePath, err := deliveryStatePath(repo, "shipped") + if err != nil { + t.Fatal(err) + } + before, err := os.ReadFile(statePath) + if err != nil { + t.Fatal(err) + } + var calls atomic.Int64 + withRecoveryGh(t, func(_ string, args ...string) (string, error) { + if calls.Add(1) <= 3 { + return phaseObservationPayload("OPEN", "", "CLEAN", rollupCheckRunPending)(repo, args...) + } + return phaseObservationPayload("MERGED", "", "", "")(repo, args...) + }) + result, err := WatchFrontier(FlowWatchOptions{Repo: repo, Interval: time.Minute, Timeout: time.Hour}) + if err != nil { + t.Fatal(err) + } + if result.Outcome != WatchOutcomeChanged { + t.Fatalf("unexpected outcome: %#v", result) + } + after, err := os.ReadFile(statePath) + if err != nil { + t.Fatal(err) + } + if string(before) != string(after) { + t.Fatal("watch mutated the delivery ledger") + } +} + +// Failure-state 1: gh fails on every tick — rows degrade to Unknown, no +// crash, and the loop still ends at the deadline. +func TestWatchStaysBoundedWhenObservationFails(t *testing.T) { + fakeWatchClock(t) + repo := watchRepoWithOpenPR(t) + withRecoveryGh(t, func(string, ...string) (string, error) { return "", errors.New("gh unavailable") }) + result, err := WatchFrontier(FlowWatchOptions{Repo: repo, Interval: 15 * time.Minute, Timeout: time.Hour}) + if err != nil { + t.Fatal(err) + } + if result.Outcome != WatchOutcomeTimeout { + t.Fatalf("unexpected outcome: %#v", result) + } + if result.Final.Rows[0].PRPhase != string(PRPhaseUnknown) { + t.Fatalf("degraded observation not fail-closed: %#v", result.Final.Rows) + } +} + +// Failure-state 2: when nothing on the frontier can move, the watch refuses +// to wait at all. +func TestWatchRefusesToWaitOnTerminalFrontier(t *testing.T) { + fakeWatchClock(t) + repo := nextTestRepo(t) + writeNextDelivery(t, repo, "done", "PUBLISHED", 1) + updateRecoveryDelivery(t, repo, "done", "feat/phase", "https://example.invalid/pr/9", "") + withRecoveryGh(t, phaseObservationPayload("MERGED", "", "", "")) + result, err := WatchFrontier(FlowWatchOptions{Repo: repo, Interval: time.Minute, Timeout: time.Hour}) + if err != nil { + t.Fatal(err) + } + if result.Outcome != WatchOutcomeTerminal || result.Ticks != 0 { + t.Fatalf("unexpected result on a terminal frontier: %#v", result) + } +} diff --git a/labs/12-product-engineering-loop/product-engineering-loop/references/workflow.md b/labs/12-product-engineering-loop/product-engineering-loop/references/workflow.md index bb636cd43..042d202a1 100644 --- a/labs/12-product-engineering-loop/product-engineering-loop/references/workflow.md +++ b/labs/12-product-engineering-loop/product-engineering-loop/references/workflow.md @@ -487,6 +487,10 @@ When `workspace.enabled` is set and an approved feature is still on the default When `workspace.enabled` is set, `boatstack-next` surfaces `workspace-cleanup` for a published feature whose managed worktree still exists locally. The `workspace-cleanup` operation checks the pull request's merge state (GitHub CLI, falling back to local ancestry) and reports it. When `workspace.cleanup_after` is `merge`, cleanup is offered only once the PR is confirmed merged; while it is still open, the workspace is kept and the human may keep waiting or override explicitly. Cleanup never removes a workspace with uncommitted or unmerged work without an explicit forced override, and it reclaims only the local worktree and branch — it never deletes a remote branch or merges anything. In `confirm` mode the human reclaims the workspace with the exact reply `c` (or keeps it with `k`); `auto` mode reclaims a merged workspace without a prompt; `off` disables cleanup. A fresh feature workspace is likewise cut from the up-to-date default branch when a new feature begins, so work never starts on a stale branch. +### `PR_OPEN -> WATCH` + +A published pull request changes asynchronously: checks finish, reviews land, merges happen. `flow watch` is the bounded waiting primitive for that interval. It re-observes the read-only frontier on an interval and exits when a row's position or owner changes, when nothing on the frontier can move, or when its timeout passes (distinct exit code). It performs no writes and executes no operation — observation and actuation stay separate, so waiting can never become acting. When the watch exits, resolve `next-status` again and continue from the fresh state. + ### `PR_OPEN -> RETRO` Record unexpected friction and outcomes. A retro may propose a loop move, but it may not mutate durable instructions automatically.