diff --git a/AUDIT.md b/AUDIT.md index 3b02c21..815145a 100644 --- a/AUDIT.md +++ b/AUDIT.md @@ -824,6 +824,22 @@ which emits the same events live. it requires intercepting a response — an opencode plugin (`~/.config/opencode/plugin/`) is the only clean hook. +### S4 — closed, 2026-07-27 + +`delivery.Fanout.Run` no longer `return`s on the first sender error — a +single ntfy hiccup used to permanently kill the notification goroutine for +the rest of the process (`main.go` only logged the `Run` error, it never +restarted the goroutine). Failed sends now go through an `OnError` hook +(default `log.Printf`), and the loop keeps going to the next sender/event. +Cursor persistence was also added: `SaveCursor` is called every time the +cursor advances, and `main.go` wires it to a `delivery-cursor` file next to +`ORCHESTRA_DATA`, loaded on startup — a restart resumes from the last +delivered event instead of re-notifying the entire log from seq 0. Covered +by `TestFanoutContinuesAfterSendError` (`internal/delivery/delivery_test.go` +— previously the package had zero tests): a failing sender and a healthy +sender both receive the event, the cursor still advances, and `Run` only +exits on context cancellation, never on the send error. + ### Design consequences (not yet implemented) 1. **Percentages are a level, not a delta.** diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index 912ee75..e666917 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -852,8 +852,24 @@ func main() { senders = append(senders, delivery.Ntfy{Topic: topic, Token: os.Getenv("ORCHESTRA_NTFY_TOKEN"), URL: os.Getenv("ORCHESTRA_NTFY_URL")}) } if len(senders) > 0 { + cursorPath := filepath.Join(dir, "delivery-cursor") + var startCursor uint64 + if b, err := os.ReadFile(cursorPath); err == nil { + if v, err := strconv.ParseUint(strings.TrimSpace(string(b)), 10, 64); err == nil { + startCursor = v + } + } go func() { - err := (&delivery.Fanout{Senders: senders}).Run(context.Background(), s.Events) + fanout := &delivery.Fanout{ + Senders: senders, + Cursor: startCursor, + SaveCursor: func(cursor uint64) { + if err := os.WriteFile(cursorPath, []byte(strconv.FormatUint(cursor, 10)), 0o644); err != nil { + log.Printf("delivery cursor persist: %v", err) + } + }, + } + err := fanout.Run(context.Background(), s.Events) if err != nil { log.Printf("delivery fanout: %v", err) } diff --git a/internal/delivery/delivery.go b/internal/delivery/delivery.go index 05afce5..6d17c6d 100644 --- a/internal/delivery/delivery.go +++ b/internal/delivery/delivery.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "fmt" + "log" "net/http" "orchestra/internal/domain" "time" @@ -101,23 +102,39 @@ type Fanout struct { Senders []Sender Cursor uint64 Retry time.Duration + // OnError is called for each failed send instead of aborting the fanout + // goroutine (S4, AUDIT.md: a single ntfy hiccup used to `return` and + // permanently kill notifications for the rest of the process). Defaults + // to log.Printf. + OnError func(err error) + // SaveCursor, if set, is called after the cursor advances past a + // processed event so a restart can resume from here instead of + // re-notifying the entire log from seq 0 (S4). + SaveCursor func(cursor uint64) } func (f *Fanout) Run(ctx context.Context, events func(uint64) []domain.Event) error { if f.Retry <= 0 { f.Retry = 5 * time.Second } + onError := f.OnError + if onError == nil { + onError = func(err error) { log.Printf("delivery: %v", err) } + } for { for _, e := range events(f.Cursor) { if msg, ok := Message(e); ok { for _, s := range f.Senders { if err := s.Send(ctx, msg); err != nil { - return err + onError(fmt.Errorf("send %s to %T: %w", msg, s, err)) } } } if e.Seq > f.Cursor { f.Cursor = e.Seq + if f.SaveCursor != nil { + f.SaveCursor(f.Cursor) + } } } select { diff --git a/internal/delivery/delivery_test.go b/internal/delivery/delivery_test.go new file mode 100644 index 0000000..4dd356e --- /dev/null +++ b/internal/delivery/delivery_test.go @@ -0,0 +1,61 @@ +package delivery_test + +import ( + "context" + "errors" + "orchestra/internal/delivery" + "orchestra/internal/domain" + "testing" + "time" +) + +type fakeSender struct { + fail bool + sent []string +} + +func (f *fakeSender) Send(_ context.Context, msg string) error { + if f.fail { + return errors.New("boom") + } + f.sent = append(f.sent, msg) + return nil +} + +// TestFanoutContinuesAfterSendError guards S4: a single sender error must +// not kill the fanout goroutine or stall the cursor for the other senders. +func TestFanoutContinuesAfterSendError(t *testing.T) { + failing := &fakeSender{fail: true} + ok := &fakeSender{} + var gotErrs int + var savedCursor uint64 + f := &delivery.Fanout{ + Senders: []delivery.Sender{failing, ok}, + Retry: time.Millisecond, + OnError: func(error) { gotErrs++ }, + SaveCursor: func(c uint64) { savedCursor = c }, + } + events := []domain.Event{{Seq: 1, Type: "TaskCompleted", TaskID: "t1"}} + served := false + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + err := f.Run(ctx, func(cursor uint64) []domain.Event { + if served { + return nil + } + served = true + return events + }) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Run returned %v, want deadline exceeded (goroutine must survive send errors)", err) + } + if gotErrs == 0 { + t.Fatal("expected OnError to be called for the failing sender") + } + if len(ok.sent) != 1 { + t.Fatalf("healthy sender got %d messages, want 1", len(ok.sent)) + } + if savedCursor != 1 { + t.Fatalf("cursor=%d, want 1 (must advance despite the other sender's error)", savedCursor) + } +} diff --git a/progress.md b/progress.md index 3778e75..7c28970 100644 --- a/progress.md +++ b/progress.md @@ -172,9 +172,44 @@ Fixed so far: in-pane prompt to write the file before exiting, which is a live-deployment fact, not something provable from source. -Not yet started: B7 (quota projection has no producer), -Codex/opencode completion producers, the turn-decision endpoint, S2–S4, -S7–S11. See `AUDIT.md` for the full plan. +- **B7 (post-hoc producer) + Phase 2 turn-decision endpoint** — landed + together, since both are new `QuotaReported`/turn-boundary paths off the + same completion/turn events. `POST /v1/harness/complete` now appends a + `QuotaReported` event (`harness_id` from the closing lease, `consumed` + from the same `usage.Numerator()` used for the receipt), so the router's + 5h/weekly availability filter and the brief's `quota_consumed` stop + evaluating against a permanent zero. New `Coordinator.TurnDecision` + (`internal/orchestrator/orchestrator.go`) mirrors `rotate()`'s per-task + logic (occupancy → turn-boundary → handoff-file → release) but runs + synchronously once per turn instead of waiting for `Monitor`'s ticker, + returning one of `continue`/`prepare_handoff`/`rotate_now`/`refuse` via the + new `POST /v1/harness/turn`. The Claude Stop hook + (`deploy/hooks/orchestra-stop.sh`) now calls this endpoint on every + ordinary turn boundary (report marker absent) instead of no-op'ing, and + exits 2 on `refuse` to stop the harness from finishing an unsafe turn. + Covered by `TestTurnDecision` (`internal/orchestrator/rotation_test.go`): + continue-below-threshold, refuse-when-not-at-boundary, and + rotate_now-releases-and-emits-a-valid-TaskReleased cases. + **Not done:** live per-harness *push* producers (Claude statusline, + Codex rollout tail) that would give B7 a second, continuous producer + independent of task completion — recorded as a design investigation in + AUDIT.md ("Real harness quota sources") but not implemented; Codex/ + opencode's own equivalents of the Claude Stop hook (whether their + turn-boundary mechanism actually calls `/v1/harness/turn`) also remain + unbuilt, same caveat as Phase 2 item 4 already named for `/complete`. + +- **S4** — `delivery.Fanout.Run` used to `return` on the first sender error, + permanently killing the notification goroutine (a single ntfy hiccup meant + no notifications for the rest of the process's lifetime, since nothing + restarts it). Failed sends now go through an `OnError` hook instead of + aborting the loop. Cursor is also persisted now (`SaveCursor` → a + `delivery-cursor` file next to `ORCHESTRA_DATA`, loaded on startup), so a + restart resumes from the last delivered event instead of re-notifying the + entire log from seq 0. `internal/delivery` previously had zero tests; + added `TestFanoutContinuesAfterSendError`. + +Not yet started: Codex/opencode completion producers, S2–S3, S7–S11. See +`AUDIT.md` for the full plan. **Phase 0 done (2026-07-27):** this box has live TCP reachability to the real herdr instance at `192.168.1.105:9245` — verified by hand (raw JSON-RPC