orchestra/internal/store/store.go

package store

import (
	"bufio"
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"orchestra/internal/authz"
	"orchestra/internal/domain"
	"os"
	"path/filepath"
	"sort"
	"sync"
	"time"
)

// QuotaUsage is an indexed native-usage receipt. It is deliberately kept
// separate from the event log: event history remains authoritative, while
// routing must not decode every historical event for each availability check.
type QuotaUsage struct {
	At       time.Time
	Consumed float64
	Known    bool
}

type quotaIndex struct {
	records       []QuotaUsage // ordered by At
	prefix        []float64
	unknownPrefix []int
}

type Store struct {
	mu       sync.Mutex
	path     string
	cas      string
	events   []domain.Event
	tasks    map[string]domain.Task
	external map[string]string
	// activeLeases is the routing occupancy index. Needs-attention retains a
	// fenced owner, so it counts as active until that lease is released.
	activeLeases map[string]map[string]struct{}
	quota        map[string]quotaIndex
	snapshot     string
	seq          uint64
}

func Open(dir string) (*Store, error) {
	if err := os.MkdirAll(dir, 0755); err != nil {
		return nil, err
	}
	s := &Store{path: filepath.Join(dir, "events.jsonl"), cas: filepath.Join(dir, "cas"), snapshot: filepath.Join(dir, "snapshot.json"), tasks: map[string]domain.Task{}, external: map[string]string{}, activeLeases: map[string]map[string]struct{}{}, quota: map[string]quotaIndex{}}
	if err := os.MkdirAll(s.cas, 0755); err != nil {
		return nil, err
	}
	// A snapshot is a disposable read cache, never recovery authority. Loading
	// it before the log let a partially-written snapshot become a different
	// history than events.jsonl after a crash. Rebuild every projection from
	// the append-only, fsynced log instead.
	f, err := os.Open(s.path)
	if os.IsNotExist(err) {
		return s, nil
	}
	if err != nil {
		return nil, err
	}
	defer f.Close()
	sc := bufio.NewScanner(f)
	var expected uint64 = 1
	for sc.Scan() {
		var e domain.Event
		if err := json.Unmarshal(sc.Bytes(), &e); err == nil {
			if err := domain.ValidateEvent(e); err != nil {
				return nil, err
			}
			if e.Seq != expected {
				return nil, fmt.Errorf("event sequence gap: got %d, want %d", e.Seq, expected)
			}
			if t, ok := s.tasks[e.TaskID]; ok && e.Version != t.Version+1 {
				return nil, domain.ErrConflict
			}
			global := e.Type == "QuotaReported" || e.Type == "StandupAdvisory" || e.Type == "ApprovalGranted" || e.Type == "ApprovalDenied"
			if _, ok := s.tasks[e.TaskID]; !ok && e.Type != "TaskCreated" && !global {
				return nil, domain.ErrNotFound
			}
			s.events = append(s.events, e)
			s.seq = e.Seq
			if err := s.apply(e); err != nil {
				return nil, err
			}
			expected++
		} else {
			return nil, err
		}
	}
	return s, sc.Err()
}

