feat(boatstack): flow watch — bounded observe-compare loop over the frontier

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.

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 <slug>` (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 <path>`; 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.

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 <check|next|tasks|frontier|report>")
+		fmt.Fprintln(os.Stderr, "usage: boatstack-helper flow <check|next|tasks|frontier|watch|report>")
 		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

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()
+}