amadan

carlos / platform Public

plan: scheduled work implementation plan

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev f83fd7ef01822cd9fe189aa5be7801d000e6b682 parent a24f689
1 file changed, +3164 −0
  • docs/superpowers/plans/2026-08-23-scheduled-work.md +3164 −0
diff --git a/docs/superpowers/plans/2026-08-23-scheduled-work.md b/docs/superpowers/plans/2026-08-23-scheduled-work.md
new file mode 100644
index 000000000..00ac76038
--- /dev/null
+++ b/docs/superpowers/plans/2026-08-23-scheduled-work.md
@@ -0,0 +1,3164 @@
+# Scheduled Work Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Apps get recurring schedules (`carlos schedule set -every 6h`) and one-shot timers (registered over the control socket) that fire as an authenticated `POST` tick, waking a hibernating instance and holding it awake while the handler runs.
+
+**Architecture:** Declarations live in the bucket (`apps/<acct>/<app>/schedules.json`, console-written, serial-bumped); the box converges them into a `timers` table in its registry; a runner goroutine delivers due ticks through the activator; boxes report per-instance status back to the bucket so `carlos schedule ls` is honest with no box access. A new package `internal/schedule` holds spec parsing, converge, runner policy and the control-socket handler; `internal/host` wires it; `internal/console` and the root CLI get the doors and verb.
+
+**Tech Stack:** Go 1.26, stdlib only (`GOPROXY=off` — no new modules), modernc sqlite via `internal/registry`, `internal/release.Store` for the bucket.
+
+**Spec:** `docs/superpowers/specs/2026-08-23-scheduled-work-design.md`
+
+## Global Constraints
+
+- Build/test env for every command: `export GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off` (may need the sandbox disabled: "read-only file system" from the toolchain is the tell).
+- Gate before every commit: `go build ./... && go vet ./... && gofmt -l <files you touched>` clean; `go test ./<package> -count=1` for the packages you touched. The whole-repo `go test ./... -count=1` and `GOOS=darwin go build .` run in the final task.
+- No new dependencies. The cron parser is ours.
+- Wire shapes are additive; every bucket object carries `"schema": 1`.
+- Names: `^[a-z0-9][a-z0-9-]{0,31}$`. `every`: 1m–720h whole minutes. Cron: 5 fields, UTC. Path: absolute, no `..`, ≤ 256 bytes. ≤ 20 schedules per app. ≤ 1000 once-timers per host. Once `at` ≤ 400 days ahead.
+- Delivery timeout 30m; retry backoff `min(1m·2^(attempts-1), 15m)`, 5 attempts; runner tick 15s; ≤ 4 hosts in parallel, one timer at a time per host.
+- Timer statuses are strings a member reads verbatim: `"204"`, `"404 (app refused)"`, `"503: instance failed to wake"`, `"timeout after 30m"`, `"failed after 5 attempts: <last>"`.
+- Every commit message ends with `Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>`. Nothing merges to `main` except through the PR (CLAUDE.md).
+- Doc comments in this repo explain WHY, at the top of files and on exported names — follow the density of the neighbours you read.
+
+---
+
+## File map
+
+| File | Responsibility |
+|---|---|
+| `internal/schedule/spec.go` | `Spec` (every/cron/at), `ParseSpec`, `Next`, `every` alignment |
+| `internal/schedule/cron.go` | the 5-field UTC cron parser and matcher |
+| `internal/schedule/declaration.go` | `Declaration`/`Schedule` JSON, validation, bucket keys, CAS upsert/remove, `TouchRun` |
+| `internal/schedule/status.go` | `Status`/`TimerStatus` JSON, status key, `WriteStatus`, `ListStatus` |
+| `internal/schedule/converge.go` | bucket declaration → `timers` rows, run-signal, prune, reporter callback |
+| `internal/schedule/runner.go` | due query → deliver → outcome policy → mark/report/event |
+| `internal/schedule/control.go` | control-socket `PUT/DELETE/GET /schedule…` handler with bearer auth |
+| `internal/registry/timers.go` | the `timers` table and its methods |
+| `internal/edge/purge.go` | `ListenPurge` becomes `ListenControl(socket, host, cache, mount)` |
+| `internal/edge/edge.go` | export `UnixProxy` |
+| `internal/activator/backend_exec.go` | mint `CARLOS_ADMIN_TOKEN`; always-on control listener; mount schedule handler |
+| `internal/host/schedule_wiring.go` | converge leg, runner start, activator-backed deliverer, status reporter |
+| `internal/host/agent.go` | call the wiring from `fastConvergeLoop`/`adoptLoop`/startup |
+| `internal/console/apischedule.go` | the four doors |
+| `internal/console/authcli.go` | register the doors |
+| `internal/console/svcroutes_test.go` | door lists |
+| `apimode.go` | CLI client calls |
+| `schedule.go` (root) | `carlos schedule ls|set|rm|run` |
+| `main.go` | dispatch + usage line |
+
+Import direction (no cycles): `schedule` → `registry`, `release`. `activator` → `schedule` (control handler). `host` → `activator`, `edge`, `schedule`. `console` → `schedule`. root → `schedule`.
+
+---
+
+### Task 1: `internal/schedule` — specs, cron, declarations, status shapes
+
+**Files:**
+- Create: `internal/schedule/spec.go`, `internal/schedule/spec_test.go`
+- Create: `internal/schedule/cron.go`, `internal/schedule/cron_test.go`
+- Create: `internal/schedule/declaration.go`, `internal/schedule/declaration_test.go`
+- Create: `internal/schedule/status.go`, `internal/schedule/status_test.go`
+
+**Interfaces:**
+- Consumes: `release.Store` (`Get/Put/List/Delete`), `release.AppPrefix`, `release.ErrCASConflict`, `release.ValidAccount`, `release.ValidApp`.
+- Produces (later tasks build on these exact names):
+
+```go
+package schedule
+
+const (
+ KindSchedule = "schedule"
+ KindOnce = "once"
+ MaxSchedulesPerApp = 20
+ MaxOncePerHost = 1000
+ MinEvery = time.Minute
+ MaxEvery = 720 * time.Hour
+ MaxOnceAhead = 400 * 24 * time.Hour
+)
+
+type Spec struct {
+ Kind string // KindSchedule | KindOnce
+ Every time.Duration // > 0 for "every N"
+ Cron *CronExpr // non-nil for "cron …"
+ At time.Time // for "at <unix>"
+}
+func ParseSpec(s string) (Spec, error) // "every 6h" | "cron 0 8 * * *" | "at 1787004000"
+func (s Spec) String() string // the canonical form ParseSpec accepts
+func (s Spec) Next(now time.Time) time.Time // zero = never
+
+type CronExpr struct{ min, hour, dom, month, dow uint64 } // bitsets
+func ParseCron(expr string) (*CronExpr, error)
+func (c *CronExpr) Next(after time.Time) time.Time // strictly after `after`, UTC, minute granularity; zero if none within 366 days
+
+func ValidName(name string) bool
+func ValidPath(path string) error
+
+type Schedule struct {
+ Name string `json:"name"`
+ Every string `json:"every,omitempty"` // Go duration text as typed, normalised by Validate
+ Cron string `json:"cron,omitempty"`
+ Path string `json:"path"`
+}
+func (s Schedule) Spec() (Spec, error)
+func (s Schedule) Validate() error
+
+type Declaration struct {
+ Schema int `json:"schema"`
+ UpdatedAt string `json:"updated_at"`
+ Schedules []Schedule `json:"schedules"`
+}
+func (d Declaration) Find(name string) (Schedule, bool)
+func DeclarationKey(acct, app string) string // apps/<acct>/<app>/schedules.json
+func RunKey(acct, app string) string // apps/<acct>/<app>/schedule-run.json
+func StatusPrefix(acct, app string) string // apps/<acct>/<app>/schedule-status/
+func StatusKey(acct, app, host string) string
+
+func ReadDeclaration(ctx, s release.Store, acct, app string) (Declaration, string /*etag*/, error) // absent → empty Declaration, "", nil
+func UpsertSchedule(ctx, s release.Store, acct, app string, sc Schedule, now time.Time) (Declaration, error) // validates, CAS, one retry on conflict
+func RemoveSchedule(ctx, s release.Store, acct, app, name string, now time.Time) (Declaration, bool, error) // bool = was present
+func TouchRun(ctx, s release.Store, acct, app, name string, now time.Time) error // restart-gen semantics: CAS conflict = success
+
+type TimerStatus struct {
+ Name string `json:"name"`; Kind string `json:"kind"`; Spec string `json:"spec"`; Path string `json:"path"`
+ NextAt int64 `json:"next_at"`; LastAt int64 `json:"last_at"`; LastStatus string `json:"last_status"`
+ LastMs int64 `json:"last_ms"`; Attempts int `json:"attempts"`
+}
+type Status struct {
+ Schema int `json:"schema"`; Host string `json:"host"`; Box string `json:"box"`; ReportedAt string `json:"reported_at"`
+ Unsupported string `json:"unsupported"`; Timers []TimerStatus `json:"timers"`
+}
+func WriteStatus(ctx, s release.Store, acct, app string, st Status) error // unconditional Put (etag "" is create-only in DirStore/S3 — so read the etag first and Put with it; on ErrCASConflict re-read once and retry; never loop)
+func ListStatus(ctx, s release.Store, acct, app string) ([]Status, error) // sorted by Host; unparseable objects skipped with the error joined
+```
+
+- [ ] **Step 1: Write the cron tests** (`cron_test.go`)
+
+```go
+package schedule
+
+import (
+ "testing"
+ "time"
+)
+
+func at(s string) time.Time {
+ t, err := time.Parse(time.RFC3339, s)
+ if err != nil {
+ panic(err)
+ }
+ return t.UTC()
+}
+
+func TestParseCronAcceptsTheGrammar(t *testing.T) {
+ for _, ok := range []string{"* * * * *", "0 8 * * *", "*/15 * * * *", "0 */6 * * *", "0 9 1-5 * *", "30 7 * * 1,3,5", "0 0 1 1 *", "0 0 * * 7", "0-59/20 * * * *"} {
+ if _, err := ParseCron(ok); err != nil {
+ t.Errorf("ParseCron(%q) = %v, want ok", ok, err)
+ }
+ }
+ for _, bad := range []string{"", "* * * *", "* * * * * *", "60 * * * *", "* 24 * * *", "* * 0 * *", "* * 32 * *", "* * * 13 *", "* * * * 8", "@daily", "*/0 * * * *", "5-3 * * * *", "a * * * *", "0 8 * * MON"} {
+ if _, err := ParseCron(bad); err == nil {
+ t.Errorf("ParseCron(%q) = ok, want error", bad)
+ }
+ }
+}
+
+func TestCronNextIsStrictlyAfterAndUTC(t *testing.T) {
+ cases := []struct{ expr, after, want string }{
+ {"0 8 * * *", "2026-08-23T07:59:00Z", "2026-08-23T08:00:00Z"},
+ {"0 8 * * *", "2026-08-23T08:00:00Z", "2026-08-24T08:00:00Z"}, // strictly after
+ {"0 8 * * *", "2026-08-23T08:00:30Z", "2026-08-24T08:00:00Z"}, // seconds truncated: 08:00 has passed
+ {"*/15 * * * *", "2026-08-23T08:07:00Z", "2026-08-23T08:15:00Z"},
+ {"0 */6 * * *", "2026-08-23T13:00:00Z", "2026-08-23T18:00:00Z"},
+ {"0 9 * * 1", "2026-08-23T09:00:00Z", "2026-08-24T09:00:00Z"}, // 2026-08-23 is a Sunday
+ {"0 0 31 * *", "2026-09-01T00:00:00Z", "2026-10-31T00:00:00Z"}, // September has no 31st
+ {"0 0 29 2 *", "2026-03-01T00:00:00Z", "2028-02-29T00:00:00Z"},
+ {"0 0 * * 7", "2026-08-23T00:00:00Z", "2026-08-30T00:00:00Z"}, // 7 = Sunday
+ }
+ for _, c := range cases {
+ expr, err := ParseCron(c.expr)
+ if err != nil {
+ t.Fatalf("%q: %v", c.expr, err)
+ }
+ if got := expr.Next(at(c.after)); !got.Equal(at(c.want)) {
+ t.Errorf("%q after %s = %s, want %s", c.expr, c.after, got.Format(time.RFC3339), c.want)
+ }
+ if got := expr.Next(at(c.after)); got.Location() != time.UTC {
+ t.Errorf("%q: Next not in UTC", c.expr)
+ }
+ }
+}
+
+func TestCronNextHonoursNonUTCInput(t *testing.T) {
+ expr, _ := ParseCron("0 8 * * *")
+ loc := time.FixedZone("plus2", 2*3600)
+ // 09:30 +02:00 is 07:30Z — the next 08:00Z is the same day.
+ got := expr.Next(time.Date(2026, 8, 23, 9, 30, 0, 0, loc))
+ if !got.Equal(at("2026-08-23T08:00:00Z")) {
+ t.Fatalf("got %s", got.Format(time.RFC3339))
+ }
+}
+```
+
+- [ ] **Step 2: Run to verify failure**
+
+Run: `go test ./internal/schedule/ -run Cron -count=1`
+Expected: FAIL — `undefined: ParseCron`.
+
+- [ ] **Step 3: Implement `cron.go`**
+
+```go
+// Package schedule is the platform's answer to "run this later": the WHEN of
+// scheduled work, so an app that hibernates does not have to keep a process
+// alive to keep a timer. Spec: docs/superpowers/specs/2026-08-23-scheduled-work-design.md.
+//
+// This file is the cron dialect. Deliberately small (five fields, UTC, no
+// names, no @aliases — spec §4.1) and deliberately ours: the box binary is
+// built with GOPROXY=off, and a parser this size is cheaper to own than to
+// vendor.
+package schedule
+
+import (
+ "errors"
+ "fmt"
+ "strconv"
+ "strings"
+ "time"
+)
+
+// CronExpr is a parsed five-field expression. Each field is a bitset over
+// its legal range (minute 0-59, hour 0-23, day-of-month 1-31, month 1-12,
+// day-of-week 0-6 with 7 folded onto 0).
+type CronExpr struct {
+ min, hour, dom, month, dow uint64
+ raw string
+ domStar, dowStar bool // "*" in a day field — the classic OR-vs-AND rule below
+}
+
+func (c *CronExpr) String() string { return c.raw }
+
+// ParseCron accepts "min hour dom month dow" with *, n, a-b, */n, a-b/n and
+// comma lists. Anything else — names, six fields, @daily — is an error with
+// the field named, because the member typing it has no box to debug on.
+func ParseCron(expr string) (*CronExpr, error) {
+ fields := strings.Fields(expr)
+ if len(fields) != 5 {
+ return nil, fmt.Errorf("cron: want 5 fields (min hour day month weekday), got %d", len(fields))
+ }
+ c := &CronExpr{raw: strings.Join(fields, " ")}
+ var err error
+ if c.min, _, err = parseField(fields[0], 0, 59, "minute"); err != nil {
+ return nil, err
+ }
+ if c.hour, _, err = parseField(fields[1], 0, 23, "hour"); err != nil {
+ return nil, err
+ }
+ if c.dom, c.domStar, err = parseField(fields[2], 1, 31, "day"); err != nil {
+ return nil, err
+ }
+ if c.month, _, err = parseField(fields[3], 1, 12, "month"); err != nil {
+ return nil, err
+ }
+ if c.dow, c.dowStar, err = parseField(fields[4], 0, 7, "weekday"); err != nil {
+ return nil, err
+ }
+ if c.dow&(1<<7) != 0 { // 7 is Sunday too
+ c.dow = (c.dow &^ (1 << 7)) | 1
+ }
+ return c, nil
+}
+
+// parseField returns the bitset and whether the field was a bare "*".
+func parseField(f string, lo, hi int, name string) (uint64, bool, error) {
+ var bits uint64
+ star := f == "*"
+ for _, part := range strings.Split(f, ",") {
+ rangePart, stepPart, hasStep := strings.Cut(part, "/")
+ step := 1
+ if hasStep {
+ n, err := strconv.Atoi(stepPart)
+ if err != nil || n <= 0 {
+ return 0, false, fmt.Errorf("cron %s: bad step %q", name, part)
+ }
+ step = n
+ }
+ a, b := lo, hi
+ switch {
+ case rangePart == "*":
+ case strings.Contains(rangePart, "-"):
+ as, bs, _ := strings.Cut(rangePart, "-")
+ var err error
+ if a, err = atoiIn(as, lo, hi); err != nil {
+ return 0, false, fmt.Errorf("cron %s: %w", name, err)
+ }
+ if b, err = atoiIn(bs, lo, hi); err != nil {
+ return 0, false, fmt.Errorf("cron %s: %w", name, err)
+ }
+ if a > b {
+ return 0, false, fmt.Errorf("cron %s: range %q runs backwards", name, rangePart)
+ }
+ default:
+ n, err := atoiIn(rangePart, lo, hi)
+ if err != nil {
+ return 0, false, fmt.Errorf("cron %s: %w", name, err)
+ }
+ a, b = n, n
+ if hasStep {
+ b = hi // "5/10" means from 5 to the end, stepping
+ }
+ }
+ for v := a; v <= b; v += step {
+ bits |= 1 << uint(v)
+ }
+ }
+ if bits == 0 {
+ return 0, false, errors.New("cron " + name + ": empty")
+ }
+ return bits, star, nil
+}
+
+func atoiIn(s string, lo, hi int) (int, error) {
+ n, err := strconv.Atoi(s)
+ if err != nil {
+ return 0, fmt.Errorf("%q is not a number", s)
+ }
+ if n < lo || n > hi {
+ return 0, fmt.Errorf("%d is outside %d-%d", n, lo, hi)
+ }
+ return n, nil
+}
+
+// Next is the first matching minute strictly after `after`, in UTC, or the
+// zero time if none falls within 366 days (a "0 0 30 2 *" never matches).
+// Minute-stepping with day skips: cheap enough that cleverness would only
+// add bugs — the runner asks once per fire, not per tick.
+func (c *CronExpr) Next(after time.Time) time.Time {
+ t := after.UTC().Truncate(time.Minute).Add(time.Minute)
+ limit := t.Add(366 * 24 * time.Hour)
+ for t.Before(limit) {
+ if c.month&(1<<uint(t.Month())) == 0 {
+ // Skip to the first of next month.
+ t = time.Date(t.Year(), t.Month()+1, 1, 0, 0, 0, 0, time.UTC)
+ continue
+ }
+ if !c.dayMatches(t) {
+ t = time.Date(t.Year(), t.Month(), t.Day()+1, 0, 0, 0, 0, time.UTC)
+ continue
+ }
+ if c.hour&(1<<uint(t.Hour())) == 0 {
+ t = time.Date(t.Year(), t.Month(), t.Day(), t.Hour()+1, 0, 0, 0, time.UTC)
+ continue
+ }
+ if c.min&(1<<uint(t.Minute())) == 0 {
+ t = t.Add(time.Minute)
+ continue
+ }
+ return t
+ }
+ return time.Time{}
+}
+
+// dayMatches applies Vixie cron's rule: when BOTH day fields are restricted
+// the day matches if EITHER does; when one is "*" the other decides.
+func (c *CronExpr) dayMatches(t time.Time) bool {
+ dom := c.dom&(1<<uint(t.Day())) != 0
+ dow := c.dow&(1<<uint(t.Weekday())) != 0
+ switch {
+ case c.domStar && c.dowStar:
+ return true
+ case c.domStar:
+ return dow
+ case c.dowStar:
+ return dom
+ default:
+ return dom || dow
+ }
+}
+```
+
+- [ ] **Step 4: Run the cron tests**
+
+Run: `go test ./internal/schedule/ -run Cron -count=1`
+Expected: PASS.
+
+- [ ] **Step 5: Write the spec tests** (`spec_test.go`)
+
+```go
+package schedule
+
+import (
+ "testing"
+ "time"
+)
+
+func TestParseSpecRoundTrips(t *testing.T) {
+ for _, s := range []string{"every 6h0m0s", "cron 0 8 * * *", "at 1787004000"} {
+ sp, err := ParseSpec(s)
+ if err != nil {
+ t.Fatalf("%q: %v", s, err)
+ }
+ if got := sp.String(); got != s {
+ t.Errorf("%q → %q", s, got)
+ }
+ }
+ for _, bad := range []string{"", "every", "every 0s", "every 30s", "every 800h", "cron * *", "at x", "sometimes"} {
+ if _, err := ParseSpec(bad); err == nil {
+ t.Errorf("ParseSpec(%q) = ok, want error", bad)
+ }
+ }
+}
+
+// Epoch alignment (spec §6.1): "every 6h" fires at 00/06/12/18 UTC on every
+// box, whatever time the row was created, so there is no phase to persist
+// and nothing drifts across restarts.
+func TestEveryIsEpochAligned(t *testing.T) {
+ sp := Spec{Kind: KindSchedule, Every: 6 * time.Hour}
+ cases := map[string]string{
+ "2026-08-23T13:07:00Z": "2026-08-23T18:00:00Z",
+ "2026-08-23T18:00:00Z": "2026-08-24T00:00:00Z", // strictly after
+ "2026-08-23T23:59:59Z": "2026-08-24T00:00:00Z",
+ }
+ for now, want := range cases {
+ if got := sp.Next(at(now)); !got.Equal(at(want)) {
+ t.Errorf("Next(%s) = %s, want %s", now, got.Format(time.RFC3339), want)
+ }
+ }
+}
+
+// Coalescing (spec §6.1): Next is computed from NOW, so a box that was down
+// across several occurrences produces one next fire, not a burst.
+func TestNextNeverReturnsThePast(t *testing.T) {
+ sp := Spec{Kind: KindSchedule, Every: time.Hour}
+ now := at("2026-08-23T13:07:00Z")
+ if got := sp.Next(now); !got.After(now) {
+ t.Fatalf("Next(%s) = %s is not after now", now, got)
+ }
+}
+
+func TestOnceFiresOnceThenNever(t *testing.T) {
+ sp := Spec{Kind: KindOnce, At: at("2026-09-01T08:00:00Z")}
+ if got := sp.Next(at("2026-08-23T00:00:00Z")); !got.Equal(sp.At) {
+ t.Fatalf("before: %s", got)
+ }
+ if got := sp.Next(at("2026-09-01T08:00:00Z")); !got.IsZero() {
+ t.Fatalf("after firing: %s, want zero", got)
+ }
+}
+```
+
+- [ ] **Step 6: Implement `spec.go`**
+
+```go
+package schedule
+
+import (
+ "errors"
+ "fmt"
+ "strconv"
+ "strings"
+ "time"
+)
+
+const (
+ KindSchedule = "schedule"
+ KindOnce = "once"
+
+ MinEvery = time.Minute
+ MaxEvery = 720 * time.Hour
+ MaxOnceAhead = 400 * 24 * time.Hour
+)
+
+// Spec is the parsed timers.spec column — one of three shapes, and the only
+// thing the runner needs to know to compute the next fire.
+type Spec struct {
+ Kind string
+ Every time.Duration
+ Cron *CronExpr
+ At time.Time
+}
+
+// ParseSpec reads the canonical column text: "every <duration>",
+// "cron <5 fields>", "at <unix>". String is its inverse.
+func ParseSpec(s string) (Spec, error) {
+ kind, rest, _ := strings.Cut(strings.TrimSpace(s), " ")
+ rest = strings.TrimSpace(rest)
+ switch kind {
+ case "every":
+ d, err := time.ParseDuration(rest)
+ if err != nil {
+ return Spec{}, fmt.Errorf("every: %w", err)
+ }
+ if err := ValidEvery(d); err != nil {
+ return Spec{}, err
+ }
+ return Spec{Kind: KindSchedule, Every: d}, nil
+ case "cron":
+ c, err := ParseCron(rest)
+ if err != nil {
+ return Spec{}, err
+ }
+ return Spec{Kind: KindSchedule, Cron: c}, nil
+ case "at":
+ n, err := strconv.ParseInt(rest, 10, 64)
+ if err != nil {
+ return Spec{}, fmt.Errorf("at: %q is not a unix time", rest)
+ }
+ return Spec{Kind: KindOnce, At: time.Unix(n, 0).UTC()}, nil
+ }
+ return Spec{}, fmt.Errorf("spec %q: want every|cron|at", s)
+}
+
+// ValidEvery is the member-facing bound: whole minutes, 1m to 30d.
+func ValidEvery(d time.Duration) error {
+ switch {
+ case d < MinEvery:
+ return errors.New("every: shortest interval is 1m")
+ case d > MaxEvery:
+ return errors.New("every: longest interval is 720h (30 days)")
+ case d%time.Minute != 0:
+ return errors.New("every: whole minutes only")
+ }
+ return nil
+}
+
+func (s Spec) String() string {
+ switch {
+ case s.Every > 0:
+ return "every " + s.Every.String()
+ case s.Cron != nil:
+ return "cron " + s.Cron.String()
+ default:
+ return "at " + strconv.FormatInt(s.At.Unix(), 10)
+ }
+}
+
+// Next is the first fire strictly after now — epoch-aligned for every,
+// minute-stepped for cron, the instant itself for a pending once and the
+// zero time for a once that has passed. Always from now, never from the
+// scheduled time: that is what coalesces a missed run into one (spec §6.1).
+func (s Spec) Next(now time.Time) time.Time {
+ now = now.UTC()
+ switch {
+ case s.Every > 0:
+ n := now.Unix()
+ e := int64(s.Every / time.Second)
+ return time.Unix((n/e+1)*e, 0).UTC()
+ case s.Cron != nil:
+ return s.Cron.Next(now)
+ default:
+ if !s.At.After(now) {
+ return time.Time{}
+ }
+ return s.At.UTC()
+ }
+}
+```
+
+- [ ] **Step 7: Run spec tests** — `go test ./internal/schedule/ -count=1` → PASS.
+
+- [ ] **Step 8: Write declaration tests** (`declaration_test.go`)
+
+```go
+package schedule
+
+import (
+ "context"
+ "encoding/json"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func TestValidNameAndPath(t *testing.T) {
+ for _, ok := range []string{"a", "sync", "daily-digest", "x1", strings.Repeat("a", 32)} {
+ if !ValidName(ok) {
+ t.Errorf("ValidName(%q) = false", ok)
+ }
+ }
+ for _, bad := range []string{"", "-a", "A", "a_b", "a b", strings.Repeat("a", 33), "sync/"} {
+ if ValidName(bad) {
+ t.Errorf("ValidName(%q) = true", bad)
+ }
+ }
+ for _, ok := range []string{"/", "/jobs/sync", "/a/b-c_d.e?x=1"} {
+ if err := ValidPath(ok); err != nil {
+ t.Errorf("ValidPath(%q) = %v", ok, err)
+ }
+ }
+ for _, bad := range []string{"", "jobs/sync", "/jobs/../etc", "https://x/y", "//host/p", "/" + strings.Repeat("a", 256)} {
+ if err := ValidPath(bad); err == nil {
+ t.Errorf("ValidPath(%q) = ok", bad)
+ }
+ }
+}
+
+func TestScheduleValidateNormalisesEvery(t *testing.T) {
+ s := Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}
+ if err := s.Validate(); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := (Schedule{Name: "x", Every: "6h", Cron: "* * * * *", Path: "/p"}).Spec(); err == nil {
+ t.Error("both every and cron accepted")
+ }
+ if err := (Schedule{Name: "x", Path: "/p"}).Validate(); err == nil {
+ t.Error("neither every nor cron accepted")
+ }
+ if err := (Schedule{Name: "x", Every: "30s", Path: "/p"}).Validate(); err == nil {
+ t.Error("30s accepted")
+ }
+}
+
+func TestUpsertRemoveAndCAS(t *testing.T) {
+ ctx := context.Background()
+ st := release.NewDirStore(t.TempDir())
+ now := at("2026-08-23T22:00:00Z")
+ d, err := UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(d.Schedules) != 1 || d.Schema != 1 || d.UpdatedAt != "2026-08-23T22:00:00Z" {
+ t.Fatalf("declaration = %+v", d)
+ }
+ // Upsert by name replaces; a second name sorts in.
+ d, _ = UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "12h", Path: "/jobs/sync"}, now)
+ d, _ = UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "alpha", Cron: "0 8 * * *", Path: "/jobs/a"}, now)
+ if len(d.Schedules) != 2 || d.Schedules[0].Name != "alpha" || d.Schedules[1].Every != "12h0m0s" {
+ t.Fatalf("declaration = %+v", d)
+ }
+ raw, _, _ := st.Get(ctx, DeclarationKey("bab", "app"))
+ var onDisk Declaration
+ if err := json.Unmarshal(raw, &onDisk); err != nil || len(onDisk.Schedules) != 2 {
+ t.Fatalf("on disk: %s %v", raw, err)
+ }
+ // The cap.
+ for i := 0; i < MaxSchedulesPerApp; i++ {
+ UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "n" + string(rune('a'+i)), Every: "1h", Path: "/p"}, now)
+ }
+ if _, err := UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "one-too-many", Every: "1h", Path: "/p"}, now); err == nil {
+ t.Fatal("cap not enforced")
+ }
+ d, present, err := RemoveSchedule(ctx, st, "bab", "app", "alpha", now)
+ if err != nil || !present {
+ t.Fatalf("remove: %v present=%v", err, present)
+ }
+ if _, ok := d.Find("alpha"); ok {
+ t.Fatal("alpha still declared")
+ }
+ if _, present, _ := RemoveSchedule(ctx, st, "bab", "app", "alpha", now); present {
+ t.Fatal("second remove reports present")
+ }
+}
+
+func TestReadDeclarationAbsentIsEmpty(t *testing.T) {
+ d, etag, err := ReadDeclaration(context.Background(), release.NewDirStore(t.TempDir()), "bab", "app")
+ if err != nil || etag != "" || len(d.Schedules) != 0 {
+ t.Fatalf("%+v %q %v", d, etag, err)
+ }
+}
+
+func TestTouchRunMovesTheEtag(t *testing.T) {
+ ctx := context.Background()
+ st := release.NewDirStore(t.TempDir())
+ if err := TouchRun(ctx, st, "bab", "app", "sync", time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ _, e1, _ := st.Get(ctx, RunKey("bab", "app"))
+ if err := TouchRun(ctx, st, "bab", "app", "sync", time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ _, e2, _ := st.Get(ctx, RunKey("bab", "app"))
+ if e1 == "" || e1 == e2 {
+ t.Fatalf("etag did not move: %q → %q", e1, e2)
+ }
+}
+```
+
+- [ ] **Step 9: Implement `declaration.go`**
+
+```go
+package schedule
+
+import (
+ "context"
+ "crypto/rand"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "regexp"
+ "sort"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+const MaxSchedulesPerApp = 20
+
+var nameRE = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,31}$`)
+
+func ValidName(name string) bool { return nameRE.MatchString(name) }
+
+// ValidPath: absolute, no scheme/host, no "..", short. A path is what the
+// tick POSTs to on the instance's own socket — nothing here can reach
+// another host, but a bad one should fail at the terminal, not on the box.
+func ValidPath(p string) error {
+ switch {
+ case p == "" || p[0] != '/':
+ return errors.New("path must start with /")
+ case strings.HasPrefix(p, "//"):
+ return errors.New("path must not start with //")
+ case len(p) > 256:
+ return errors.New("path is longer than 256 bytes")
+ case strings.ContainsAny(p, " \t\r\n"):
+ return errors.New("path must not contain whitespace")
+ }
+ for _, seg := range strings.Split(strings.SplitN(p, "?", 2)[0], "/") {
+ if seg == ".." {
+ return errors.New("path must not contain ..")
+ }
+ }
+ return nil
+}
+
+// Schedule is one declared recurring job, as the member typed it.
+type Schedule struct {
+ Name string `json:"name"`
+ Every string `json:"every,omitempty"`
+ Cron string `json:"cron,omitempty"`
+ Path string `json:"path"`
+}
+
+// Spec is the schedule as the box stores it. Exactly one of every/cron.
+func (s Schedule) Spec() (Spec, error) {
+ switch {
+ case s.Every != "" && s.Cron != "":
+ return Spec{}, errors.New("give -every or -cron, not both")
+ case s.Every != "":
+ d, err := time.ParseDuration(s.Every)
+ if err != nil {
+ return Spec{}, fmt.Errorf("every: %w", err)
+ }
+ if err := ValidEvery(d); err != nil {
+ return Spec{}, err
+ }
+ return Spec{Kind: KindSchedule, Every: d}, nil
+ case s.Cron != "":
+ c, err := ParseCron(s.Cron)
+ if err != nil {
+ return Spec{}, err
+ }
+ return Spec{Kind: KindSchedule, Cron: c}, nil
+ }
+ return Spec{}, errors.New("give -every <duration> or -cron \"<5 fields>\"")
+}
+
+// Validate checks every field and NORMALISES every/cron to their canonical
+// text, so what lands in the bucket is what ParseSpec reads back.
+func (s *Schedule) Validate() error {
+ if !ValidName(s.Name) {
+ return fmt.Errorf("name %q: want [a-z0-9][a-z0-9-]{0,31}", s.Name)
+ }
+ if err := ValidPath(s.Path); err != nil {
+ return err
+ }
+ sp, err := s.Spec()
+ if err != nil {
+ return err
+ }
+ if sp.Every > 0 {
+ s.Every = sp.Every.String()
+ } else {
+ s.Cron = sp.Cron.String()
+ }
+ return nil
+}
+
+// Declaration is apps/<acct>/<app>/schedules.json (spec §4.1).
+type Declaration struct {
+ Schema int `json:"schema"`
+ UpdatedAt string `json:"updated_at"`
+ Schedules []Schedule `json:"schedules"`
+}
+
+func (d Declaration) Find(name string) (Schedule, bool) {
+ for _, s := range d.Schedules {
+ if s.Name == name {
+ return s, true
+ }
+ }
+ return Schedule{}, false
+}
+
+func DeclarationKey(acct, app string) string { return release.AppPrefix(acct, app) + "schedules.json" }
+func RunKey(acct, app string) string { return release.AppPrefix(acct, app) + "schedule-run.json" }
+func StatusPrefix(acct, app string) string { return release.AppPrefix(acct, app) + "schedule-status/" }
+func StatusKey(acct, app, host string) string { return StatusPrefix(acct, app) + host + ".json" }
+
+// ReadDeclaration: absent is the empty declaration with etag "" — legal
+// forever, like restart-gen.
+func ReadDeclaration(ctx context.Context, s release.Store, acct, app string) (Declaration, string, error) {
+ raw, etag, err := s.Get(ctx, DeclarationKey(acct, app))
+ if errors.Is(err, os.ErrNotExist) {
+ return Declaration{Schema: 1}, "", nil
+ }
+ if err != nil {
+ return Declaration{}, "", err
+ }
+ var d Declaration
+ if err := json.Unmarshal(raw, &d); err != nil {
+ return Declaration{}, "", fmt.Errorf("schedules.json: %w", err)
+ }
+ return d, etag, nil
+}
+
+func checkIDs(acct, app string) error {
+ if !release.ValidAccount(acct) {
+ return fmt.Errorf("schedule: invalid account %q", acct)
+ }
+ if !release.ValidApp(app) {
+ return fmt.Errorf("schedule: invalid app %q", app)
+ }
+ return nil
+}
+
+// modify applies fn under CAS with ONE retry on conflict — two members
+// editing the same app at once should both land, not one of them fail.
+func modify(ctx context.Context, s release.Store, acct, app string, now time.Time, fn func(*Declaration) error) (Declaration, error) {
+ if err := checkIDs(acct, app); err != nil {
+ return Declaration{}, err
+ }
+ var last error
+ for attempt := 0; attempt < 2; attempt++ {
+ d, etag, err := ReadDeclaration(ctx, s, acct, app)
+ if err != nil {
+ return Declaration{}, err
+ }
+ if err := fn(&d); err != nil {
+ return Declaration{}, err
+ }
+ sort.Slice(d.Schedules, func(i, j int) bool { return d.Schedules[i].Name < d.Schedules[j].Name })
+ d.Schema = 1
+ d.UpdatedAt = now.UTC().Format(time.RFC3339)
+ raw, err := json.Marshal(d)
+ if err != nil {
+ return Declaration{}, err
+ }
+ if _, err := s.Put(ctx, DeclarationKey(acct, app), raw, etag); err != nil {
+ if errors.Is(err, release.ErrCASConflict) {
+ last = err
+ continue
+ }
+ return Declaration{}, err
+ }
+ return d, nil
+ }
+ return Declaration{}, fmt.Errorf("schedules.json changed underneath us twice: %w", last)
+}
+
+// UpsertSchedule validates sc, then replaces-or-appends it by name.
+func UpsertSchedule(ctx context.Context, s release.Store, acct, app string, sc Schedule, now time.Time) (Declaration, error) {
+ if err := sc.Validate(); err != nil {
+ return Declaration{}, err
+ }
+ return modify(ctx, s, acct, app, now, func(d *Declaration) error {
+ for i := range d.Schedules {
+ if d.Schedules[i].Name == sc.Name {
+ d.Schedules[i] = sc
+ return nil
+ }
+ }
+ if len(d.Schedules) >= MaxSchedulesPerApp {
+ return fmt.Errorf("an app may declare at most %d schedules", MaxSchedulesPerApp)
+ }
+ d.Schedules = append(d.Schedules, sc)
+ return nil
+ })
+}
+
+// RemoveSchedule drops name; present reports whether it was there. Removing
+// an absent name still rewrites updated_at — harmless, and it keeps the
+// door's answer uniform.
+func RemoveSchedule(ctx context.Context, s release.Store, acct, app, name string, now time.Time) (Declaration, bool, error) {
+ present := false
+ d, err := modify(ctx, s, acct, app, now, func(d *Declaration) error {
+ kept := d.Schedules[:0]
+ for _, sc := range d.Schedules {
+ if sc.Name == name {
+ present = true
+ continue
+ }
+ kept = append(kept, sc)
+ }
+ d.Schedules = kept
+ return nil
+ })
+ return d, present, err
+}
+
+// RunSignal is schedule-run.json: a touch object whose etag movement is the
+// signal, exactly restart-gen's shape and rules (release/restartgen.go).
+type RunSignal struct {
+ Schema int `json:"schema"`
+ Name string `json:"name"`
+ TouchedAt string `json:"touched_at"`
+ Nonce string `json:"nonce"`
+}
+
+// TouchRun bumps the run signal for name. A CAS conflict is success: the
+// signal already fired.
+func TouchRun(ctx context.Context, s release.Store, acct, app, name string, now time.Time) error {
+ if err := checkIDs(acct, app); err != nil {
+ return err
+ }
+ if !ValidName(name) {
+ return fmt.Errorf("schedule: invalid name %q", name)
+ }
+ key := RunKey(acct, app)
+ _, etag, err := s.Get(ctx, key)
+ if err != nil && !errors.Is(err, os.ErrNotExist) {
+ return err
+ }
+ if errors.Is(err, os.ErrNotExist) {
+ etag = ""
+ }
+ nonce := make([]byte, 16)
+ if _, err := io.ReadFull(rand.Reader, nonce); err != nil {
+ return err
+ }
+ raw, err := json.Marshal(RunSignal{Schema: 1, Name: name, TouchedAt: now.UTC().Format(time.RFC3339), Nonce: hex.EncodeToString(nonce)})
+ if err != nil {
+ return err
+ }
+ if _, err := s.Put(ctx, key, raw, etag); err != nil && !errors.Is(err, release.ErrCASConflict) {
+ return err
+ }
+ return nil
+}
+```
+
+- [ ] **Step 10: Write status tests** (`status_test.go`)
+
+```go
+package schedule
+
+import (
+ "context"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func TestWriteAndListStatus(t *testing.T) {
+ ctx := context.Background()
+ st := release.NewDirStore(t.TempDir())
+ b := Status{Host: "b.bab.oncarlos.com", Box: "box1", ReportedAt: "2026-08-23T22:00:00Z", Timers: []TimerStatus{{Name: "sync", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: 100}}}
+ a := Status{Host: "a.bab.oncarlos.com", Box: "box1", ReportedAt: "2026-08-23T22:00:00Z", Unsupported: "unit-backed instance"}
+ for _, s := range []Status{b, a, b} { // the rewrite must not conflict
+ if err := WriteStatus(ctx, st, "bab", "app", s); err != nil {
+ t.Fatal(err)
+ }
+ }
+ got, err := ListStatus(ctx, st, "bab", "app")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(got) != 2 || got[0].Host != a.Host || got[1].Timers[0].Name != "sync" || got[0].Schema != 1 {
+ t.Fatalf("list = %+v", got)
+ }
+ if got[0].Timers == nil {
+ t.Fatal("timers must marshal as [] not null — the CLI ranges over it")
+ }
+}
+
+func TestListStatusEmptyPrefix(t *testing.T) {
+ got, err := ListStatus(context.Background(), release.NewDirStore(t.TempDir()), "bab", "app")
+ if err != nil || len(got) != 0 {
+ t.Fatalf("%v %v", got, err)
+ }
+}
+```
+
+- [ ] **Step 11: Implement `status.go`**
+
+```go
+package schedule
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "sort"
+ "strings"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// TimerStatus is one row of a box's report, and one row of `carlos schedule
+// ls` — the same shape end to end so the member reads what the box knows.
+type TimerStatus struct {
+ Name string `json:"name"`
+ Kind string `json:"kind"`
+ Spec string `json:"spec"`
+ Path string `json:"path"`
+ NextAt int64 `json:"next_at"`
+ LastAt int64 `json:"last_at"`
+ LastStatus string `json:"last_status"`
+ LastMs int64 `json:"last_ms"`
+ Attempts int `json:"attempts"`
+}
+
+// Status is apps/<acct>/<app>/schedule-status/<host>.json (spec §4.3):
+// what THIS instance will do next and what happened last, written by the
+// box that holds it. Unsupported, when set, says why Timers is empty.
+type Status struct {
+ Schema int `json:"schema"`
+ Host string `json:"host"`
+ Box string `json:"box"`
+ ReportedAt string `json:"reported_at"`
+ Unsupported string `json:"unsupported"`
+ Timers []TimerStatus `json:"timers"`
+}
+
+// WriteStatus replaces the host's report. The store's Put is CAS-only, so
+// this reads the current etag first; a conflict (two writers for one host —
+// not a designed state) re-reads once. Never loops: a report is
+// best-effort and the next fire rewrites it anyway.
+func WriteStatus(ctx context.Context, s release.Store, acct, app string, st Status) error {
+ if err := checkIDs(acct, app); err != nil {
+ return err
+ }
+ st.Schema = 1
+ if st.Timers == nil {
+ st.Timers = []TimerStatus{}
+ }
+ raw, err := json.Marshal(st)
+ if err != nil {
+ return err
+ }
+ key := StatusKey(acct, app, st.Host)
+ for attempt := 0; attempt < 2; attempt++ {
+ _, etag, err := s.Get(ctx, key)
+ if errors.Is(err, os.ErrNotExist) {
+ etag = ""
+ } else if err != nil {
+ return err
+ }
+ if _, err := s.Put(ctx, key, raw, etag); err == nil {
+ return nil
+ } else if !errors.Is(err, release.ErrCASConflict) {
+ return err
+ }
+ }
+ return fmt.Errorf("status for %s: conflicting writer", st.Host)
+}
+
+// ListStatus reads every host's report for the app, sorted by host. An
+// unreadable object is skipped and its error joined into the return, with
+// the readable ones still returned — one bad report must not blank the list.
+func ListStatus(ctx context.Context, s release.Store, acct, app string) ([]Status, error) {
+ if err := checkIDs(acct, app); err != nil {
+ return nil, err
+ }
+ keys, err := s.List(ctx, StatusPrefix(acct, app))
+ if err != nil {
+ return nil, err
+ }
+ var out []Status
+ var errs []error
+ for _, k := range keys {
+ if !strings.HasSuffix(k, ".json") {
+ continue
+ }
+ raw, _, err := s.Get(ctx, k)
+ if err != nil {
+ errs = append(errs, err)
+ continue
+ }
+ var st Status
+ if err := json.Unmarshal(raw, &st); err != nil {
+ errs = append(errs, fmt.Errorf("%s: %w", k, err))
+ continue
+ }
+ if st.Timers == nil {
+ st.Timers = []TimerStatus{}
+ }
+ out = append(out, st)
+ }
+ sort.Slice(out, func(i, j int) bool { return out[i].Host < out[j].Host })
+ return out, errors.Join(errs...)
+}
+```
+
+Check `release.Store.List` returns FULL keys (it does for DirStore — see `store.go`); if it returns names relative to the prefix, join them. The test tells you.
+
+- [ ] **Step 12: Gate and commit**
+
+Run: `go test ./internal/schedule/ -count=1 && go vet ./internal/schedule/ && gofmt -l internal/schedule`
+Expected: PASS, nothing listed.
+
+```bash
+git add internal/schedule
+git commit -m "schedule: specs, cron dialect, declarations and status objects
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+```
+
+---
+
+### Task 2: `internal/registry` — the `timers` table
+
+**Files:**
+- Create: `internal/registry/timers.go`, `internal/registry/timers_test.go`
+- Modify: `internal/registry/registry.go` — the `migrate()` function (add the CREATE TABLE after the existing ALTERs; look at how `aliases`/`wildcard_domains` tables are created and do the same).
+
+**Interfaces:**
+- Produces:
+
+```go
+package registry
+
+type Timer struct {
+ Host, Name, Kind, Spec, Path string
+ NextAt, LastAt int64
+ LastStatus string
+ LastMs int64
+ Attempts int
+ CreatedAt int64
+}
+
+// UpsertTimer inserts, or updates spec/path/next_at of an existing (host,name)
+// while PRESERVING last_at/last_status/last_ms/attempts. Kind is fixed at insert:
+// updating a row of a different kind returns ErrTimerKind.
+func (r *Registry) UpsertTimer(t Timer) error
+func (r *Registry) DeleteTimer(host, name string) error // idempotent
+func (r *Registry) GetTimer(host, name string) (*Timer, error) // os.ErrNotExist when absent
+func (r *Registry) TimersFor(host string) ([]Timer, error) // ordered by name
+func (r *Registry) DueTimers(now int64, limit int) ([]Timer, error) // next_at > 0 AND next_at <= now, ordered by next_at, host, name
+func (r *Registry) FireTimerNow(host, name string, now int64) error // sets next_at=now; no-op when absent
+func (r *Registry) MarkTimerFired(host, name string, at int64, status string, ms int64, attempts int, next int64) error
+func (r *Registry) CountTimers(host, kind string) (int, error)
+// PruneTimers deletes rows whose host is not in alive, and once-rows with next_at=0 and last_at < doneBefore. Returns the count.
+func (r *Registry) PruneTimers(alive map[string]bool, doneBefore int64) (int, error)
+// DeleteTimersExcept deletes rows of `kind` for host whose name is not in keep.
+func (r *Registry) DeleteTimersExcept(host, kind string, keep map[string]bool) (int, error)
+var ErrTimerKind = errors.New("registry: timer exists with a different kind")
+```
+
+- [ ] **Step 1: Write the tests** (`timers_test.go`) — open a registry the way `registry_test.go` does (find its helper; it is `Open(filepath.Join(t.TempDir(), "r.db"))` or a wrapper — use the same).
+
+```go
+package registry
+
+import (
+ "errors"
+ "os"
+ "path/filepath"
+ "testing"
+)
+
+func openTimersReg(t *testing.T) *Registry {
+ t.Helper()
+ r, err := Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { r.Close() })
+ return r
+}
+
+func TestTimersUpsertPreservesRunnerFields(t *testing.T) {
+ r := openTimersReg(t)
+ if err := r.UpsertTimer(Timer{Host: "h", Name: "sync", Kind: "schedule", Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: 100, CreatedAt: 1}); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.MarkTimerFired("h", "sync", 100, "204", 300, 0, 21700); err != nil {
+ t.Fatal(err)
+ }
+ // Converge sees the same declaration again: spec/path unchanged, next_at
+ // passed along as "whatever it is now" (0 here means "don't touch").
+ if err := r.UpsertTimer(Timer{Host: "h", Name: "sync", Kind: "schedule", Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: 0, CreatedAt: 99}); err != nil {
+ t.Fatal(err)
+ }
+ got, err := r.GetTimer("h", "sync")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got.LastAt != 100 || got.LastStatus != "204" || got.LastMs != 300 || got.NextAt != 21700 || got.CreatedAt != 1 {
+ t.Fatalf("runner fields clobbered: %+v", got)
+ }
+ // A changed spec DOES move next_at.
+ if err := r.UpsertTimer(Timer{Host: "h", Name: "sync", Kind: "schedule", Spec: "every 12h0m0s", Path: "/jobs/sync", NextAt: 43200}); err != nil {
+ t.Fatal(err)
+ }
+ got, _ = r.GetTimer("h", "sync")
+ if got.NextAt != 43200 || got.Spec != "every 12h0m0s" || got.LastStatus != "204" {
+ t.Fatalf("after spec change: %+v", got)
+ }
+ if err := r.UpsertTimer(Timer{Host: "h", Name: "sync", Kind: "once", Spec: "at 5", Path: "/p", NextAt: 5}); !errors.Is(err, ErrTimerKind) {
+ t.Fatalf("kind change = %v, want ErrTimerKind", err)
+ }
+}
+
+func TestTimersDueOrderingAndFireNow(t *testing.T) {
+ r := openTimersReg(t)
+ for _, tm := range []Timer{
+ {Host: "b", Name: "late", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 500},
+ {Host: "a", Name: "soon", Kind: "once", Spec: "at 10", Path: "/p", NextAt: 10},
+ {Host: "a", Name: "done", Kind: "once", Spec: "at 1", Path: "/p", NextAt: 0},
+ {Host: "c", Name: "future", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 9999},
+ } {
+ if err := r.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ }
+ due, err := r.DueTimers(600, 10)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(due) != 2 || due[0].Name != "soon" || due[1].Name != "late" {
+ t.Fatalf("due = %+v", due)
+ }
+ if due, _ := r.DueTimers(600, 1); len(due) != 1 {
+ t.Fatal("limit ignored")
+ }
+ if err := r.FireTimerNow("c", "future", 600); err != nil {
+ t.Fatal(err)
+ }
+ if due, _ := r.DueTimers(600, 10); len(due) != 3 {
+ t.Fatalf("after FireTimerNow: %d due", len(due))
+ }
+ if err := r.FireTimerNow("nope", "x", 600); err != nil {
+ t.Fatalf("FireTimerNow on absent = %v, want nil", err)
+ }
+}
+
+func TestTimersPruneAndExcept(t *testing.T) {
+ r := openTimersReg(t)
+ for _, tm := range []Timer{
+ {Host: "alive", Name: "s1", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 1},
+ {Host: "alive", Name: "s2", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 1},
+ {Host: "alive", Name: "o1", Kind: "once", Spec: "at 1", Path: "/p", NextAt: 0, LastAt: 10},
+ {Host: "alive", Name: "o2", Kind: "once", Spec: "at 1", Path: "/p", NextAt: 0, LastAt: 500},
+ {Host: "gone", Name: "s1", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 1},
+ } {
+ if err := r.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ }
+ // o1 has last_at=10 via insert — MarkTimerFired is the real path but insert suffices to seed.
+ n, err := r.PruneTimers(map[string]bool{"alive": true}, 100)
+ if err != nil || n != 2 { // gone/s1 and alive/o1
+ t.Fatalf("prune = %d %v", n, err)
+ }
+ n, err = r.DeleteTimersExcept("alive", "schedule", map[string]bool{"s2": true})
+ if err != nil || n != 1 {
+ t.Fatalf("except = %d %v", n, err)
+ }
+ left, _ := r.TimersFor("alive")
+ if len(left) != 2 || left[0].Name != "o2" || left[1].Name != "s2" {
+ t.Fatalf("left = %+v", left)
+ }
+ if c, _ := r.CountTimers("alive", "once"); c != 1 {
+ t.Fatalf("count = %d", c)
+ }
+ if _, err := r.GetTimer("gone", "s1"); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("gone = %v", err)
+ }
+ if err := r.DeleteTimer("alive", "s2"); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.DeleteTimer("alive", "s2"); err != nil {
+ t.Fatal("delete not idempotent")
+ }
+}
+```
+
+Note: `UpsertTimer` with `NextAt: 0` on an EXISTING row whose spec/path are unchanged leaves `next_at` alone; on INSERT it stores 0 as given (that is how a done once-row is seeded in the prune test). `LastAt` on insert is stored as given (seeding); on update it is never touched.
+
+- [ ] **Step 2: Run to verify failure** — `go test ./internal/registry/ -run Timers -count=1` → undefined.
+
+- [ ] **Step 3: Migration** — in `migrate()`, after the routes ALTER loop and alongside the other tables' `CREATE TABLE IF NOT EXISTS`:
+
+```go
+ if _, err := r.sql.Exec(`CREATE TABLE IF NOT EXISTS timers (
+ host TEXT NOT NULL,
+ name TEXT NOT NULL,
+ kind TEXT NOT NULL,
+ spec TEXT NOT NULL,
+ path TEXT NOT NULL,
+ next_at INTEGER NOT NULL,
+ last_at INTEGER NOT NULL DEFAULT 0,
+ last_status TEXT NOT NULL DEFAULT '',
+ last_ms INTEGER NOT NULL DEFAULT 0,
+ attempts INTEGER NOT NULL DEFAULT 0,
+ created_at INTEGER NOT NULL,
+ PRIMARY KEY (host, name)
+);
+CREATE INDEX IF NOT EXISTS timers_due ON timers (next_at) WHERE next_at > 0;`); err != nil {
+ return err
+ }
+```
+
+(Use whatever the file's actual db handle field is — read `migrate()` first; it may be `r.db`.) Do NOT touch `Fingerprint()` or `SnapshotFingerprint()`.
+
+- [ ] **Step 4: Implement `timers.go`**
+
+```go
+package registry
+
+// The timers table: scheduled work materialized per (host, name) — spec
+// docs/superpowers/specs/2026-08-23-scheduled-work-design.md §5. Rows come
+// from two writers with disjoint columns: converge (schedule.Converge) and
+// the control socket own spec/path/kind; the runner owns next_at after the
+// first fire, last_*, attempts. UpsertTimer encodes that split so neither
+// writer can clobber the other. Deliberately NOT in Fingerprint(): a timer
+// changes nothing about routing.
+
+import (
+ "database/sql"
+ "errors"
+ "os"
+ "strings"
+)
+
+type Timer struct {
+ Host, Name, Kind, Spec, Path string
+ NextAt, LastAt int64
+ LastStatus string
+ LastMs int64
+ Attempts int
+ CreatedAt int64
+}
+
+var ErrTimerKind = errors.New("registry: timer exists with a different kind")
+
+const timerCols = `host, name, kind, spec, path, next_at, last_at, last_status, last_ms, attempts, created_at`
+
+func scanTimer(row interface{ Scan(...any) error }) (Timer, error) {
+ var t Timer
+ err := row.Scan(&t.Host, &t.Name, &t.Kind, &t.Spec, &t.Path, &t.NextAt, &t.LastAt, &t.LastStatus, &t.LastMs, &t.Attempts, &t.CreatedAt)
+ return t, err
+}
+
+// UpsertTimer inserts t, or updates an existing row's spec/path — moving
+// next_at only when spec or path changed or t.NextAt is non-zero — and
+// never touches the runner's columns on update.
+func (r *Registry) UpsertTimer(t Timer) error {
+ existing, err := r.GetTimer(t.Host, t.Name)
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ _, err := r.sql.Exec(`INSERT INTO timers (`+timerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?)`,
+ t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt)
+ return err
+ case err != nil:
+ return err
+ }
+ if existing.Kind != t.Kind {
+ return ErrTimerKind
+ }
+ next := existing.NextAt
+ if existing.Spec != t.Spec || existing.Path != t.Path || t.NextAt != 0 {
+ next = t.NextAt
+ }
+ _, err = r.sql.Exec(`UPDATE timers SET spec=?, path=?, next_at=? WHERE host=? AND name=?`, t.Spec, t.Path, next, t.Host, t.Name)
+ return err
+}
+
+func (r *Registry) GetTimer(host, name string) (*Timer, error) {
+ t, err := scanTimer(r.sql.QueryRow(`SELECT `+timerCols+` FROM timers WHERE host=? AND name=?`, host, name))
+ if errors.Is(err, sql.ErrNoRows) {
+ return nil, os.ErrNotExist
+ }
+ if err != nil {
+ return nil, err
+ }
+ return &t, nil
+}
+
+func (r *Registry) DeleteTimer(host, name string) error {
+ _, err := r.sql.Exec(`DELETE FROM timers WHERE host=? AND name=?`, host, name)
+ return err
+}
+
+func (r *Registry) queryTimers(q string, args ...any) ([]Timer, error) {
+ rows, err := r.sql.Query(q, args...)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ var out []Timer
+ for rows.Next() {
+ t, err := scanTimer(rows)
+ if err != nil {
+ return nil, err
+ }
+ out = append(out, t)
+ }
+ return out, rows.Err()
+}
+
+func (r *Registry) TimersFor(host string) ([]Timer, error) {
+ return r.queryTimers(`SELECT `+timerCols+` FROM timers WHERE host=? ORDER BY name`, host)
+}
+
+func (r *Registry) DueTimers(now int64, limit int) ([]Timer, error) {
+ return r.queryTimers(`SELECT `+timerCols+` FROM timers WHERE next_at > 0 AND next_at <= ? ORDER BY next_at, host, name LIMIT ?`, now, limit)
+}
+
+func (r *Registry) FireTimerNow(host, name string, now int64) error {
+ _, err := r.sql.Exec(`UPDATE timers SET next_at=? WHERE host=? AND name=?`, now, host, name)
+ return err
+}
+
+func (r *Registry) MarkTimerFired(host, name string, at int64, status string, ms int64, attempts int, next int64) error {
+ _, err := r.sql.Exec(`UPDATE timers SET last_at=?, last_status=?, last_ms=?, attempts=?, next_at=? WHERE host=? AND name=?`,
+ at, status, ms, attempts, next, host, name)
+ return err
+}
+
+func (r *Registry) CountTimers(host, kind string) (int, error) {
+ var n int
+ err := r.sql.QueryRow(`SELECT COUNT(*) FROM timers WHERE host=? AND kind=?`, host, kind).Scan(&n)
+ return n, err
+}
+
+// PruneTimers: rows for hosts no longer routed here, and finished once-rows
+// older than doneBefore. An empty alive set deletes every row — callers pass
+// the registry's own host list, so that only happens on an empty registry.
+func (r *Registry) PruneTimers(alive map[string]bool, doneBefore int64) (int, error) {
+ hosts := make([]any, 0, len(alive))
+ for h := range alive {
+ hosts = append(hosts, h)
+ }
+ q := `DELETE FROM timers WHERE (kind='once' AND next_at=0 AND last_at < ?)`
+ args := []any{doneBefore}
+ if len(hosts) == 0 {
+ q += ` OR 1=1`
+ } else {
+ q += ` OR host NOT IN (?` + strings.Repeat(",?", len(hosts)-1) + `)`
+ args = append(args, hosts...)
+ }
+ res, err := r.sql.Exec(q, args...)
+ if err != nil {
+ return 0, err
+ }
+ n, _ := res.RowsAffected()
+ return int(n), nil
+}
+
+func (r *Registry) DeleteTimersExcept(host, kind string, keep map[string]bool) (int, error) {
+ names := make([]any, 0, len(keep))
+ for n := range keep {
+ names = append(names, n)
+ }
+ q := `DELETE FROM timers WHERE host=? AND kind=?`
+ args := []any{host, kind}
+ if len(names) > 0 {
+ q += ` AND name NOT IN (?` + strings.Repeat(",?", len(names)-1) + `)`
+ args = append(args, names...)
+ }
+ res, err := r.sql.Exec(q, args...)
+ if err != nil {
+ return 0, err
+ }
+ n, _ := res.RowsAffected()
+ return int(n), nil
+}
+```
+
+Adjust `r.sql` to the Registry's actual `*sql.DB` field name. The registry uses `SetMaxOpenConns(1)`; keep the single-statement style above (no nested queries inside an open `rows`).
+
+- [ ] **Step 5: Run and commit**
+
+Run: `go test ./internal/registry/ -count=1 && go vet ./internal/registry/ && gofmt -l internal/registry`
+
+```bash
+git add internal/registry
+git commit -m "registry: timers table for scheduled work
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+```
+
+---
+
+### Task 3: `schedule.Converge` — declaration → rows
+
+**Files:**
+- Create: `internal/schedule/converge.go`, `internal/schedule/converge_test.go`
+
+**Interfaces:**
+- Consumes: Task 1 (`ReadDeclaration`, `RunKey`, `Schedule.Spec`, `Status`, `WriteStatus`), Task 2 (`registry.Timer`, `UpsertTimer`, `DeleteTimersExcept`, `FireTimerNow`, `PruneTimers`, `TimersFor`), `registry.Route{Host, Kind, Account, App, Backing}`, `registry.BackingUnit`, `release.RouteAccount(rt)` (the accessor `activator/logsink.go` uses — check its signature and use it, not `rt.Account` directly).
+- Produces:
+
+```go
+// Converger keeps etag memory across passes so an unchanged declaration costs
+// one GET and no parse, and so a run-signal is acted on exactly once.
+type Converger struct {
+ Reg *registry.Registry
+ Store release.Store
+ Now func() time.Time
+ Box string // reported in Status.Box
+ Report func(rt registry.Route, st Status) // called for every host whose rows changed or that is unsupported; nil = no reports
+ declEtag map[[2]string]string // (acct, app) → last etag seen
+ runEtag map[[2]string]string
+ seededRun bool
+}
+func NewConverger(reg *registry.Registry, store release.Store, box string, report func(registry.Route, Status)) *Converger
+func (c *Converger) Pass(ctx context.Context) error // die-free per app; returns joined errors for logging
+func StatusFor(reg *registry.Registry, rt registry.Route, box string, now time.Time) (Status, error) // builds the report from TimersFor
+```
+
+- [ ] **Step 1: Write the tests** (`converge_test.go`). Open a registry via `registry.Open(filepath.Join(t.TempDir(), "r.db"))`; add routes with `reg.Upsert(registry.Route{Host: "a.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: true})` — read `registry.Route` and `Upsert` to fill whatever fields Upsert requires (check `registry_test.go` for the minimal route it uses).
+
+```go
+package schedule
+
+import (
+ "context"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func convergeFixture(t *testing.T) (*registry.Registry, *release.DirStore) {
+ t.Helper()
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ for _, rt := range []registry.Route{
+ {Host: "a.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: true},
+ {Host: "b.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: true},
+ {Host: "u.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Backing: registry.BackingUnit},
+ {Host: "other.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "other", Status: "ready"},
+ {Host: "site.bab.oncarlos.com", Kind: "static", Account: "bab", App: "site", Status: "ready"},
+ } {
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ }
+ return reg, release.NewDirStore(t.TempDir())
+}
+
+func TestConvergeMaterialisesDeclarationPerExecInstance(t *testing.T) {
+ reg, st := convergeFixture(t)
+ ctx := context.Background()
+ now := at("2026-08-23T13:07:00Z")
+ UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
+ var reports []Status
+ c := NewConverger(reg, st, "box1", func(_ registry.Route, s Status) { reports = append(reports, s) })
+ c.Now = func() time.Time { return now }
+ if err := c.Pass(ctx); err != nil {
+ t.Fatal(err)
+ }
+ for _, h := range []string{"a.bab.oncarlos.com", "b.bab.oncarlos.com"} {
+ tm, err := reg.GetTimer(h, "sync")
+ if err != nil {
+ t.Fatalf("%s: %v", h, err)
+ }
+ if tm.Kind != KindSchedule || tm.Spec != "every 6h0m0s" || tm.Path != "/jobs/sync" || tm.NextAt != at("2026-08-23T18:00:00Z").Unix() {
+ t.Fatalf("%s: %+v", h, tm)
+ }
+ }
+ if rows, _ := reg.TimersFor("u.bab.oncarlos.com"); len(rows) != 0 {
+ t.Fatal("unit-backed host got rows")
+ }
+ if rows, _ := reg.TimersFor("other.bab.oncarlos.com"); len(rows) != 0 {
+ t.Fatal("undeclared app got rows")
+ }
+ // Reports: a and b with a timer each, u as unsupported. Nothing for site.
+ byHost := map[string]Status{}
+ for _, r := range reports {
+ byHost[r.Host] = r
+ }
+ if len(byHost["a.bab.oncarlos.com"].Timers) != 1 || byHost["a.bab.oncarlos.com"].Box != "box1" {
+ t.Fatalf("report a = %+v", byHost["a.bab.oncarlos.com"])
+ }
+ if byHost["u.bab.oncarlos.com"].Unsupported == "" {
+ t.Fatalf("unit host not reported unsupported: %+v", byHost)
+ }
+ if _, ok := byHost["site.bab.oncarlos.com"]; ok {
+ t.Fatal("static route reported")
+ }
+
+ // A second pass with nothing changed: no new reports, next_at untouched.
+ reports = nil
+ reg.MarkTimerFired("a.bab.oncarlos.com", "sync", now.Unix(), "204", 10, 0, 999)
+ if err := c.Pass(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if len(reports) != 0 {
+ t.Fatalf("unchanged pass reported %d hosts", len(reports))
+ }
+ if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 999 {
+ t.Fatalf("unchanged pass moved next_at: %+v", tm)
+ }
+}
+
+func TestConvergeRemovesUndeclaredAndKeepsOnce(t *testing.T) {
+ reg, st := convergeFixture(t)
+ ctx := context.Background()
+ now := at("2026-08-23T13:07:00Z")
+ UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
+ UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "gone", Every: "1h", Path: "/jobs/gone"}, now)
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "remind", Kind: KindOnce, Spec: "at 99999999999", Path: "/r", NextAt: 99999999999})
+ c := NewConverger(reg, st, "box1", nil)
+ c.Now = func() time.Time { return now }
+ c.Pass(ctx)
+ RemoveSchedule(ctx, st, "bab", "app", "gone", now)
+ if err := c.Pass(ctx); err != nil {
+ t.Fatal(err)
+ }
+ rows, _ := reg.TimersFor("a.bab.oncarlos.com")
+ names := map[string]bool{}
+ for _, r := range rows {
+ names[r.Name] = true
+ }
+ if !names["sync"] || !names["remind"] || names["gone"] {
+ t.Fatalf("rows = %v", names)
+ }
+}
+
+func TestConvergeRunSignalFiresOnceAcrossInstances(t *testing.T) {
+ reg, st := convergeFixture(t)
+ ctx := context.Background()
+ now := at("2026-08-23T13:07:00Z")
+ UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
+ c := NewConverger(reg, st, "box1", nil)
+ c.Now = func() time.Time { return now }
+ c.Pass(ctx)
+ // A run signal that predates this converger is history, not a request.
+ TouchRun(ctx, st, "bab", "app", "sync", now)
+ c2 := NewConverger(reg, st, "box1", nil)
+ c2.Now = func() time.Time { return now }
+ c2.Pass(ctx)
+ if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt == now.Unix() {
+ t.Fatal("a pre-existing run signal was replayed at seed")
+ }
+ TouchRun(ctx, st, "bab", "app", "sync", now)
+ c2.Pass(ctx)
+ for _, h := range []string{"a.bab.oncarlos.com", "b.bab.oncarlos.com"} {
+ if tm, _ := reg.GetTimer(h, "sync"); tm.NextAt != now.Unix() {
+ t.Fatalf("%s not fired: %+v", h, tm)
+ }
+ }
+ reg.MarkTimerFired("a.bab.oncarlos.com", "sync", now.Unix(), "204", 1, 0, 777)
+ c2.Pass(ctx) // same etag: must not re-fire
+ if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 777 {
+ t.Fatal("run signal replayed on an unchanged etag")
+ }
+}
+
+func TestConvergePrunesRemovedHosts(t *testing.T) {
+ reg, st := convergeFixture(t)
+ ctx := context.Background()
+ reg.UpsertTimer(registry.Timer{Host: "removed.bab.oncarlos.com", Name: "x", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/p", NextAt: 1})
+ c := NewConverger(reg, st, "box1", nil)
+ if err := c.Pass(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if rows, _ := reg.TimersFor("removed.bab.oncarlos.com"); len(rows) != 0 {
+ t.Fatal("orphan rows survived")
+ }
+}
+```
+
+- [ ] **Step 2: Run to verify failure.**
+
+- [ ] **Step 3: Implement `converge.go`**
+
+```go
+package schedule
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log"
+ "os"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// Converger turns apps/<acct>/<app>/schedules.json into timers rows for every
+// exec-backed instance this box serves, acts on schedule-run.json, and prunes
+// what no longer belongs (spec §6.2). One instance per agent; Pass is called
+// on every serial move, hourly, and once at start.
+//
+// Etag memory is what keeps an idle box cheap and a run-signal
+// exactly-once: an unchanged declaration is one GET and no writes; a run
+// signal fires when its etag MOVES, and the first sighting only seeds
+// (a signal older than this process is history — restart-gen's rule).
+type Converger struct {
+ Reg *registry.Registry
+ Store release.Store
+ Now func() time.Time
+ Box string
+ Report func(rt registry.Route, st Status)
+
+ declEtag map[[2]string]string
+ declDoc map[[2]string]Declaration
+ runEtag map[[2]string]string
+}
+
+const onceRetention = 7 * 24 * time.Hour
+
+func NewConverger(reg *registry.Registry, store release.Store, box string, report func(registry.Route, Status)) *Converger {
+ return &Converger{Reg: reg, Store: store, Now: time.Now, Box: box, Report: report,
+ declEtag: map[[2]string]string{}, declDoc: map[[2]string]Declaration{}, runEtag: map[[2]string]string{}}
+}
+
+// Pass is die-free per app: one app's unreadable declaration is logged and
+// its rows left exactly as they were (a baseline with holes manufactures
+// deletions), and every other app still converges.
+func (c *Converger) Pass(ctx context.Context) error {
+ routes, err := c.Reg.List()
+ if err != nil {
+ return err
+ }
+ now := c.Now()
+ alive := map[string]bool{}
+ byApp := map[[2]string][]registry.Route{}
+ for _, rt := range routes {
+ alive[rt.Host] = true
+ if rt.Kind != "instance" {
+ continue
+ }
+ k := [2]string{release.RouteAccount(rt), rt.App}
+ if k[0] == "" || k[1] == "" {
+ continue
+ }
+ byApp[k] = append(byApp[k], rt)
+ }
+ var errs []error
+ for k, rts := range byApp {
+ if err := c.convergeApp(ctx, k[0], k[1], rts, now); err != nil {
+ errs = append(errs, fmt.Errorf("%s/%s: %w", k[0], k[1], err))
+ }
+ }
+ if n, err := c.Reg.PruneTimers(alive, now.Add(-onceRetention).Unix()); err != nil {
+ errs = append(errs, err)
+ } else if n > 0 {
+ log.Printf("schedule: pruned %d timer row(s)", n)
+ }
+ return errors.Join(errs...)
+}
+
+func (c *Converger) convergeApp(ctx context.Context, acct, app string, rts []registry.Route, now time.Time) error {
+ k := [2]string{acct, app}
+ raw, etag, err := c.Store.Get(ctx, DeclarationKey(acct, app))
+ var decl Declaration
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ etag = ""
+ case err != nil:
+ return fmt.Errorf("read schedules.json: %w", err)
+ case etag == c.declEtag[k]:
+ decl = c.declDoc[k]
+ default:
+ if err := jsonUnmarshal(raw, &decl); err != nil {
+ return fmt.Errorf("parse schedules.json: %w", err)
+ }
+ }
+ c.declEtag[k], c.declDoc[k] = etag, decl
+
+ // Run signal: seed silently on first sight, fire on movement.
+ runRaw, runEtag, err := c.Store.Get(ctx, RunKey(acct, app))
+ fireName := ""
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ runEtag = ""
+ case err != nil:
+ return fmt.Errorf("read schedule-run.json: %w", err)
+ }
+ if prev, seen := c.runEtag[k]; seen && runEtag != prev && runEtag != "" {
+ var sig RunSignal
+ if err := jsonUnmarshal(runRaw, &sig); err == nil {
+ if _, ok := decl.Find(sig.Name); ok {
+ fireName = sig.Name
+ } else {
+ log.Printf("schedule: %s/%s: run requested for undeclared schedule %q — ignored", acct, app, sig.Name)
+ }
+ }
+ }
+ c.runEtag[k] = runEtag
+
+ var errs []error
+ for _, rt := range rts {
+ changed, err := c.convergeHost(rt, decl, fireName, now)
+ if err != nil {
+ errs = append(errs, fmt.Errorf("%s: %w", rt.Host, err))
+ continue
+ }
+ if changed && c.Report != nil {
+ st, err := StatusFor(c.Reg, rt, c.Box, now)
+ if err != nil {
+ errs = append(errs, err)
+ continue
+ }
+ c.Report(rt, st)
+ }
+ }
+ return errors.Join(errs...)
+}
+
+// convergeHost returns whether anything about the host's rows changed (or
+// the host is unsupported and should be reported so).
+func (c *Converger) convergeHost(rt registry.Route, decl Declaration, fireName string, now time.Time) (bool, error) {
+ if rt.Backing == registry.BackingUnit {
+ // Reported once per declaration etag change via the caller's
+ // "changed" — cheap enough to always say so when there IS a
+ // declaration; nothing when there is none.
+ return len(decl.Schedules) > 0 && c.unsupportedUnreported(rt.Host), nil
+ }
+ before, err := c.Reg.TimersFor(rt.Host)
+ if err != nil {
+ return false, err
+ }
+ keep := map[string]bool{}
+ changed := false
+ for _, sc := range decl.Schedules {
+ sp, err := sc.Spec()
+ if err != nil {
+ log.Printf("schedule: %s: schedule %q has an invalid spec (%v) — skipped", rt.Host, sc.Name, err)
+ continue
+ }
+ keep[sc.Name] = true
+ var next int64
+ prev := findTimer(before, sc.Name)
+ if prev == nil || prev.Spec != sp.String() || prev.Path != sc.Path {
+ next = sp.Next(now).Unix()
+ changed = true
+ }
+ err = c.Reg.UpsertTimer(registry.Timer{Host: rt.Host, Name: sc.Name, Kind: KindSchedule, Spec: sp.String(), Path: sc.Path, NextAt: next, CreatedAt: now.Unix()})
+ if errors.Is(err, registry.ErrTimerKind) {
+ log.Printf("schedule: %s: %q is a once-timer the instance registered; the declared schedule of that name is skipped", rt.Host, sc.Name)
+ continue
+ }
+ if err != nil {
+ return changed, err
+ }
+ }
+ n, err := c.Reg.DeleteTimersExcept(rt.Host, KindSchedule, keep)
+ if err != nil {
+ return changed, err
+ }
+ if n > 0 {
+ changed = true
+ }
+ if fireName != "" && keep[fireName] {
+ if err := c.Reg.FireTimerNow(rt.Host, fireName, now.Unix()); err != nil {
+ return changed, err
+ }
+ changed = true
+ }
+ return changed, nil
+}
+
+// unsupportedUnreported tracks which unit-backed hosts have been reported
+// already so the report is written once, not every pass.
+func (c *Converger) unsupportedUnreported(host string) bool {
+ if c.reportedUnsupported == nil {
+ c.reportedUnsupported = map[string]bool{}
+ }
+ if c.reportedUnsupported[host] {
+ return false
+ }
+ c.reportedUnsupported[host] = true
+ return true
+}
+
+func findTimer(rows []registry.Timer, name string) *registry.Timer {
+ for i := range rows {
+ if rows[i].Name == name {
+ return &rows[i]
+ }
+ }
+ return nil
+}
+
+const UnsupportedUnit = "unit-backed instance — scheduled work delivers to exec-backed instances in this release"
+
+// StatusFor builds the host's report from its rows.
+func StatusFor(reg *registry.Registry, rt registry.Route, box string, now time.Time) (Status, error) {
+ st := Status{Schema: 1, Host: rt.Host, Box: box, ReportedAt: now.UTC().Format(time.RFC3339), Timers: []TimerStatus{}}
+ if rt.Backing == registry.BackingUnit {
+ st.Unsupported = UnsupportedUnit
+ return st, nil
+ }
+ rows, err := reg.TimersFor(rt.Host)
+ if err != nil {
+ return st, err
+ }
+ for _, r := range rows {
+ st.Timers = append(st.Timers, TimerStatus{Name: r.Name, Kind: r.Kind, Spec: r.Spec, Path: r.Path, NextAt: r.NextAt, LastAt: r.LastAt, LastStatus: r.LastStatus, LastMs: r.LastMs, Attempts: r.Attempts})
+ }
+ return st, nil
+}
+```
+
+Add `reportedUnsupported map[string]bool` to the struct, and `func jsonUnmarshal(b []byte, v any) error { return json.Unmarshal(b, v) }` (or call `json.Unmarshal` directly — the helper only exists to keep the import list honest). If `release.RouteAccount` does not exist under that name, use whatever `activator/logsink.go:35` calls.
+
+- [ ] **Step 4: Run tests, gate, commit**
+
+```bash
+git add internal/schedule
+git commit -m "schedule: converge bucket declarations into timers rows
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+```
+
+---
+
+### Task 4: `schedule.Runner` — delivery and the outcome policy
+
+**Files:**
+- Create: `internal/schedule/runner.go`, `internal/schedule/runner_test.go`
+
+**Interfaces:**
+- Consumes: Task 1, Task 2 (`DueTimers`, `MarkTimerFired`, `GetTimer`), `registry.Registry.Get(host)`.
+- Produces:
+
+```go
+// Deliverer sends one tick. status is the HTTP status the handler answered;
+// err is a transport-level failure (wake failed, connect refused, timeout —
+// wrap the timeout as ErrTimeout so the policy can tell it apart).
+type Deliverer interface {
+ Deliver(ctx context.Context, rt registry.Route, req *http.Request) (status int, err error)
+}
+var ErrTimeout = errors.New("schedule: delivery timed out")
+
+type Runner struct {
+ Reg *registry.Registry
+ Deliver Deliverer
+ Now func() time.Time
+ Timeout time.Duration // DeliveryTimeout
+ Token func(host string) (string, error) // the instance's CARLOS_ADMIN_TOKEN
+ Event func(rt registry.Route, level, msg string) // nil ok
+ Report func(rt registry.Route, st Status) // nil ok
+ Box string
+ Build string // User-Agent suffix
+}
+const (
+ DeliveryTimeout = 30 * time.Minute
+ TickEvery = 15 * time.Second
+ MaxAttempts = 5
+ MaxParallelHosts = 4
+ dueBatch = 64
+)
+func (r *Runner) Start(ctx context.Context) // goroutine: loop Tick every TickEvery until ctx done
+func (r *Runner) Tick(ctx context.Context) // one pass, testable
+// Outcome is the policy table (spec §6.3), pure and tested on its own.
+type Outcome struct { Status string; Retry bool }
+func Classify(status int, err error) Outcome
+func BuildRequest(rt registry.Route, t registry.Timer, token, build string) *http.Request
+```
+
+- [ ] **Step 1: Write the tests**
+
+```go
+package schedule
+
+import (
+ "context"
+ "errors"
+ "net/http"
+ "path/filepath"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+type fakeDeliverer struct {
+ mu sync.Mutex
+ status int
+ err error
+ calls []*http.Request
+ hosts []string
+}
+
+func (f *fakeDeliverer) Deliver(_ context.Context, rt registry.Route, req *http.Request) (int, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ f.calls = append(f.calls, req)
+ f.hosts = append(f.hosts, rt.Host)
+ return f.status, f.err
+}
+
+func runnerFixture(t *testing.T, d *fakeDeliverer) (*registry.Registry, *Runner) {
+ t.Helper()
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ if err := reg.Upsert(registry.Route{Host: "a.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: true}); err != nil {
+ t.Fatal(err)
+ }
+ r := &Runner{Reg: reg, Deliver: d, Timeout: time.Minute, Box: "box1", Build: "test",
+ Token: func(string) (string, error) { return "tok", nil }}
+ return reg, r
+}
+
+func TestClassify(t *testing.T) {
+ cases := []struct {
+ status int
+ err error
+ want Outcome
+ }{
+ {204, nil, Outcome{Status: "204", Retry: false}},
+ {200, nil, Outcome{Status: "200", Retry: false}},
+ {404, nil, Outcome{Status: "404 (app refused)", Retry: false}},
+ {403, nil, Outcome{Status: "403 (app refused)", Retry: false}},
+ {302, nil, Outcome{Status: "302 (app refused)", Retry: false}},
+ {500, nil, Outcome{Status: "500", Retry: true}},
+ {503, nil, Outcome{Status: "503", Retry: true}},
+ {429, nil, Outcome{Status: "429", Retry: true}},
+ {408, nil, Outcome{Status: "408", Retry: true}},
+ {0, errors.New("instance failed to wake"), Outcome{Status: "503: instance failed to wake", Retry: true}},
+ {0, ErrTimeout, Outcome{Status: "timeout after 1m0s", Retry: false}},
+ }
+ for _, c := range cases {
+ if got := classify(c.status, c.err, time.Minute); got != c.want {
+ t.Errorf("classify(%d, %v) = %+v, want %+v", c.status, c.err, got, c.want)
+ }
+ }
+}
+
+func TestBuildRequestCarriesTheContract(t *testing.T) {
+ rt := registry.Route{Host: "a.bab.oncarlos.com"}
+ tm := registry.Timer{Name: "sync", Kind: KindSchedule, Path: "/jobs/sync?x=1", NextAt: 1787004000}
+ req := BuildRequest(rt, tm, "tok", "abc")
+ if req.Method != http.MethodPost || req.Host != rt.Host || req.URL.Path != "/jobs/sync" || req.URL.RawQuery != "x=1" {
+ t.Fatalf("req = %s %s %s", req.Method, req.Host, req.URL)
+ }
+ h := req.Header
+ if h.Get("Authorization") != "Bearer tok" || h.Get("X-Carlos-Schedule") != "sync" || h.Get("X-Carlos-Schedule-Kind") != "schedule" ||
+ h.Get("X-Carlos-Schedule-At") != "1787004000" || h.Get("User-Agent") != "carlos-schedule/abc" || req.ContentLength != 0 {
+ t.Fatalf("headers = %v", h)
+ }
+}
+
+func TestTickFiresDueAdvancesScheduleAndFinishesOnce(t *testing.T) {
+ d := &fakeDeliverer{status: 204}
+ reg, r := runnerFixture(t, d)
+ now := at("2026-08-23T13:07:00Z")
+ r.Now = func() time.Time { return now }
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: now.Unix() - 5})
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 1})
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "later", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/l", NextAt: now.Unix() + 60})
+ var events []string
+ r.Event = func(_ registry.Route, level, msg string) { events = append(events, level+": "+msg) }
+ var reports int
+ r.Report = func(registry.Route, Status) { reports++ }
+ r.Tick(context.Background())
+ if len(d.calls) != 2 {
+ t.Fatalf("delivered %d, want 2 (not the future one)", len(d.calls))
+ }
+ sync, _ := reg.GetTimer("a.bab.oncarlos.com", "sync")
+ if sync.LastStatus != "204" || sync.LastAt != now.Unix() || sync.NextAt != at("2026-08-23T18:00:00Z").Unix() || sync.Attempts != 0 {
+ t.Fatalf("sync = %+v", sync)
+ }
+ once, _ := reg.GetTimer("a.bab.oncarlos.com", "remind")
+ if once.NextAt != 0 || once.LastStatus != "204" {
+ t.Fatalf("once = %+v", once)
+ }
+ if len(events) != 2 || events[0][:5] != "info:" {
+ t.Fatalf("events = %v", events)
+ }
+ if reports != 1 { // one report per host per tick, not per timer
+ t.Fatalf("reports = %d", reports)
+ }
+}
+
+func TestTickRetriesWithBackoffThenGivesUp(t *testing.T) {
+ d := &fakeDeliverer{status: 503}
+ reg, r := runnerFixture(t, d)
+ now := at("2026-08-23T13:07:00Z")
+ r.Now = func() time.Time { return now }
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: now.Unix()})
+ wantBackoff := []time.Duration{time.Minute, 2 * time.Minute, 4 * time.Minute, 8 * time.Minute}
+ for i, b := range wantBackoff {
+ r.Tick(context.Background())
+ tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync")
+ if tm.Attempts != i+1 || tm.NextAt != now.Add(b).Unix() || tm.LastStatus != "503" {
+ t.Fatalf("attempt %d: %+v (want next=+%s)", i+1, tm, b)
+ }
+ now = now.Add(b)
+ }
+ r.Tick(context.Background()) // 5th failure: give up, advance to the natural next
+ tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync")
+ if tm.Attempts != 0 || tm.LastStatus != "failed after 5 attempts: 503" || tm.NextAt != at("2026-08-23T18:00:00Z").Unix() {
+ t.Fatalf("after giving up: %+v", tm)
+ }
+}
+
+func TestBackoffNeverPassesTheNaturalNext(t *testing.T) {
+ d := &fakeDeliverer{status: 503}
+ reg, r := runnerFixture(t, d)
+ now := at("2026-08-23T13:59:30Z") // next natural 1m fire is 14:00:00, 30s away
+ r.Now = func() time.Time { return now }
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "m", Kind: KindSchedule, Spec: "every 1m0s", Path: "/m", NextAt: now.Unix()})
+ r.Tick(context.Background())
+ tm, _ := reg.GetTimer("a.bab.oncarlos.com", "m")
+ if tm.NextAt != at("2026-08-23T14:00:00Z").Unix() {
+ t.Fatalf("next = %d, want the natural next, not now+1m", tm.NextAt)
+ }
+}
+
+func TestAppRefusalDoesNotRetry(t *testing.T) {
+ d := &fakeDeliverer{status: 404}
+ reg, r := runnerFixture(t, d)
+ now := at("2026-08-23T13:07:00Z")
+ r.Now = func() time.Time { return now }
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "o", Kind: KindOnce, Spec: "at 1", Path: "/o", NextAt: 1})
+ var levels []string
+ r.Event = func(_ registry.Route, level, _ string) { levels = append(levels, level) }
+ r.Tick(context.Background())
+ tm, _ := reg.GetTimer("a.bab.oncarlos.com", "o")
+ if tm.NextAt != 0 || tm.LastStatus != "404 (app refused)" || tm.Attempts != 0 {
+ t.Fatalf("%+v", tm)
+ }
+ if len(levels) != 1 || levels[0] != "warn" {
+ t.Fatalf("levels = %v", levels)
+ }
+}
+
+func TestTimeoutCountsAsDelivered(t *testing.T) {
+ d := &fakeDeliverer{err: ErrTimeout}
+ reg, r := runnerFixture(t, d)
+ now := at("2026-08-23T13:07:00Z")
+ r.Now = func() time.Time { return now }
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/s", NextAt: now.Unix()})
+ r.Tick(context.Background())
+ tm, _ := reg.GetTimer("a.bab.oncarlos.com", "s")
+ if tm.Attempts != 0 || tm.LastStatus != "timeout after 1m0s" || tm.NextAt != at("2026-08-23T18:00:00Z").Unix() {
+ t.Fatalf("%+v", tm)
+ }
+}
+
+func TestNoTokenIsRecordedNotDelivered(t *testing.T) {
+ d := &fakeDeliverer{status: 204}
+ reg, r := runnerFixture(t, d)
+ r.Token = func(string) (string, error) { return "", errors.New("no CARLOS_ADMIN_TOKEN") }
+ now := at("2026-08-23T13:07:00Z")
+ r.Now = func() time.Time { return now }
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/s", NextAt: now.Unix()})
+ r.Tick(context.Background())
+ if len(d.calls) != 0 {
+ t.Fatal("delivered without a token")
+ }
+ tm, _ := reg.GetTimer("a.bab.oncarlos.com", "s")
+ if tm.Attempts != 1 || tm.LastStatus != "no instance token: no CARLOS_ADMIN_TOKEN" {
+ t.Fatalf("%+v", tm)
+ }
+}
+
+func TestTickSkipsRowsWhoseRouteIsGone(t *testing.T) {
+ d := &fakeDeliverer{status: 204}
+ reg, r := runnerFixture(t, d)
+ reg.UpsertTimer(registry.Timer{Host: "ghost.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/s", NextAt: 1})
+ r.Tick(context.Background())
+ if len(d.calls) != 0 {
+ t.Fatal("delivered to a host with no route")
+ }
+}
+```
+
+- [ ] **Step 2: Run to verify failure.**
+
+- [ ] **Step 3: Implement `runner.go`**
+
+```go
+package schedule
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log"
+ "net/http"
+ "os"
+ "strconv"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+const (
+ DeliveryTimeout = 30 * time.Minute
+ TickEvery = 15 * time.Second
+ MaxAttempts = 5
+ MaxParallelHosts = 4
+ dueBatch = 64
+ backoffBase = time.Minute
+ backoffCap = 15 * time.Minute
+)
+
+// ErrTimeout is what a Deliverer returns when the handler ran past the
+// delivery timeout. The policy treats it as delivered (spec §6.3): the job
+// may have finished, or still be running — a retry would double it.
+var ErrTimeout = errors.New("schedule: delivery timed out")
+
+// Deliverer sends one tick and reports the handler's status, or a
+// transport-level error (wake failed, connect refused, ErrTimeout).
+type Deliverer interface {
+ Deliver(ctx context.Context, rt registry.Route, req *http.Request) (int, error)
+}
+
+// Runner fires due timers: at-least-once, serialized per host, bounded
+// parallelism across hosts (spec §6.3). Every seam is a field so a tick is
+// testable with no goroutines, sockets or clocks.
+type Runner struct {
+ Reg *registry.Registry
+ Deliver Deliverer
+ Now func() time.Time
+ Timeout time.Duration
+ Token func(host string) (string, error)
+ Event func(rt registry.Route, level, msg string)
+ Report func(rt registry.Route, st Status)
+ Box string
+ Build string
+}
+
+// Start runs Tick every TickEvery until ctx ends.
+func (r *Runner) Start(ctx context.Context) {
+ go func() {
+ t := time.NewTicker(TickEvery)
+ defer t.Stop()
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-t.C:
+ r.Tick(ctx)
+ }
+ }
+ }()
+}
+
+func (r *Runner) now() time.Time {
+ if r.Now != nil {
+ return r.Now()
+ }
+ return time.Now()
+}
+
+func (r *Runner) timeout() time.Duration {
+ if r.Timeout > 0 {
+ return r.Timeout
+ }
+ return DeliveryTimeout
+}
+
+// Tick is one pass: everything due, grouped by host, hosts in parallel up to
+// MaxParallelHosts, timers within a host one after another.
+func (r *Runner) Tick(ctx context.Context) {
+ due, err := r.Reg.DueTimers(r.now().Unix(), dueBatch)
+ if err != nil {
+ log.Printf("schedule: due query: %v", err)
+ return
+ }
+ if len(due) == 0 {
+ return
+ }
+ byHost := map[string][]registry.Timer{}
+ var order []string
+ for _, t := range due {
+ if _, seen := byHost[t.Host]; !seen {
+ order = append(order, t.Host)
+ }
+ byHost[t.Host] = append(byHost[t.Host], t)
+ }
+ sem := make(chan struct{}, MaxParallelHosts)
+ var wg sync.WaitGroup
+ for _, host := range order {
+ rt, err := r.Reg.Get(host)
+ if err != nil || rt == nil {
+ // Converge prunes these; until it does, they are not ours to fire.
+ continue
+ }
+ wg.Add(1)
+ sem <- struct{}{}
+ go func(rt registry.Route, timers []registry.Timer) {
+ defer wg.Done()
+ defer func() { <-sem }()
+ for _, t := range timers {
+ r.fire(ctx, rt, t)
+ }
+ if r.Report != nil {
+ if st, err := StatusFor(r.Reg, rt, r.Box, r.now()); err == nil {
+ r.Report(rt, st)
+ }
+ }
+ }(*rt, byHost[host])
+ }
+ wg.Wait()
+}
+
+func (r *Runner) fire(ctx context.Context, rt registry.Route, t registry.Timer) {
+ sp, err := ParseSpec(t.Spec)
+ if err != nil {
+ log.Printf("schedule: %s/%s: bad spec %q: %v — dropping the row", rt.Host, t.Name, t.Spec, err)
+ r.Reg.DeleteTimer(rt.Host, t.Name)
+ return
+ }
+ start := r.now()
+ token, err := r.Token(rt.Host)
+ if err != nil {
+ r.record(rt, t, sp, start, Outcome{Status: "no instance token: " + err.Error(), Retry: true}, 0)
+ return
+ }
+ dctx, cancel := context.WithTimeout(ctx, r.timeout())
+ req := BuildRequest(rt, t, token, r.Build).WithContext(dctx)
+ status, derr := r.Deliver.Deliver(dctx, rt, req)
+ cancel()
+ if errors.Is(derr, context.DeadlineExceeded) {
+ derr = ErrTimeout
+ }
+ r.record(rt, t, sp, start, classify(status, derr, r.timeout()), r.now().Sub(start))
+}
+
+// record applies the policy table to one outcome and writes the row, the
+// member-facing event, and nothing else.
+func (r *Runner) record(rt registry.Route, t registry.Timer, sp Spec, at time.Time, o Outcome, took time.Duration) {
+ attempts := 0
+ status := o.Status
+ natural := sp.Next(at) // zero for a finished once
+ next := natural.Unix()
+ if natural.IsZero() {
+ next = 0
+ }
+ level := "info"
+ if o.Retry {
+ attempts = t.Attempts + 1
+ if attempts >= MaxAttempts {
+ status = fmt.Sprintf("failed after %d attempts: %s", MaxAttempts, o.Status)
+ attempts = 0
+ level = "warn"
+ } else {
+ backoff := backoffBase << (attempts - 1)
+ if backoff > backoffCap {
+ backoff = backoffCap
+ }
+ retryAt := at.Add(backoff)
+ if !natural.IsZero() && retryAt.After(natural) {
+ retryAt = natural
+ }
+ next = retryAt.Unix()
+ level = "warn"
+ }
+ } else if !isSuccess(o.Status) {
+ level = "warn"
+ }
+ if err := r.Reg.MarkTimerFired(rt.Host, t.Name, at.Unix(), status, took.Milliseconds(), attempts, next); err != nil {
+ log.Printf("schedule: %s/%s: record fire: %v", rt.Host, t.Name, err)
+ }
+ if r.Event != nil {
+ msg := fmt.Sprintf("schedule %s fired POST %s → %s in %s", t.Name, t.Path, status, took.Truncate(time.Millisecond))
+ if attempts > 0 {
+ msg += fmt.Sprintf(" (attempt %d of %d; retrying at %s)", attempts, MaxAttempts, time.Unix(next, 0).UTC().Format(time.RFC3339))
+ }
+ r.Event(rt, level, msg)
+ }
+}
+
+func isSuccess(status string) bool { return len(status) == 3 && status[0] == '2' }
+
+// Outcome is one line of the policy table (spec §6.3).
+type Outcome struct {
+ Status string
+ Retry bool
+}
+
+func classify(status int, err error, timeout time.Duration) Outcome {
+ switch {
+ case errors.Is(err, ErrTimeout):
+ return Outcome{Status: "timeout after " + timeout.String()}
+ case err != nil:
+ return Outcome{Status: "503: " + err.Error(), Retry: true}
+ case status >= 200 && status < 300:
+ return Outcome{Status: strconv.Itoa(status)}
+ case status >= 500, status == 408, status == 429:
+ return Outcome{Status: strconv.Itoa(status), Retry: true}
+ default:
+ return Outcome{Status: strconv.Itoa(status) + " (app refused)"}
+ }
+}
+
+// Classify is the exported policy for the host package's tests.
+func Classify(status int, err error) Outcome { return classify(status, err, DeliveryTimeout) }
+
+// BuildRequest is the tick (spec §6.3's header table). Content-Length 0 and
+// no body: a tick carries no data, by design.
+func BuildRequest(rt registry.Route, t registry.Timer, token, build string) *http.Request {
+ req, err := http.NewRequest(http.MethodPost, "http://"+rt.Host+t.Path, http.NoBody)
+ if err != nil {
+ // Path was validated at declaration; a failure here is a bug, not
+ // a runtime condition. Fall back to the root so the app still hears.
+ req, _ = http.NewRequest(http.MethodPost, "http://"+rt.Host+"/", http.NoBody)
+ }
+ req.Host = rt.Host
+ req.ContentLength = 0
+ req.Header.Set("Authorization", "Bearer "+token)
+ req.Header.Set("X-Carlos-Schedule", t.Name)
+ req.Header.Set("X-Carlos-Schedule-Kind", t.Kind)
+ req.Header.Set("X-Carlos-Schedule-At", strconv.FormatInt(t.NextAt, 10))
+ req.Header.Set("User-Agent", "carlos-schedule/"+build)
+ return req
+}
+
+// TokenFromEnvFile is the production Token: CARLOS_ADMIN_TOKEN from
+// <dataDir>/<host>.env, the file the activator owns (spec §6.4).
+func TokenFromEnvFile(dataDir string) func(host string) (string, error) {
+ return func(host string) (string, error) {
+ env, err := readEnvFile(dataDir + "/" + host + ".env")
+ if err != nil {
+ return "", err
+ }
+ if env["CARLOS_ADMIN_TOKEN"] == "" {
+ return "", errors.New("no CARLOS_ADMIN_TOKEN")
+ }
+ return env["CARLOS_ADMIN_TOKEN"], nil
+ }
+}
+```
+
+`readEnvFile`: copy the small parser from `activator/backend_exec.go:827` into `internal/schedule/envfile.go` (KEY=VALUE lines, `#` comments, optional surrounding quotes — match that function's behaviour exactly; it is ~20 lines) rather than importing activator (cycle). Note the `X-Carlos-Schedule-At` header carries `t.NextAt` — the occurrence that was due — which is why `fire` reads it before `record` rewrites the row.
+
+- [ ] **Step 4: Run tests, gate, commit**
+
+```bash
+git add internal/schedule
+git commit -m "schedule: runner — due ticks, outcome policy, backoff, events
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+```
+
+---
+
+### Task 5: `schedule.ControlHandler` — the instance's once-timer verbs
+
+**Files:**
+- Create: `internal/schedule/control.go`, `internal/schedule/control_test.go`
+
+**Interfaces:**
+- Consumes: Task 1, Task 2 (`UpsertTimer`, `DeleteTimer`, `TimersFor`, `CountTimers`, `GetTimer`), `registry.ErrTimerKind`.
+- Produces:
+
+```go
+// TimerStore is the registry subset the control handler needs.
+type TimerStore interface {
+ UpsertTimer(registry.Timer) error
+ DeleteTimer(host, name string) error
+ GetTimer(host, name string) (*registry.Timer, error)
+ TimersFor(host string) ([]registry.Timer, error)
+ CountTimers(host, kind string) (int, error)
+}
+// Mount registers PUT /schedule/{name}, DELETE /schedule/{name}, GET /schedule on mux.
+// host is fixed at construction and never read from a request. token reads the
+// host's CARLOS_ADMIN_TOKEN on every request (the file may not exist yet at listener creation).
+func Mount(mux *http.ServeMux, host string, store TimerStore, token func() (string, error), now func() time.Time)
+```
+
+Request/response shapes:
+
+```
+PUT /schedule/{name} body {"at":"2026-09-01T08:00:00Z","path":"/jobs/remind"}
+ 200 {"name":..,"kind":"once","spec":"at N","path":..,"next_at":N,...} (TimerStatus)
+ 400 bad name/path/at; 401 missing/wrong bearer; 409 name is a declared schedule; 429 over MaxOncePerHost
+DELETE /schedule/{name} 204 always (401 on bad auth)
+GET /schedule 200 {"timers":[TimerStatus...]}
+```
+
+- [ ] **Step 1: Write the tests**
+
+```go
+package schedule
+
+import (
+ "encoding/json"
+ "net/http"
+ "net/http/httptest"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+func controlFixture(t *testing.T) (*registry.Registry, *http.ServeMux) {
+ t.Helper()
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ mux := http.NewServeMux()
+ Mount(mux, "a.bab.oncarlos.com", reg, func() (string, error) { return "tok", nil }, func() time.Time { return at("2026-08-23T13:07:00Z") })
+ return reg, mux
+}
+
+func do(mux *http.ServeMux, method, path, auth, body string) *httptest.ResponseRecorder {
+ req := httptest.NewRequest(method, path, strings.NewReader(body))
+ if auth != "" {
+ req.Header.Set("Authorization", auth)
+ }
+ rec := httptest.NewRecorder()
+ mux.ServeHTTP(rec, req)
+ return rec
+}
+
+func TestControlRequiresTheHostsToken(t *testing.T) {
+ _, mux := controlFixture(t)
+ for _, auth := range []string{"", "Bearer wrong", "Basic tok"} {
+ if rec := do(mux, "PUT", "/schedule/r", auth, `{"at":"2026-09-01T08:00:00Z","path":"/r"}`); rec.Code != http.StatusUnauthorized {
+ t.Errorf("auth %q → %d, want 401", auth, rec.Code)
+ }
+ if rec := do(mux, "GET", "/schedule", auth, ""); rec.Code != http.StatusUnauthorized {
+ t.Errorf("GET auth %q → %d", auth, rec.Code)
+ }
+ }
+}
+
+func TestControlPutGetDelete(t *testing.T) {
+ reg, mux := controlFixture(t)
+ rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", `{"at":"2026-09-01T08:00:00Z","path":"/jobs/remind"}`)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("put = %d %s", rec.Code, rec.Body)
+ }
+ var ts TimerStatus
+ json.Unmarshal(rec.Body.Bytes(), &ts)
+ if ts.Kind != KindOnce || ts.NextAt != at("2026-09-01T08:00:00Z").Unix() || ts.Path != "/jobs/remind" || ts.Spec != "at "+itoa(ts.NextAt) {
+ t.Fatalf("put body = %+v", ts)
+ }
+ tm, _ := reg.GetTimer("a.bab.oncarlos.com", "remind")
+ if tm == nil || tm.Kind != KindOnce {
+ t.Fatalf("row = %+v", tm)
+ }
+ // Upsert by name moves the time.
+ do(mux, "PUT", "/schedule/remind", "Bearer tok", `{"at":"2026-09-02T08:00:00Z","path":"/jobs/remind"}`)
+ tm, _ = reg.GetTimer("a.bab.oncarlos.com", "remind")
+ if tm.NextAt != at("2026-09-02T08:00:00Z").Unix() {
+ t.Fatalf("upsert did not move: %+v", tm)
+ }
+ rec = do(mux, "GET", "/schedule", "Bearer tok", "")
+ var list struct{ Timers []TimerStatus }
+ json.Unmarshal(rec.Body.Bytes(), &list)
+ if len(list.Timers) != 1 || list.Timers[0].Name != "remind" {
+ t.Fatalf("list = %+v", list)
+ }
+ if rec := do(mux, "DELETE", "/schedule/remind", "Bearer tok", ""); rec.Code != http.StatusNoContent {
+ t.Fatalf("delete = %d", rec.Code)
+ }
+ if rec := do(mux, "DELETE", "/schedule/remind", "Bearer tok", ""); rec.Code != http.StatusNoContent {
+ t.Fatalf("second delete = %d, want 204 (idempotent)", rec.Code)
+ }
+}
+
+func TestControlRefusals(t *testing.T) {
+ reg, mux := controlFixture(t)
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/s", NextAt: 1})
+ cases := []struct {
+ name, body string
+ want int
+ }{
+ {"sync", `{"at":"2026-09-01T08:00:00Z","path":"/p"}`, http.StatusConflict},
+ {"Bad", `{"at":"2026-09-01T08:00:00Z","path":"/p"}`, http.StatusBadRequest},
+ {"ok", `{"at":"2026-09-01T08:00:00Z","path":"p"}`, http.StatusBadRequest},
+ {"ok", `{"at":"soon","path":"/p"}`, http.StatusBadRequest},
+ {"ok", `{"at":"2028-09-01T08:00:00Z","path":"/p"}`, http.StatusBadRequest}, // > 400 days ahead
+ {"ok", `not json`, http.StatusBadRequest},
+ }
+ for _, c := range cases {
+ if rec := do(mux, "PUT", "/schedule/"+c.name, "Bearer tok", c.body); rec.Code != c.want {
+ t.Errorf("PUT %s %s → %d, want %d (%s)", c.name, c.body, rec.Code, c.want, rec.Body)
+ }
+ }
+ // A past `at` is accepted and fires on the next tick.
+ if rec := do(mux, "PUT", "/schedule/past", "Bearer tok", `{"at":"2020-01-01T00:00:00Z","path":"/p"}`); rec.Code != http.StatusOK {
+ t.Fatalf("past at → %d", rec.Code)
+ }
+ if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "past"); tm.NextAt != at("2020-01-01T00:00:00Z").Unix() {
+ t.Fatalf("past row = %+v", tm)
+ }
+}
+
+func TestControlCapsOncePerHost(t *testing.T) {
+ reg, mux := controlFixture(t)
+ for i := 0; i < MaxOncePerHost; i++ {
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "n" + itoa(int64(i)), Kind: KindOnce, Spec: "at 1", Path: "/p", NextAt: 1})
+ }
+ if rec := do(mux, "PUT", "/schedule/over", "Bearer tok", `{"at":"2026-09-01T08:00:00Z","path":"/p"}`); rec.Code != http.StatusTooManyRequests {
+ t.Fatalf("over cap → %d", rec.Code)
+ }
+ // Replacing an existing name is not a new row and must still work.
+ if rec := do(mux, "PUT", "/schedule/n1", "Bearer tok", `{"at":"2026-09-01T08:00:00Z","path":"/p"}`); rec.Code != http.StatusOK {
+ t.Fatalf("replace at cap → %d", rec.Code)
+ }
+}
+
+func itoa(n int64) string { return strconv.FormatInt(n, 10) }
+```
+
+(add `"strconv"` to the imports.)
+
+- [ ] **Step 2: Run to verify failure.**
+
+- [ ] **Step 3: Implement `control.go`**
+
+```go
+package schedule
+
+import (
+ "crypto/sha256"
+ "crypto/subtle"
+ "encoding/json"
+ "errors"
+ "net/http"
+ "os"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+const MaxOncePerHost = 1000
+
+// TimerStore is the registry subset the control channel needs.
+type TimerStore interface {
+ UpsertTimer(registry.Timer) error
+ DeleteTimer(host, name string) error
+ GetTimer(host, name string) (*registry.Timer, error)
+ TimersFor(host string) ([]registry.Timer, error)
+ CountTimers(host, kind string) (int, error)
+}
+
+// Mount adds the once-timer verbs to an instance's control socket (spec
+// §7.2). host is fixed here and never read from a request — the
+// confused-deputy guarantee edge.ListenPurge gives — and every verb needs
+// the host's own CARLOS_ADMIN_TOKEN as a bearer, which is what closes the
+// same-uid-sibling gap purge.go documents: a sibling can compute this
+// socket's path, but not this instance's token.
+func Mount(mux *http.ServeMux, host string, store TimerStore, token func() (string, error), now func() time.Time) {
+ authed := func(fn http.HandlerFunc) http.HandlerFunc {
+ return func(w http.ResponseWriter, r *http.Request) {
+ want, err := token()
+ if err != nil || want == "" {
+ http.Error(w, "unauthorized", http.StatusUnauthorized)
+ return
+ }
+ got := sha256.Sum256([]byte(r.Header.Get("Authorization")))
+ exp := sha256.Sum256([]byte("Bearer " + want))
+ if subtle.ConstantTimeCompare(got[:], exp[:]) != 1 {
+ http.Error(w, "unauthorized", http.StatusUnauthorized)
+ return
+ }
+ fn(w, r)
+ }
+ }
+
+ mux.HandleFunc("PUT /schedule/{name}", authed(func(w http.ResponseWriter, r *http.Request) {
+ name := r.PathValue("name")
+ if !ValidName(name) {
+ http.Error(w, "name: want [a-z0-9][a-z0-9-]{0,31}", http.StatusBadRequest)
+ return
+ }
+ var body struct {
+ At string `json:"at"`
+ Path string `json:"path"`
+ }
+ if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4096)).Decode(&body); err != nil {
+ http.Error(w, "bad request: "+err.Error(), http.StatusBadRequest)
+ return
+ }
+ atT, err := time.Parse(time.RFC3339, body.At)
+ if err != nil {
+ http.Error(w, "at: want RFC3339, e.g. 2026-09-01T08:00:00Z", http.StatusBadRequest)
+ return
+ }
+ if atT.After(now().Add(MaxOnceAhead)) {
+ http.Error(w, "at: more than 400 days ahead", http.StatusBadRequest)
+ return
+ }
+ if err := ValidPath(body.Path); err != nil {
+ http.Error(w, err.Error(), http.StatusBadRequest)
+ return
+ }
+ existing, err := store.GetTimer(host, name)
+ switch {
+ case err == nil && existing.Kind != KindOnce:
+ http.Error(w, "name is a declared schedule", http.StatusConflict)
+ return
+ case err != nil && !errors.Is(err, os.ErrNotExist):
+ http.Error(w, "registry error", http.StatusInternalServerError)
+ return
+ case err != nil:
+ n, cerr := store.CountTimers(host, KindOnce)
+ if cerr != nil {
+ http.Error(w, "registry error", http.StatusInternalServerError)
+ return
+ }
+ if n >= MaxOncePerHost {
+ http.Error(w, "too many one-shot timers (limit 1000) — delete finished ones or use a schedule", http.StatusTooManyRequests)
+ return
+ }
+ }
+ sp := Spec{Kind: KindOnce, At: atT.UTC()}
+ t := registry.Timer{Host: host, Name: name, Kind: KindOnce, Spec: sp.String(), Path: body.Path, NextAt: atT.Unix(), CreatedAt: now().Unix()}
+ if err := store.UpsertTimer(t); err != nil {
+ http.Error(w, "registry error", http.StatusInternalServerError)
+ return
+ }
+ row, err := store.GetTimer(host, name)
+ if err != nil {
+ http.Error(w, "registry error", http.StatusInternalServerError)
+ return
+ }
+ writeTimer(w, *row)
+ }))
+
+ mux.HandleFunc("DELETE /schedule/{name}", authed(func(w http.ResponseWriter, r *http.Request) {
+ name := r.PathValue("name")
+ if !ValidName(name) {
+ http.Error(w, "bad name", http.StatusBadRequest)
+ return
+ }
+ if row, err := store.GetTimer(host, name); err == nil && row.Kind != KindOnce {
+ http.Error(w, "name is a declared schedule — remove it with carlos schedule rm", http.StatusConflict)
+ return
+ }
+ if err := store.DeleteTimer(host, name); err != nil {
+ http.Error(w, "registry error", http.StatusInternalServerError)
+ return
+ }
+ w.WriteHeader(http.StatusNoContent)
+ }))
+
+ mux.HandleFunc("GET /schedule", authed(func(w http.ResponseWriter, r *http.Request) {
+ rows, err := store.TimersFor(host)
+ if err != nil {
+ http.Error(w, "registry error", http.StatusInternalServerError)
+ return
+ }
+ out := struct {
+ Timers []TimerStatus `json:"timers"`
+ }{Timers: []TimerStatus{}}
+ for _, r := range rows {
+ out.Timers = append(out.Timers, toStatus(r))
+ }
+ w.Header().Set("Content-Type", "application/json")
+ json.NewEncoder(w).Encode(out)
+ }))
+}
+
+func toStatus(r registry.Timer) TimerStatus {
+ return TimerStatus{Name: r.Name, Kind: r.Kind, Spec: r.Spec, Path: r.Path, NextAt: r.NextAt, LastAt: r.LastAt, LastStatus: r.LastStatus, LastMs: r.LastMs, Attempts: r.Attempts}
+}
+
+func writeTimer(w http.ResponseWriter, r registry.Timer) {
+ w.Header().Set("Content-Type", "application/json")
+ json.NewEncoder(w).Encode(toStatus(r))
+}
+```
+
+Refactor `StatusFor` (Task 3) to use `toStatus`.
+
+- [ ] **Step 4: Run tests, gate, commit**
+
+```bash
+git add internal/schedule
+git commit -m "schedule: control-socket verbs for once-timers, bearer-gated
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+```
+
+---
+
+### Task 6: Activator — agent-minted admin token, always-on control listener
+
+**Files:**
+- Modify: `internal/edge/purge.go` — `ListenPurge(socketPath, host, cache)` becomes `ListenControl(socketPath, host string, cache *Cache, mount func(*http.ServeMux)) (*ControlListener, error)`; keep `type PurgeListener = ControlListener` and a `ListenPurge` wrapper (`ListenControl(path, host, cache, nil)`) so existing tests compile. When `cache == nil`, do not register `POST /purge`. Rename the doc comment's subject accordingly; keep every word of the security reasoning.
+- Modify: `internal/edge/edge.go` — add `func UnixProxy(socketPath string) http.Handler { return unixProxy(socketPath) }` with a one-line doc: exported for the schedule runner (host package), which delivers ticks to a route's socket through the activator without the edge's cache/compression/usage layers.
+- Modify: `internal/activator/backend_exec.go`:
+ 1. New `ensureAdminToken(path string) error` in a new file `internal/activator/admintoken.go`: read the env file (tolerate absent), return nil if `CARLOS_ADMIN_TOKEN` is set and non-empty; otherwise mint 32 random bytes → hex, append `CARLOS_ADMIN_TOKEN=<hex>\n` (create with `0600` if absent; open `O_APPEND|O_WRONLY|O_CREATE`). Called from `Start` (the spawn path, around line 495, BEFORE `instanceEnv` is read so the spawned child sees it) and from `Seed`. A mint failure fails the wake with a clear error — an instance without its token cannot receive ticks or answer next-due, and a wake that silently skips it would be the lie §110 was about.
+ 2. Control listener: the `if b.Cache != nil { … ListenPurge … }` block at ~575 becomes unconditional: `edge.ListenControl(controlSocket, rt.Host, b.Cache, func(mux *http.ServeMux) { schedule.Mount(mux, rt.Host, b.Timers, tokenReader, time.Now) })` where `b.Timers schedule.TimerStore` is a new `ExecBackend` field (nil-tolerant: when nil, pass a `mount` of nil — no schedule verbs) and `tokenReader` reads `CARLOS_ADMIN_TOKEN` from `<DataDir>/<host>.env` on each call (reuse `readEnvFile`). The env-var block at ~529: `CARLOS_CONTROL_SOCKET` is set whenever the listener was created; `CARLOS_PURGE=header` still only when `b.Cache != nil`. Update the comment block there — it argues the "only when the channel exists" rule, and the rule still holds: the channel now always exists for exec children.
+- Test: `internal/activator/admintoken_test.go` (mint creates 0600 file with one line; existing token untouched; existing file without the var gets the line appended and other vars preserved), and extend an existing `backend_exec_test.go` spawn test (find the one that inspects the child's env — search for `CARLOS_CONTROL_SOCKET` in `_test.go`) to assert the var is now set with a nil cache and that `CARLOS_PURGE` is not.
+
+**Interfaces:**
+- Consumes: Task 5 `schedule.Mount`, `schedule.TimerStore`.
+- Produces: `ExecBackend.Timers schedule.TimerStore` (host wiring sets it to the registry); `edge.UnixProxy`; `edge.ListenControl`.
+
+- [ ] **Step 1: Write `admintoken_test.go`**
+
+```go
+package activator
+
+import (
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+)
+
+func TestEnsureAdminTokenMintsOnce(t *testing.T) {
+ p := filepath.Join(t.TempDir(), "h.env")
+ if err := ensureAdminToken(p); err != nil {
+ t.Fatal(err)
+ }
+ st, _ := os.Stat(p)
+ if st.Mode().Perm() != 0o600 {
+ t.Fatalf("mode = %o", st.Mode().Perm())
+ }
+ env, _ := readEnvFile(p)
+ tok := env["CARLOS_ADMIN_TOKEN"]
+ if len(tok) != 64 {
+ t.Fatalf("token = %q", tok)
+ }
+ if err := ensureAdminToken(p); err != nil {
+ t.Fatal(err)
+ }
+ env, _ = readEnvFile(p)
+ if env["CARLOS_ADMIN_TOKEN"] != tok {
+ t.Fatal("token rotated on second call")
+ }
+}
+
+func TestEnsureAdminTokenPreservesOtherVars(t *testing.T) {
+ p := filepath.Join(t.TempDir(), "h.env")
+ os.WriteFile(p, []byte("DATABASE_URL=sqlite\n# comment\n"), 0o600)
+ if err := ensureAdminToken(p); err != nil {
+ t.Fatal(err)
+ }
+ raw, _ := os.ReadFile(p)
+ if !strings.HasPrefix(string(raw), "DATABASE_URL=sqlite\n# comment\n") || !strings.Contains(string(raw), "CARLOS_ADMIN_TOKEN=") {
+ t.Fatalf("file = %q", raw)
+ }
+ env, _ := readEnvFile(p)
+ if env["DATABASE_URL"] != "sqlite" {
+ t.Fatal("other var lost")
+ }
+}
+```
+
+- [ ] **Step 2: Implement `admintoken.go`**
+
+```go
+package activator
+
+// The instance secret. CARLOS_ADMIN_TOKEN is what the agent presents to an
+// instance for the things only the platform may ask (GET /api/next-due, and
+// now the scheduled-work tick — spec 2026-08-23-scheduled-work-design.md
+// §6.4) and what the instance presents back on its control socket. Until
+// this file nothing minted it: Eleven's tooling wrote it, console-created
+// instances never had one, and the next-due probe failed silently for all
+// of them. Minted here, once, on the first spawn that finds it missing;
+// never rotated by the platform.
+
+import (
+ "crypto/rand"
+ "encoding/hex"
+ "fmt"
+ "os"
+)
+
+func ensureAdminToken(path string) error {
+ env, err := readEnvFile(path)
+ if err != nil && !os.IsNotExist(err) {
+ return fmt.Errorf("read %s: %w", path, err)
+ }
+ if env["CARLOS_ADMIN_TOKEN"] != "" {
+ return nil
+ }
+ raw := make([]byte, 32)
+ if _, err := rand.Read(raw); err != nil {
+ return err
+ }
+ f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0o600)
+ if err != nil {
+ return fmt.Errorf("open %s: %w", path, err)
+ }
+ defer f.Close()
+ prefix := ""
+ if st, err := f.Stat(); err == nil && st.Size() > 0 {
+ // Land on a fresh line even if the file lacks a trailing newline.
+ buf := make([]byte, 1)
+ if rf, err := os.Open(path); err == nil {
+ rf.Seek(-1, 2)
+ rf.Read(buf)
+ rf.Close()
+ }
+ if buf[0] != '\n' {
+ prefix = "\n"
+ }
+ }
+ _, err = fmt.Fprintf(f, "%sCARLOS_ADMIN_TOKEN=%s\n", prefix, hex.EncodeToString(raw))
+ return err
+}
+```
+
+- [ ] **Step 3: Wire into `Start`/`Seed`, generalize the listener, export `UnixProxy`.** Read the surrounding comments before editing and update them where they now say something false (the "only when there's a purge listener" sentence). Keep the `else` branch that strips `CARLOS_PURGE` when there is no cache.
+
+- [ ] **Step 4: Gate**
+
+Run: `go build ./... && go test ./internal/activator/ ./internal/edge/ -count=1 && go vet ./internal/activator/ ./internal/edge/ && gofmt -l internal/activator internal/edge`
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add internal/activator internal/edge
+git commit -m "activator: mint CARLOS_ADMIN_TOKEN; control socket always on, schedule verbs mounted
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+```
+
+---
+
+### Task 7: Host wiring — converge leg, runner, deliverer, reporter
+
+**Files:**
+- Create: `internal/host/schedule_wiring.go`, `internal/host/schedule_wiring_test.go`
+- Modify: `internal/host/agent.go` — in the agent's startup (where `svc` is built, ~line 263: set `eb.Timers = reg`); in `fastConvergeLoop` (~1168: seed; ~1196: add a `scheduleLeg` after `restartLeg` — add a `scheduleLeg func(ctx)` field to `fastDeps` and call it in `fastTick` right after `d.restartLeg`); in `adoptLoop` (after the hourly `release.Pass`, call the converger's `Pass`).
+
+**Interfaces:**
+- Consumes: Task 3 `NewConverger`, Task 4 `Runner`/`Deliverer`/`TokenFromEnvFile`, Task 6 `edge.UnixProxy`, `activator.Service.ServeHTTP`, `activator.EventSink` (the `logSink` the agent already builds), `schedule.WriteStatus`.
+- Produces:
+
+```go
+// scheduleDeps is built once in the agent and shared by the seed, the fast leg, the hourly leg and the runner.
+type scheduleDeps struct {
+ conv *schedule.Converger
+ runner *schedule.Runner
+}
+func newScheduleDeps(reg *registry.Registry, store release.Store, svc *activator.Service, sink activator.EventSink, dd string) *scheduleDeps
+// activatorDeliverer sends a tick through svc.ServeHTTP with edge.UnixProxy(rt.Socket);
+// when ServeHTTP returns false (not hibernate-managed) it serves the proxy directly.
+type activatorDeliverer struct{ svc *activator.Service }
+func (d activatorDeliverer) Deliver(ctx context.Context, rt registry.Route, req *http.Request) (int, error)
+// statusRecorder is the ResponseWriter that captures the status and discards the body.
+```
+
+Deliver semantics: a recorded status of 503 whose body starts with "instance" (the activator's own wake refusals: "instance is waking", "instance failed to wake", "instance is waking elsewhere") is returned as `(0, errors.New(<body first line>))` so the policy reads it as a wake failure rather than an app 503 — both retry, but the member-facing status says which. Any other status is returned as-is. A context deadline is returned as `ctx.Err()` (the runner maps it to `ErrTimeout`).
+
+Reporter: `func(rt registry.Route, st schedule.Status) { ctx 10s; schedule.WriteStatus(ctx, store, release.RouteAccount(rt), rt.App, st); log on error }`. Box name: `os.Hostname()`.
+
+Event: `func(rt, level, msg) { sink.Event(release.RouteAccount(rt), rt.App, rt.Host, level, msg) }` — nil sink → nil.
+
+- [ ] **Step 1: Write `schedule_wiring_test.go`** — test `activatorDeliverer` with a nil-activator fallback path is hard; instead test `statusRecorder` and the wake-refusal mapping with a fake `http.Handler`:
+
+```go
+package host
+
+import (
+ "context"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+func TestDeliverViaHandlerMapsWakeRefusals(t *testing.T) {
+ cases := []struct {
+ status int
+ body string
+ want int
+ wantErr string
+ }{
+ {204, "", 204, ""},
+ {503, "instance failed to wake\n", 0, "instance failed to wake"},
+ {503, "instance is waking; retry shortly\n", 0, "instance is waking; retry shortly"},
+ {503, "app says no", 503, ""},
+ }
+ for _, c := range cases {
+ h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ w.WriteHeader(c.status)
+ w.Write([]byte(c.body))
+ })
+ req := httptest.NewRequest("POST", "http://a.bab.oncarlos.com/jobs/x", nil)
+ got, err := deliverVia(context.Background(), h, registry.Route{Host: "a.bab.oncarlos.com"}, req)
+ if got != c.want || (err == nil) != (c.wantErr == "") || (err != nil && !strings.Contains(err.Error(), c.wantErr)) {
+ t.Errorf("%d %q → %d %v", c.status, c.body, got, err)
+ }
+ }
+}
+```
+
+Implement `deliverVia(ctx, h http.Handler, rt, req) (int, error)` as the pure core, and `activatorDeliverer.Deliver` as: `proxy := edge.UnixProxy(rt.Socket)`; if `d.svc != nil` and `d.svc.ServeHTTP(rec, req, rt, proxy)` returns true → use rec's outcome; else `deliverVia(ctx, proxy, rt, req)`. To make `svc.ServeHTTP` write into the same recorder, structure `deliverVia` as `handled := func(w http.ResponseWriter, r *http.Request) bool` — simplest: `deliverVia` takes an `http.Handler`; build for the activator case a handler `http.HandlerFunc(func(w, r) { if !d.svc.ServeHTTP(w, r, rt, proxy) { proxy.ServeHTTP(w, r) } })`.
+
+- [ ] **Step 2: Implement `schedule_wiring.go`** with those pieces plus `newScheduleDeps` and the three call sites in `agent.go`. Log one line at arming: `agent: scheduled work armed (converge on serial + hourly; runner every 15s)`. The runner's `Start` takes the agent's background context (`context.Background()` — the agent runs until the process ends, like the other loops).
+
+- [ ] **Step 3: Gate**
+
+Run: `go build ./... && go test ./internal/host/ -count=1 && go vet ./internal/host/ && gofmt -l internal/host && GOOS=darwin go build .`
+
+The darwin build matters here: `internal/host` is linked into the CLI. Nothing in this task may use a Linux-only type.
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add internal/host
+git commit -m "host: wire scheduled work — converge leg, runner, activator-backed delivery, status reports
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+```
+
+---
+
+### Task 8: Console doors
+
+**Files:**
+- Create: `internal/console/apischedule.go`, `internal/console/apischedule_test.go`
+- Modify: `internal/console/authcli.go` (register four doors next to `restart`, all `svcrole.Allow(svcrole.Operate)`), `internal/console/svcroutes_test.go` (if there is an Operate-door or "every door named" list that enumerates patterns, add the four; `closedDoors` is NOT touched — these doors are open to Operate).
+
+**Interfaces:**
+- Consumes: Task 1 (`schedule.ReadDeclaration`, `UpsertSchedule`, `RemoveSchedule`, `TouchRun`, `ListStatus`, `Schedule`, `Declaration`, `Status`), `s.apiAccountApp`, `s.allowWrite`, `s.audit`, `s.bumpSerial`, `readJSON`, `writeJSON`, `jsonError`, `s.cfg.Releases`.
+- Produces (the CLI in Task 9 decodes exactly this):
+
+```go
+type apiScheduleList struct {
+ Account string `json:"account"`
+ App string `json:"app"`
+ Schedules []schedule.Schedule `json:"schedules"`
+ Instances []schedule.Status `json:"instances"`
+ UpdatedAt string `json:"updated_at"`
+}
+```
+
+Doors:
+
+| Pattern | Handler | Body in | Out |
+|---|---|---|---|
+| `GET /api/cli/apps/{account}/{app}/schedules` | `handleAPIScheduleList` | — | 200 `apiScheduleList` |
+| `POST /api/cli/apps/{account}/{app}/schedules` | `handleAPIScheduleSet` | `schedule.Schedule` | 200 `apiScheduleList` (instances omitted: `[]`), 400 on validation with the validator's message, 429 budget |
+| `POST /api/cli/apps/{account}/{app}/schedules/{name}/delete` | `handleAPIScheduleRemove` | — | 200 `apiScheduleList`; 404 (the byte-identical one — call the same helper apiAccountApp uses for its 404) when the name is not declared |
+| `POST /api/cli/apps/{account}/{app}/schedules/{name}/run` | `handleAPIScheduleRun` | — | 200 `{"requested": true, "account", "app", "name"}`; 404 as above when undeclared |
+
+Every handler: `apiAccountApp` → (writes) `allowWrite` → `s.cfg.Releases == nil` → 503 with the restart door's sentence adapted ("this console cannot record a schedule") → 15s ctx → work → audit → bumpSerial (writes only) → respond. Store errors: log, return 502 with a fixed sentence ("could not record the schedule — nothing changed; try again"). Read `apicli.go:768-860` first and mirror its shape and tone.
+
+- [ ] **Step 1: Write the tests** modelled on `apirestart_test.go` (use `restartFixture`-style setup: an app `titogo` in account `bab`, a member token). Pin: (1) set → declaration in bucket has the schedule, serial is 1, response lists it; (2) set with `every: "30s"` → 400 with "shortest interval" in the body and NO bucket write, serial 0; (3) list shows declaration + a status object you write directly with `schedule.WriteStatus` (host `x.bab.oncarlos.com`); (4) delete removes and bumps; delete of an unknown name → 404 whose body equals the 404 body from probing a nonexistent app (fetch both, compare bytes); (5) run → `schedule-run.json` etag moves and serial bumps; run on an undeclared name → 404, no touch; (6) a non-member's token → 404 on GET (probe returns the same bytes as unknown-app).
+
+- [ ] **Step 2: Implement `apischedule.go`** and register the doors.
+
+- [ ] **Step 3: Gate**
+
+Run: `go test ./internal/console/ -count=1 && go vet ./internal/console/ && gofmt -l internal/console`
+Expected: PASS including `TestEveryCLIDoorNamesItsRoles` and the closed-door equality.
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add internal/console
+git commit -m "console: schedule doors — list, set, remove, run (Operate)
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+```
+
+---
+
+### Task 9: CLI — `carlos schedule`
+
+**Files:**
+- Create: `schedule.go`, `schedule_test.go` (root package `main`)
+- Modify: `apimode.go` (client methods), `main.go` (dispatch `case "schedule": runSchedule(args)` and the usage line under *Your apps*, after `restart`).
+
+**Interfaces:**
+- Consumes: Task 8's `apiScheduleList` shape (redeclare it in `apimode.go` as `apiScheduleList` with the same JSON tags — the CLI imports `internal/schedule` for `Schedule`/`Status`/validation, never `internal/console`), `apiPath`, `postJSON`, `getJSON`, `apiErrorFromBody`, `notFoundErr`, `accountFlags`, `resolvedAccount`, `resolvedApp`, `validAppName`, `printUsing`, `colorFor`, `resolveMode`.
+- Produces:
+
+```go
+func (s *apiSession) scheduleList(ctx, a apiAcct, app string) (apiScheduleList, error) // GET …/schedules
+func (s *apiSession) scheduleSet(ctx, a apiAcct, app string, sc schedule.Schedule) (apiScheduleList, error) // POST …/schedules
+func (s *apiSession) scheduleRemove(ctx, a apiAcct, app, name string) (apiScheduleList, error) // POST …/schedules/{name}/delete
+func (s *apiSession) scheduleRun(ctx, a apiAcct, app, name string) error // POST …/schedules/{name}/run
+func runSchedule(args []string) // sub-verbs ls | set | rm | run
+```
+
+Verb surface (exactly):
+
+```
+carlos schedule ls -app <a> [-account …] [-console …]
+carlos schedule set -app <a> -name <n> (-every <dur> | -cron "<5 fields>") -path </p>
+carlos schedule rm -app <a> -name <n>
+carlos schedule run -app <a> -name <n>
+```
+
+`set` validates with `(&sc).Validate()` BEFORE the round trip and prints, on success:
+```
+recorded sync for bab/app — every 6h0m0s → POST /jobs/sync; first fire at 2026-08-23T18:00:00Z (UTC); boxes pick it up within seconds — `carlos schedule ls -app app` shows each instance's next run
+```
+(`first fire` from `sp.Next(time.Now())`.) `run` prints `run requested for sync on bab/app — fires on every instance within ~15s; watch it with carlos logs -app app -f`. `rm` prints `removed sync from bab/app`.
+
+`ls` output:
+
+```
+schedules for bab/app (updated 2026-08-23T22:14:09Z)
+ NAME WHEN PATH
+ reminders cron 0 8 * * * /jobs/reminders
+ sync every 6h0m0s /jobs/sync
+
+instances
+ a.bab.oncarlos.com (box carlos-ie-1, reported 12s ago)
+ sync next in 4h52m last 1h08m ago → 204 in 3.2s
+ reminders next in 9h53m last —
+ u.bab.oncarlos.com unit-backed instance — scheduled work delivers to exec-backed instances in this release
+```
+
+No schedules: `no schedules declared for bab/app — carlos schedule set -app app -name <n> -every 6h -path /jobs/<n>`. No instances: `no instance has reported yet — boxes report within seconds of a schedule landing`. Relative times via a small helper (`in 4h52m` / `1h08m ago` / `—` for zero); truncate to minutes above an hour, seconds below.
+
+- [ ] **Step 1: Write `schedule_test.go`**: wire-shape tests for all four client calls (as `restart_test.go` does — path, method, bearer, canonical names back), a test that `set` with `-every 30s` exits before any HTTP call (spawn the binary as `restart_test.go`'s e2e tests do — read how `carlosTestEnv` and `cmd.Dir` are used and copy that exactly; the CLAUDE.md rule about test envs is non-negotiable), and a test of the `ls` renderer against a fixed `apiScheduleList` and a fixed `now`.
+
+- [ ] **Step 2: Implement.** `runSchedule` refuses bucket mode with a `requireRestartAPISession`-shaped message of its own (`carlos schedule works through a console…`).
+
+- [ ] **Step 3: Gate**
+
+Run: `go build . && go test . -count=1 -run 'Schedule|Usage' && go vet . && gofmt -l *.go && GOOS=darwin go build .`
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add schedule.go schedule_test.go apimode.go main.go
+git commit -m "cli: carlos schedule ls | set | rm | run
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+```
+
+---
+
+### Task 10: Whole-tree gate, review, PR
+
+- [ ] **Step 1: Full gate**
+
+```bash
+go build ./... && go vet ./... && go test ./... -count=1 && GOOS=darwin go build . && gofmt -l $(git diff --name-only main -- '*.go')
+```
+Expected: all green, gofmt prints nothing.
+
+- [ ] **Step 2: Reviews** — Opus reviews Tasks 6, 7, 8 (production paths, authority); a Fable whole-branch review last. Fix findings, re-gate.
+
+- [ ] **Step 3: Push and open the PR** (`gh pr create`) with a body that carries: the spec link, the decisions marked **(decision)** in the spec as a checklist for Paul, the rollout note (§11: agent + console, either order safe), the live verification recipe (§11), and the follow-ups (seapointish migration, rastrillo helper, unit-backed delivery, website + CLI release batch). End the body with `🤖 Generated with [Claude Code](https://claude.com/claude-code)`.
+
+---
+
+## Self-review
+
+- **Spec coverage:** §4.1/4.2/4.3 → T1; §5 → T2; §6.1 → T1; §6.2 → T3; §6.3 → T4 + T7; §6.4 → T6; §6.5 → T3 (`UnsupportedUnit`) + T7 reporter; §7.1 → contract lives in the spec (no member-docs dir in this repo; website batch is a follow-up); §7.2 → T5 + T6; §8 → T8; §9 → T9; §11 → T10.
+- **Placeholder scan:** none; every code step carries code.
+- **Type consistency:** `schedule.TimerStatus`/`Status` used identically in T1/T3/T5/T8/T9; `registry.Timer` fields match across T2/T3/T4/T5; `Runner.Token func(host string) (string, error)` matches `TokenFromEnvFile`; `Converger.Report`/`Runner.Report` share `func(registry.Route, schedule.Status)`; `edge.UnixProxy` returns `http.Handler` (T6) and is used as one (T7).