// NormalizeLegacyEventSequence repairs one explicitly recognized pre-v2 log
// shape: a prefix of two or more records all numbered seq=1, followed by a
// contiguous suffix numbered 2, 3, ... that restarts the first task at
// version 2. Early Orchestra releases emitted precisely that shape: records
// after the first seq=1 were never included in the compatibility snapshot,
// then were replayed again after restart. The migration retains the initial
// TaskCreated plus the contiguous suffix, discarding only the provably
// abandoned duplicate prefix. It is not a general corruption repair tool:
// any other gap or duplicate is rejected so Open's fail-closed recovery
// guarantee remains intact.
//
// The caller must stop every coordinator using dir first. The original log is
// durably copied to events.jsonl.legacy-<unix-nano> before an fsync+rename
// replacement is installed. The return value reports whether a migration was
// needed.
func NormalizeLegacyEventSequence(dir string) (bool, error) {
	path := filepath.Join(dir, "events.jsonl")
	raw, err := os.ReadFile(path)
	if err != nil {
		return false, err
	}
	lines := bytes.Split(bytes.TrimSpace(raw), []byte{'\n'})
	if len(lines) == 0 || (len(lines) == 1 && len(lines[0]) == 0) {
		return false, nil
	}
	events := make([]domain.Event, len(lines))
	for i, line := range lines {
		if err := json.Unmarshal(line, &events[i]); err != nil {
			return false, fmt.Errorf("event %d: %w", i+1, err)
		}
	}
	canonical := true
	for i, e := range events {
		if e.Seq != uint64(i+1) {
			canonical = false
			break
		}
	}
	if canonical {
		return false, nil
	}
	prefix := 0
	for prefix < len(events) && events[prefix].Seq == 1 {
		prefix++
	}
	if prefix < 2 {
		return false, fmt.Errorf("refusing non-legacy event sequence")
	}
	if events[0].Type != "TaskCreated" || events[0].Version != 1 {
		return false, fmt.Errorf("refusing legacy sequence without an initial task creation")
	}
	for i := 1; i < prefix; i++ {
		if events[i].TaskID != events[0].TaskID || events[i].Version < 2 {
			return false, fmt.Errorf("refusing non-legacy duplicate prefix at record %d", i+1)
		}
	}
	if prefix == len(events) || events[prefix].Type != "TaskLeased" || events[prefix].TaskID != events[0].TaskID || events[prefix].Version != 2 {
		return false, fmt.Errorf("refusing legacy sequence without a task-version-2 restart")
	}
	for i := prefix; i < len(events); i++ {
		want := uint64(i - prefix + 2)
		if events[i].Seq != want {
			return false, fmt.Errorf("refusing non-legacy event sequence at record %d: got %d, want %d", i+1, events[i].Seq, want)
		}
	}
	backup := fmt.Sprintf("%s.legacy-%d", path, time.Now().UTC().UnixNano())
	backupFile, err := os.OpenFile(backup, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0644)
	if err != nil {
		return false, err
	}
	if _, err = backupFile.Write(raw); err == nil {
		err = backupFile.Sync()
	}
	if closeErr := backupFile.Close(); err == nil {
		err = closeErr
	}
	if err != nil {
		return false, err
	}
	tmp := path + ".sequence-migration.tmp"
	out, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0644)
	if err != nil {
		return false, err
	}
	kept := append(events[:1:1], events[prefix:]...)
	for i := range kept {
		kept[i].Seq = uint64(i + 1)
		line, marshalErr := json.Marshal(kept[i])
		if marshalErr != nil {
			err = marshalErr
			break
		}
		if _, err = out.Write(append(line, '\n')); err != nil {
			break
		}
	}
	if err == nil {
		err = out.Sync()
	}
	if closeErr := out.Close(); err == nil {
		err = closeErr
	}
	if err != nil {
		return false, err
	}
	if err := os.Rename(tmp, path); err != nil {
		return false, err
	}
	d, err := os.Open(dir)
	if err != nil {
		return false, err
	}
	defer d.Close()
	if err := d.Sync(); err != nil {
		return false, err
	}
	return true, nil
}
func (s *Store) apply(e domain.Event) error {
	var p map[string]any
	if err := json.Unmarshal(e.Payload, &p); err != nil {
		return err
	}
	t := s.tasks[e.TaskID]
	if e.Type == "QuotaReported" {
		known := true
		if v, ok := p["known"].(bool); ok {
			known = v
		}
		harness, _ := p["harness_id"].(string)
		consumed, _ := p["consumed"].(float64)
		s.addQuotaUsage(harness, QuotaUsage{At: e.At, Consumed: consumed, Known: known})
		return nil
	}
	if e.Type == "StandupAdvisory" || e.Type == "ApprovalGranted" || e.Type == "ApprovalDenied" {
		return nil
	}
	switch e.Type {
	case "TaskCreated":
		if err := domain.ValidateCreated(p); err != nil {
			return err
		}
		t = domain.Task{ID: e.TaskID, Source: p["source"].(string), ExternalID: p["external_id"].(string), Project: p["project"].(string), State: domain.StateQueued}
		if v, ok := p["parent"].(string); ok {
			t.Parent = v
		}
		if v, ok := p["inherent_priority"].(float64); ok {
			t.InherentPriority = int(v)
		}
		if v, ok := p["due"].(string); ok {
			if d, err := time.Parse(time.RFC3339, v); err == nil {
				t.Due = &d
			}
		}
		if v, ok := p["estimate"].(map[string]any); ok {
			t.Estimate = &domain.Estimate{}
			t.Estimate.Value, _ = v["value"].(float64)
			t.Estimate.Who, _ = v["who"].(string)
			t.Estimate.Confidence, _ = v["confidence"].(float64)
		}
		if v, ok := p["capability"].([]any); ok {
			for _, x := range v {
				if z, ok := x.(string); ok {
					t.Capability = append(t.Capability, z)
				}
			}
		}
		if v, ok := p["title"].(string); ok {
			t.Title = v
		}
		if v, ok := p["description"].(string); ok {
			t.Description = v
		}
		if v, ok := p["acceptance"].([]any); ok {
			for _, item := range v {
				if text, ok := item.(string); ok {
					t.Acceptance = append(t.Acceptance, text)
				}
			}
		}
		if v, ok := p["quality_gate"].(string); ok {
			t.QualityGate = v
		}
		s.external[t.Source+"\x00"+t.ExternalID] = t.ID
	case "TaskLeased":
		t.State = domain.StateLeased
		t.LifecyclePhase = "lease_issued"
		t.LastError = ""
		epoch, _ := p["lease_epoch"].(string)
		if epoch == "" {
			// A pre-fencing event cannot safely be renewed by an old worker.
			// Deriving a stable token from the durable event identity makes the
			// recovered lease observable but non-renewable until it expires.
			epoch = "legacy:" + e.ID
		}
		t.Lease = &domain.Lease{HarnessID: p["harness_id"].(string), Epoch: epoch, Until: time.Unix(0, int64(p["until_ns"].(float64)))}
	case "TaskLeaseRenewed":
		epoch, _ := p["lease_epoch"].(string)
		if epoch == "" && t.Lease != nil {
			epoch = t.Lease.Epoch
		}
		t.Lease = &domain.Lease{HarnessID: p["harness_id"].(string), Epoch: epoch, Until: time.Unix(0, int64(p["until_ns"].(float64)))}
	case "TaskLaunchAcknowledged":
		t.LifecyclePhase = "started"
	case "TaskReleased":
		t.State = domain.StateQueued
		t.LifecyclePhase = "reclaimed"
		t.Lease = nil
		t.HandoffRef, _ = p["handoff_ref"].(string)
		t.ReleaseTransaction, _ = p["transaction_id"].(string)
		t.ReleaseAnchor, _ = p["anchor_sha"].(string)
		t.PickupTransaction, t.PickupLeaseVersion = "", 0
		if t.HandoffRef == "" {
			// A handoff-less release is a reclaim. Persist the retry decision
			// here so expiry, pane exit, and a worker NACK all use the same
			// crash-safe transition instead of router-local counters.
			t.Attempt++
			t.FailureClass, _ = p["failure_class"].(string)
			if t.FailureClass == "" {
				t.FailureClass, _ = p["reason"].(string)
			}
			t.NextRetryAt = e.At.Add(retryBackoff(t.Attempt))
		} else {
			t.NextRetryAt = time.Time{}
			t.FailureClass = ""
		}
	case "TaskPickupValidated":
		t.PickupTransaction, _ = p["transaction_id"].(string)
		if v, ok := p["lease_version"].(float64); ok {
			t.PickupLeaseVersion = int(v)
		}
	case "TaskCompleted":
		t.State = domain.StateCompleted
		t.Lease = nil
	case "TaskFailed":
		t.State = domain.StateFailed
		t.Lease = nil
	case "TaskBlocked", "TaskNeedsAttention":
		if e.Type == "TaskBlocked" {
			t.State = domain.StateBlocked
			t.Lease = nil
		} else {
			// Recovery diagnostics must not revoke the fenced owner. A late
			// completion is still valid only from this exact lease epoch.
			t.State = domain.StateNeedsAttention
		}
		t.Blocker, _ = p["blocker"].(string)
		t.BlockReason = domain.InferBlockReason(t.Blocker)
		if v, ok := p["block_reason"].(string); ok && domain.BlockReason(v).Valid() {
			t.BlockReason = domain.BlockReason(v)
		}
		t.BlockedAt = e.At
		t.LastPaneID, _ = p["pane_id"].(string)
		t.LastHarness, _ = p["harness_id"].(string)
		t.PaneState, _ = p["pane_state"].(string)
		if t.PaneState == "" {
			t.PaneState = "unknown"
		}
	case "TaskAmended":
		if v, ok := p["title"].(string); ok {
			t.Title = v
		}
		if v, ok := p["description"].(string); ok {
			t.Description = v
		}
		if v, ok := p["inherent_priority"].(float64); ok {
			t.InherentPriority = int(v)
		}
		if v, ok := p["due"].(string); ok {
			if d, err := time.Parse(time.RFC3339, v); err == nil {
				t.Due = &d
			}
		}
	case "TaskCorrected":
		if v, ok := p["title"].(string); ok {
			t.Title = v
		}
		if v, ok := p["description"].(string); ok {
			t.Description = v
		}
		if v, ok := p["inherent_priority"].(float64); ok {
			t.InherentPriority = int(v)
		}
		if v, ok := p["due"].(string); ok {
			if d, err := time.Parse(time.RFC3339, v); err == nil {
				t.Due = &d
			}
		}
		if v, ok := p["state"].(string); ok {
			t.State = domain.TaskState(v)
			if t.State != domain.StateLeased {
				t.Lease = nil
			}
		}
	}
	if phase, ok := p["lifecycle_phase"].(string); ok && phase != "" {
		t.LifecyclePhase = phase
	}
	if last, ok := p["last_error"].(string); ok {
		t.LastError = last
	}
	// Terminal and release events may carry an owner-produced snapshot from
	// immediately before a worker/coordinator drops its live session mapping.
	// Preserve it independently of the current task state so historical task
	// pages never have to imply a pane is live just because its ID is known.
	if raw, ok := p["session_evidence"].(map[string]any); ok {
		var evidence domain.SessionEvidence
		if b, err := json.Marshal(raw); err == nil && json.Unmarshal(b, &evidence) == nil {
			t.LastSession = evidence
			if evidence.PaneID != "" {
				t.LastPaneID = evidence.PaneID
			}
			if evidence.HarnessID != "" {
				t.LastHarness = evidence.HarnessID
			}
			if evidence.PaneState != "" {
				t.PaneState = evidence.PaneState
			}
		}
	}
	t.Version = e.Version
	s.replaceTask(e.TaskID, t)
	return nil
}

