From c3d8271e1506564d7d1e912e76588bba330486ff Mon Sep 17 00:00:00 2001 From: kami Date: Sun, 26 Jul 2026 20:44:15 +0400 Subject: [PATCH] close worktree transport and lifecycle contract gaps --- internal/continuity/continuity.go | 38 +++++++++++ internal/continuity/continuity_test.go | 17 +++++ internal/domain/domain.go | 19 +++++- internal/orchestrator/orchestrator.go | 90 +++++++++++++++++++++++++- internal/store/store.go | 11 ++++ internal/store/store_test.go | 15 +++++ 6 files changed, 185 insertions(+), 5 deletions(-) diff --git a/internal/continuity/continuity.go b/internal/continuity/continuity.go index 51cdf28..ebeb4df 100644 --- a/internal/continuity/continuity.go +++ b/internal/continuity/continuity.go @@ -128,6 +128,22 @@ func ValidatePickup(root string, h Handoff, taskFileSHA string) error { return nil } +// VerifyTaskFile ensures the worktree contains the original, immutable task. +func VerifyTaskFile(root, taskFileSHA string) error { + if taskFileSHA == "" { + return errors.New("TASK.md hash required") + } + b, err := os.ReadFile(filepath.Join(root, "TASK.md")) + if err != nil { + return err + } + sum := sha256.Sum256(b) + if hex.EncodeToString(sum[:]) != taskFileSHA { + return errors.New("TASK.md changed") + } + return nil +} + type CAS interface { PutArtifact([]byte) (string, error) Artifact(string) ([]byte, error) @@ -175,6 +191,9 @@ func ScratchCommit(root, branch, message string) error { if branch == "" || strings.ContainsAny(branch, " \t\n") { return errors.New("invalid scratch branch") } + if strings.TrimSpace(message) == "" { + return errors.New("scratch commit message required") + } for _, args := range [][]string{{"switch", "-c", branch}, {"add", "-A"}, {"commit", "-m", message}} { if err := exec.Command("git", append([]string{"-C", root}, args...)...).Run(); err != nil { return err @@ -182,3 +201,22 @@ func ScratchCommit(root, branch, message string) error { } return nil } + +// ScratchSync pushes/pulls a scratch branch. Pull uses fast-forward-only to +// avoid silently merging independent WIP histories. +func ScratchSync(root, branch, remote string, push bool) error { + if branch == "" || strings.ContainsAny(branch, " \t\n") || remote == "" { + return errors.New("invalid scratch sync") + } + args := []string{"-C", root, "push", remote, branch} + if !push { + args = []string{"-C", root, "fetch", remote, branch} + } + if err := exec.Command("git", args...).Run(); err != nil { + return err + } + if !push { + return exec.Command("git", "-C", root, "merge", "--ff-only", "FETCH_HEAD").Run() + } + return nil +} diff --git a/internal/continuity/continuity_test.go b/internal/continuity/continuity_test.go index f0d9ae1..14c941b 100644 --- a/internal/continuity/continuity_test.go +++ b/internal/continuity/continuity_test.go @@ -54,3 +54,20 @@ func TestDecodeRejectsUnknownKnowledgeFields(t *testing.T) { t.Fatal("expected strict schema error") } } + +func TestVerifyTaskFileRejectsMutation(t *testing.T) { + root := t.TempDir() + if err := os.WriteFile(filepath.Join(root, "TASK.md"), []byte("task"), 0644); err != nil { + t.Fatal(err) + } + sum := sha256.Sum256([]byte("task")) + if err := VerifyTaskFile(root, hex.EncodeToString(sum[:])); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(root, "TASK.md"), []byte("changed"), 0644); err != nil { + t.Fatal(err) + } + if err := VerifyTaskFile(root, hex.EncodeToString(sum[:])); err == nil { + t.Fatal("expected immutable task check to fail") + } +} diff --git a/internal/domain/domain.go b/internal/domain/domain.go index 3b292c0..a41f12e 100644 --- a/internal/domain/domain.go +++ b/internal/domain/domain.go @@ -119,13 +119,21 @@ func ValidatePayload(typ string, p map[string]any) error { if err := requiredString("harness_id"); err != nil { return err } - if _, ok := p["until_ns"].(float64); !ok { - return fmt.Errorf("%w: until_ns required", ErrInvalid) + until, untilOK := p["until_ns"].(float64) + if ttl, ok := p["ttl"].(float64); ok { + if ttl <= 0 { + return fmt.Errorf("%w: ttl invalid", ErrInvalid) + } + } else if !untilOK || until <= float64(time.Now().UnixNano()) { + return fmt.Errorf("%w: ttl required", ErrInvalid) } - if v, ok := p["expected_version"].(float64); ok && v < 0 { + if v, ok := p["expected_version"].(float64); !ok || v < 0 || v != float64(int(v)) { return fmt.Errorf("%w: expected_version invalid", ErrInvalid) } case "TaskReleased": + if v, ok := p["anchor_sha"].(string); ok && (len(v) != 40 || strings.TrimSpace(v) != v) { + return fmt.Errorf("%w: anchor_sha invalid", ErrInvalid) + } if err := requiredString("handoff_ref"); err != nil && p["reason"] == nil { return err } @@ -133,6 +141,11 @@ func ValidatePayload(typ string, p map[string]any) error { if err := requiredString("report_ref"); err != nil { return err } + if receipt, ok := p["receipt"]; ok { + if m, ok := receipt.(map[string]any); !ok || len(m) == 0 { + return fmt.Errorf("%w: receipt invalid", ErrInvalid) + } + } case "TaskFailed": if err := requiredString("reason"); err != nil { return err diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go index 836b018..16db447 100644 --- a/internal/orchestrator/orchestrator.go +++ b/internal/orchestrator/orchestrator.go @@ -7,6 +7,7 @@ import ( "context" "encoding/json" "fmt" + "orchestra/internal/continuity" "orchestra/internal/domain" "orchestra/internal/herdr" "orchestra/internal/store" @@ -28,8 +29,9 @@ type Adapters interface { // to be a clone containing the project's remote; callers may set a separate // root per deployment. type GitWorktrees struct { - Root string - Repo string + Root string + Repo string + TaskFileSHA string } func (w GitWorktrees) Create(ctx context.Context, t domain.Task) (string, error) { @@ -41,6 +43,11 @@ func (w GitWorktrees) Create(ctx context.Context, t domain.Task) (string, error) } p := filepath.Join(w.Root, t.ID) if _, err := os.Stat(p); err == nil { + if w.TaskFileSHA != "" { + if err := continuity.VerifyTaskFile(p, w.TaskFileSHA); err != nil { + return "", err + } + } return p, nil } branch := "orchestra/" + t.ID @@ -48,6 +55,11 @@ func (w GitWorktrees) Create(ctx context.Context, t domain.Task) (string, error) if out, err := cmd.CombinedOutput(); err != nil { return "", fmt.Errorf("%s: %w", string(out), err) } + if w.TaskFileSHA != "" { + if err := continuity.VerifyTaskFile(p, w.TaskFileSHA); err != nil { + return "", err + } + } return p, nil } @@ -69,6 +81,33 @@ type Coordinator struct { mu sync.Mutex sessions map[string]herdr.Session loaded bool + healthMu sync.RWMutex + health MonitorHealth +} + +type MonitorHealth struct { + Running bool `json:"running"` + LastRun time.Time `json:"last_run"` + LastError string `json:"last_error,omitempty"` + Expired int `json:"expired"` +} + +func (c *Coordinator) MonitorHealth() MonitorHealth { + c.healthMu.RLock() + defer c.healthMu.RUnlock() + return c.health +} +func (c *Coordinator) setMonitorHealth(err error, expired int) { + c.healthMu.Lock() + defer c.healthMu.Unlock() + c.health.Running = err == nil + c.health.LastRun = time.Now().UTC() + c.health.Expired += expired + if err != nil { + c.health.LastError = err.Error() + } else { + c.health.LastError = "" + } } func (c *Coordinator) loadSessions() { @@ -128,6 +167,7 @@ func (c *Coordinator) Reconcile(ctx context.Context) error { // and frees the lease for pickup by the router. func (c *Coordinator) Monitor(ctx context.Context, hard float64, interval time.Duration) error { if err := c.Reconcile(ctx); err != nil { + c.setMonitorHealth(err, 0) return err } if interval <= 0 { @@ -138,12 +178,58 @@ func (c *Coordinator) Monitor(ctx context.Context, hard float64, interval time.D for { select { case <-ctx.Done(): + c.healthMu.Lock() + c.health.Running = false + c.healthMu.Unlock() return ctx.Err() case <-t.C: + expired, err := c.expire(ctx) + c.setMonitorHealth(err, len(expired)) + if err != nil { + continue + } c.rotate(ctx, hard) } } } + +func (c *Coordinator) expire(ctx context.Context) ([]domain.Event, error) { + // pane.exited is the low-latency path; lease expiry below remains the + // authoritative backstop when herdr misses an exit notification. + c.loadSessions() + c.mu.Lock() + for taskID, s := range c.sessions { + if t, ok := c.Store.Task(taskID); ok && t.State == domain.StateLeased { + if a, ae := c.Adapters.Adapter(s.Harness); ae == nil { + if p, ok := a.(herdr.PaneExit); ok { + if exited, ee := p.PaneExited(ctx, s); ee == nil && exited { + b, _ := json.Marshal(map[string]string{"reason": "pane_exited", "harness_id": s.Harness}) + _ = c.Store.Append(domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: taskID, Version: t.Version + 1, Payload: b}) + } + } + } + } + } + c.mu.Unlock() + events, err := c.Store.ExpireLeases(time.Now()) + if err != nil { + return events, err + } + for _, e := range events { + c.loadSessions() + c.mu.Lock() + s, ok := c.sessions[e.TaskID] + delete(c.sessions, e.TaskID) + if ok { + if a, ae := c.Adapters.Adapter(s.Harness); ae == nil { + _ = a.Kill(ctx, s) + } + } + _ = c.saveSessionsLocked() + c.mu.Unlock() + } + return events, nil +} func (c *Coordinator) rotate(ctx context.Context, hard float64) { c.loadSessions() c.mu.Lock() diff --git a/internal/store/store.go b/internal/store/store.go index 0e06374..7f64398 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -185,6 +185,17 @@ func (s *Store) Append(e domain.Event) error { if taskExists && e.Version != t.Version+1 { return domain.ErrConflict } + // Every optimistic lifecycle writer may carry its observed version. Enforce + // it at the append boundary so non-HTTP producers receive the same CAS. + var contract map[string]any + if err := json.Unmarshal(e.Payload, &contract); err != nil { + return err + } + if expected, ok := contract["expected_version"].(float64); ok { + if expected != float64(int(expected)) || !taskExists || int(expected) != t.Version { + return domain.ErrConflict + } + } if e.Type == "TaskLeased" { var p struct { ExpectedVersion *int `json:"expected_version"` diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 4ddaf6c..a701ab8 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -92,3 +92,18 @@ func TestLifecycleEventsRequireEvidence(t *testing.T) { }) } } + +func TestExpectedVersionIsCheckedForEveryWriter(t *testing.T) { + s, err := Open(t.TempDir()) + if err != nil { + t.Fatal(err) + } + if err := s.Append(created("create")); err != nil { + t.Fatal(err) + } + p := json.RawMessage(`{"reason":"rotate","expected_version":0}`) + err = s.Append(domain.Event{Type: "TaskReleased", TaskID: "task-1", Version: 2, Payload: p}) + if err != domain.ErrConflict { + t.Fatalf("expected CAS conflict, got %v", err) + } +}