func activeLease(t domain.Task) (string, bool) {
	if (t.State != domain.StateLeased && t.State != domain.StateNeedsAttention) || t.Lease == nil {
		return "", false
	}
	return t.Lease.HarnessID, t.Lease.HarnessID != ""
}

func (s *Store) replaceTask(id string, next domain.Task) {
	if previous, ok := s.tasks[id]; ok {
		if harness, active := activeLease(previous); active {
			delete(s.activeLeases[harness], id)
			if len(s.activeLeases[harness]) == 0 {
				delete(s.activeLeases, harness)
			}
		}
	}
	if harness, active := activeLease(next); active {
		if s.activeLeases[harness] == nil {
			s.activeLeases[harness] = map[string]struct{}{}
		}
		s.activeLeases[harness][id] = struct{}{}
	}
	s.tasks[id] = next
}

func (s *Store) addQuotaUsage(harness string, usage QuotaUsage) {
	index := s.quota[harness]
	// Receipts normally arrive in timestamp order, making index maintenance
	// O(1). Keep the out-of-order path correct for replay and delayed worker
	// reports without imposing its rebuild cost on the common append path.
	at := sort.Search(len(index.records), func(i int) bool { return usage.At.Before(index.records[i].At) })
	if at == len(index.records) {
		if len(index.prefix) == 0 {
			index.prefix = []float64{0}
			index.unknownPrefix = []int{0}
		}
		prefix := index.prefix[len(index.prefix)-1]
		unknown := index.unknownPrefix[len(index.unknownPrefix)-1]
		index.records = append(index.records, usage)
		index.prefix = append(index.prefix, prefix+usage.Consumed)
		if !usage.Known {
			unknown++
		}
		index.unknownPrefix = append(index.unknownPrefix, unknown)
		s.quota[harness] = index
		return
	}
	index.records = append(index.records, QuotaUsage{})
	copy(index.records[at+1:], index.records[at:])
	index.records[at] = usage
	index.prefix = make([]float64, len(index.records)+1)
	index.unknownPrefix = make([]int, len(index.records)+1)
	for i, receipt := range index.records {
		index.prefix[i+1] = index.prefix[i] + receipt.Consumed
		index.unknownPrefix[i+1] = index.unknownPrefix[i]
		if !receipt.Known {
			index.unknownPrefix[i+1]++
		}
	}
	s.quota[harness] = index
}

func retryBackoff(attempt int) time.Duration {
	if attempt < 1 {
		attempt = 1
	}
	backoff := time.Minute
	for i := 1; i < attempt && backoff < 30*time.Minute; i++ {
		backoff *= 2
	}
	if backoff > 30*time.Minute {
		return 30 * time.Minute
	}
	return backoff
}
func (s *Store) Append(e domain.Event) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if e.At.IsZero() {
		e.At = time.Now().UTC()
	}
	if e.Seq == 0 {
		e.Seq = s.seq + 1
	}
	if e.SchemaVersion == 0 {
		e.SchemaVersion = domain.CurrentEventSchema
	}
	if err := domain.ValidateEvent(e); err != nil {
		return err
	}
	// Enforced once, at the append boundary, per spec §7.1/invariant 4 — every
	// producer (HTTP handler, router, coordinator, provider, federation relay)
	// must declare its Surface here; there is no separate in-process bypass.
	if err := authz.AuthorizeEvent(authz.Surface(e.Surface), e.Type); err != nil {
		return err
	}
	if e.Type == "TaskCreated" {
		var p map[string]any
		if err := json.Unmarshal(e.Payload, &p); err != nil {
			return err
		}
		if id := s.external[p["source"].(string)+"\x00"+p["external_id"].(string)]; id != "" {
			return domain.ErrDuplicate
		}
	}
	t, taskExists := s.tasks[e.TaskID]
	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 err := s.validateTransition(e, t, taskExists, contract); err != nil {
		return err
	}
	if e.Type == "TaskLeased" {
		var p struct {
			ExpectedVersion *int `json:"expected_version"`
		}
		if err := json.Unmarshal(e.Payload, &p); err != nil {
			return err
		}
		if p.ExpectedVersion != nil && (t.Version != *p.ExpectedVersion) {
			return domain.ErrConflict
		}
	}
	if e.Type == "TaskCorrected" {
		var p map[string]any
		_ = json.Unmarshal(e.Payload, &p)
		corrects, _ := p["corrects"].(string)
		found := false
		for _, prior := range s.events {
			if prior.ID == corrects && prior.TaskID == e.TaskID {
				found = true
				break
			}
		}
		if !found {
			return fmt.Errorf("%w: corrects references unknown event %q for this task", domain.ErrInvalid, corrects)
		}
	}
	global := e.Type == "QuotaReported" || e.Type == "StandupAdvisory" || e.Type == "ApprovalGranted" || e.Type == "ApprovalDenied"
	if !taskExists && e.Type != "TaskCreated" && !global {
		return domain.ErrNotFound
	}
	if e.Type != "TaskCreated" && (e.Type == "TaskCompleted" || e.Type == "TaskBlocked" || e.Type == "TaskNeedsAttention" || e.Type == "TaskReleased") {
		var p map[string]any
		_ = json.Unmarshal(e.Payload, &p)
		for _, k := range []string{"handoff_ref", "report_ref"} {
			if ref, ok := p[k].(string); ok {
				if _, err := s.Artifact(ref); err != nil {
					return fmt.Errorf("%w: missing artifact %s", domain.ErrInvalid, ref)
				}
			}
		}
	}
	f, err := os.OpenFile(s.path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644)
	if err != nil {
		return err
	}
	defer f.Close()
	b, _ := json.Marshal(e)
	if _, err = f.Write(append(b, '\n')); err != nil {
		return err
	}
	if err = f.Sync(); err != nil {
		return err
	}
	// The event is the commit record. Do not expose a projection that cannot
	// be recovered from it after a power loss.
	if err := s.apply(e); err != nil {
		return err
	}
	s.events = append(s.events, e)
	s.seq = e.Seq
	// Snapshot failure does not roll back a committed event. Open always
	// rebuilds from the log, so leaving a stale cache is safe. Keep only the
	// initial compatibility cache: repeatedly serializing the full projection
	// turns an otherwise O(1) append into O(tasks) work and is never used for
	// recovery.
	if s.seq == 1 {
		_ = s.writeSnapshot()
	}
	return nil
}

// validateTransition keeps lifecycle authority at the durable append
// boundary. A task may be completed/failed while queued by an external
// provider, but once a lease exists its owner and fencing epoch are required
// for every lifecycle mutation.
func (s *Store) validateTransition(e domain.Event, t domain.Task, exists bool, p map[string]any) error {
	if !exists {
		if e.Type != "TaskCreated" && e.Type != "QuotaReported" && e.Type != "StandupAdvisory" && e.Type != "ApprovalGranted" && e.Type != "ApprovalDenied" {
			return domain.ErrNotFound
		}
		return nil
	}
	if e.Type == "TaskLeased" && t.State != domain.StateQueued {
		return domain.ErrConflict
	}
	// Needs-attention is specifically a recoverable leased state, never a
	// second spelling of a terminal operator block on an unowned task.
	if e.Type == "TaskNeedsAttention" && ((t.State != domain.StateLeased && t.State != domain.StateNeedsAttention) || t.Lease == nil) {
		return domain.ErrConflict
	}
	if (t.State != domain.StateLeased && t.State != domain.StateNeedsAttention) || t.Lease == nil {
		return nil
	}
	switch e.Type {
	case "TaskLeaseRenewed", "TaskLaunchAcknowledged", "TaskReleased", "TaskPickupValidated", "TaskCompleted", "TaskBlocked", "TaskNeedsAttention", "TaskFailed":
		owner, _ := p["harness_id"].(string)
		epoch, _ := p["lease_epoch"].(string)
		// Expiry is the one coordinator-owned relinquish path. It still binds
		// the exact epoch that was observed when the timer fired.
		if e.Type == "TaskReleased" {
			if reason, _ := p["reason"].(string); (reason == "lease_expired" || reason == "pane_exited") && owner == t.Lease.HarnessID && epoch == t.Lease.Epoch {
				return nil
			}
		}
		if owner != t.Lease.HarnessID || epoch == "" || epoch != t.Lease.Epoch {
			return domain.ErrConflict
		}
	case "TaskCorrected":
		// Corrections may repair metadata while a task is leased, but cannot
		// smuggle in a lifecycle transition around the current fenced owner.
		if _, changesState := p["state"]; changesState {
			owner, _ := p["harness_id"].(string)
			epoch, _ := p["lease_epoch"].(string)
			if owner != t.Lease.HarnessID || epoch == "" || epoch != t.Lease.Epoch {
				return domain.ErrConflict
			}
		}
	}
	return nil
}
func (s *Store) writeSnapshot() error {
	tasks := make([]domain.Task, 0, len(s.tasks))
	for _, t := range s.tasks {
		tasks = append(tasks, t)
	}
	b, err := json.Marshal(struct {
		Seq   uint64        `json:"seq"`
		Tasks []domain.Task `json:"tasks"`
	}{s.seq, tasks})
	if err != nil {
		return err
	}
	tmp := s.snapshot + ".tmp"
	f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0644)
	if err != nil {
		return err
	}
	if _, err = f.Write(b); err == nil {
		err = f.Sync()
	}
	if closeErr := f.Close(); err == nil {
		err = closeErr
	}
	if err != nil {
		return err
	}
	if err = os.Rename(tmp, s.snapshot); err != nil {
		return err
	}
	dir, err := os.Open(filepath.Dir(s.snapshot))
	if err != nil {
		return err
	}
	defer dir.Close()
	return dir.Sync()
}
func (s *Store) Tasks() []domain.Task {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make([]domain.Task, 0, len(s.tasks))
	for _, t := range s.tasks {
		out = append(out, t)
	}
	return out
}

// SchedulingSnapshot captures the task projection and lease occupancy under
// one lock. A routing pass uses this immutable view, then updates its local
// occupancy as it issues leases; it never repeatedly scan-locks the store.
type SchedulingSnapshot struct {
	Tasks        []domain.Task
	ActiveLeases map[string]int
}

func (s *Store) SchedulingSnapshot() SchedulingSnapshot {
	s.mu.Lock()
	defer s.mu.Unlock()
	snapshot := SchedulingSnapshot{
		Tasks:        make([]domain.Task, 0, len(s.tasks)),
		ActiveLeases: make(map[string]int, len(s.activeLeases)),
	}
	for _, task := range s.tasks {
		snapshot.Tasks = append(snapshot.Tasks, task)
	}
	for harness, leases := range s.activeLeases {
		snapshot.ActiveLeases[harness] = len(leases)
	}
	return snapshot
}

// QuotaSince answers a rolling-window usage query from the per-harness index
// instead of walking events.jsonl. known is false when the interval has no
// native receipt or any receipt explicitly reports unknown usage.
func (s *Store) QuotaSince(harness string, since time.Time) (consumed float64, known bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	index, ok := s.quota[harness]
	if !ok {
		return 0, false
	}
	start := sort.Search(len(index.records), func(i int) bool { return !index.records[i].At.Before(since) })
	if start == len(index.records) {
		return 0, false
	}
	return index.prefix[len(index.records)] - index.prefix[start], index.unknownPrefix[len(index.records)] == index.unknownPrefix[start]
}
func (s *Store) Events(since uint64) []domain.Event {
	s.mu.Lock()
	defer s.mu.Unlock()
	var out []domain.Event
	for _, e := range s.events {
		if e.Seq > since {
			out = append(out, e)
		}
	}
	return out
}
func (s *Store) PutArtifact(b []byte) (string, error) {
	h := domain.Hash(b)
	p := filepath.Join(s.cas, h)
	if _, err := os.Stat(p); errors.Is(err, os.ErrNotExist) {
		tmp := p + ".tmp-" + domain.NewID()
		f, openErr := os.OpenFile(tmp, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0644)
		if openErr != nil {
			return "", openErr
		}
		if _, err = f.Write(b); err == nil {
			err = f.Sync()
		}
		if closeErr := f.Close(); err == nil {
			err = closeErr
		}
		if err != nil {
			_ = os.Remove(tmp)
			return "", err
		}
		if err = os.Rename(tmp, p); err != nil && !errors.Is(err, os.ErrExist) {
			_ = os.Remove(tmp)
			return "", err
		}
		if !errors.Is(err, os.ErrExist) {
			dir, openErr := os.Open(s.cas)
			if openErr != nil {
				return "", openErr
			}
			syncErr := dir.Sync()
			closeErr := dir.Close()
			if syncErr != nil {
				return "", syncErr
			}
			if closeErr != nil {
				return "", closeErr
			}
		}
	}
	return h, nil
}

// Artifact returns a CAS artifact after verifying its content address.
func (s *Store) Artifact(ref string) ([]byte, error) {
	if len(ref) != 64 {
		return nil, fmt.Errorf("%w: invalid artifact reference", domain.ErrInvalid)
	}
	b, err := os.ReadFile(filepath.Join(s.cas, ref))
	if err != nil {
		return nil, err
	}
	if domain.Hash(b) != ref {
		return nil, fmt.Errorf("%w: corrupt artifact", domain.ErrInvalid)
	}
	return b, nil
}

func (s *Store) Task(id string) (domain.Task, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	t, ok := s.tasks[id]
	return t, ok
}

// TaskBySource resolves the task ingested for a given (source, external_id)
// pair — the dedup key Append.ErrDuplicate rejects re-ingestion against.
func (s *Store) TaskBySource(source, externalID string) (domain.Task, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	id := s.external[source+"\x00"+externalID]
	if id == "" {
		return domain.Task{}, false
	}
	t, ok := s.tasks[id]
	return t, ok
}

func (s *Store) Lease(id, harness string, ttl time.Duration) (domain.Event, error) {
	if ttl <= 0 {
		return domain.Event{}, fmt.Errorf("%w: ttl must be positive", domain.ErrInvalid)
	}
	t, ok := s.Task(id)
	if !ok {
		return domain.Event{}, domain.ErrNotFound
	}
	if t.State != domain.StateQueued {
		return domain.Event{}, domain.ErrConflict
	}
	payload := map[string]any{"harness_id": harness, "lease_epoch": domain.NewID(), "ttl": ttl.Seconds(), "until_ns": time.Now().Add(ttl).UnixNano(), "expected_version": t.Version}
	if t.HandoffRef != "" {
		payload["handoff_ref"] = t.HandoffRef
		payload["transaction_id"] = t.ReleaseTransaction
		payload["anchor_sha"] = t.ReleaseAnchor
	}
	p, _ := json.Marshal(payload)
	e := domain.Event{ID: domain.NewID(), Type: "TaskLeased", TaskID: id, Version: t.Version + 1, Payload: p, Surface: string(authz.System)}
	return e, s.Append(e)
}

// RenewLease atomically extends the current owner's lease. The observed task
// version is part of the request so an old worker can never renew a lease
// after release/reassignment.
func (s *Store) RenewLease(id, harness, epoch string, expectedVersion int, ttl time.Duration) (domain.Event, error) {
	if ttl <= 0 {
		return domain.Event{}, fmt.Errorf("%w: ttl must be positive", domain.ErrInvalid)
	}
	t, ok := s.Task(id)
	if !ok {
		return domain.Event{}, domain.ErrNotFound
	}
	if (t.State != domain.StateLeased && t.State != domain.StateNeedsAttention) || t.Lease == nil || t.Lease.HarnessID != harness || t.Lease.Epoch != epoch || t.Version != expectedVersion {
		return domain.Event{}, domain.ErrConflict
	}
	p, _ := json.Marshal(map[string]any{"harness_id": harness, "lease_epoch": epoch, "until_ns": time.Now().Add(ttl).UnixNano(), "expected_version": expectedVersion})
	e := domain.Event{ID: domain.NewID(), Type: "TaskLeaseRenewed", TaskID: id, Version: t.Version + 1, Payload: p, Surface: string(authz.System)}
	return e, s.Append(e)
}

func (s *Store) ExpireLeases(now time.Time) ([]domain.Event, error) {
	var out []domain.Event
	for _, t := range s.Tasks() {
		if e, err := s.ExpireLease(t.ID, now); err != nil {
			if !errors.Is(err, domain.ErrConflict) {
				return out, err
			}
		} else if e.ID != "" {
			out = append(out, e)
		}
	}
	return out, nil
}

// ExpireLease releases exactly the observed lease if, and only if, its TTL
// has elapsed. Coordinators use this one-task form to stop their local pane
// before publishing the release event; the batch helper remains for
// deployments without a local coordinator.
func (s *Store) ExpireLease(id string, now time.Time) (domain.Event, error) {
	t, ok := s.Task(id)
	if !ok {
		return domain.Event{}, domain.ErrNotFound
	}
	if (t.State != domain.StateLeased && t.State != domain.StateNeedsAttention) || t.Lease == nil || t.Lease.Until.After(now) {
		return domain.Event{}, domain.ErrConflict
	}
	p, _ := json.Marshal(map[string]any{"reason": "lease_expired", "harness_id": t.Lease.HarnessID, "lease_epoch": t.Lease.Epoch, "expected_version": t.Version})
	e := domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: t.ID, Version: t.Version + 1, Payload: p, Surface: string(authz.System)}
	return e, s.Append(e)
}