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
	// cursors and decisionSource support human-input reconciliation: how far
	// each (task, provider) pair has been read, and which external utterance
	// each recorded decision came from.
	cursors        map[string]string
	cursorPath     string
	decisionSource map[string]string
	// PreLease runs immediately before a lease is minted, which is the single
	// point where ownership of a task begins. Reconciliation of newer human
	// input belongs here rather than in any individual launch path, because a
	// rotation or an autonomous pickup would otherwise bypass it. A returned
	// error refuses the lease: if Orchestra cannot establish whether newer
	// human instructions exist, starting a successor from an older intent
	// recreates the exact failure this guards against.
	PreLease func(taskID string) error
}

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"), cursorPath: filepath.Join(dir, "source-cursors.json"), tasks: map[string]domain.Task{}, external: map[string]string{}, activeLeases: map[string]map[string]struct{}{}, quota: map[string]quotaIndex{}, cursors: map[string]string{}, decisionSource: map[string]string{}}
	if err := s.loadCursors(); err != nil {
		return nil, err
	}
	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 == domain.EventHumanDecisionRecorded {
		// A decision records what the human decided. It deliberately mutates
		// no task field: authority is reduced on read by ReduceIntent, never
		// folded into the contract projection.
		var p struct {
			DecisionID string                     `json:"decision_id"`
			Source     domain.HumanDecisionSource `json:"source"`
		}
		if err := json.Unmarshal(e.Payload, &p); err != nil {
			return err
		}
		if p.Source.ExternalID != "" {
			s.decisionSource[p.Source.Provider+"\x00"+p.Source.ExternalID] = p.DecisionID
		}
	}
	if e.Type == "StandupAdvisory" || e.Type == "ApprovalGranted" || e.Type == "ApprovalDenied" {
		return nil
	}
	if e.Type == domain.EventWorkPhaseChanged {
		var p struct {
			Phase       domain.WorkPhase `json:"phase"`
			ArtifactRef string           `json:"artifact_ref"`
			ResultSHA   string           `json:"result_sha"`
		}
		if err := json.Unmarshal(e.Payload, &p); err != nil {
			return err
		}
		// The artifact is attributed to the phase being left, not the one
		// being entered: research seals research, plan seals plan.
		switch t.WorkPhase {
		case domain.WorkPhaseResearch:
			if p.ArtifactRef != "" {
				t.ResearchRef = p.ArtifactRef
			}
		case domain.WorkPhasePlan:
			if p.ArtifactRef != "" {
				t.PlanRef = p.ArtifactRef
			}
		}
		t.WorkPhase = p.Phase
		if p.ResultSHA != "" {
			t.ReviewTargetSHA = p.ResultSHA
		}
		t.Version = e.Version
		s.replaceTask(e.TaskID, t)
		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"
		}
		if m, ok := p["decision_request"].(map[string]any); ok {
			req := domain.DecodeDecisionRequest(m)
			t.DecisionRequest = &req
		}
	case domain.EventTaskChangesRequested:
		// Back to the queue. The submission stays on the task as history: it
		// records that this commit was reviewed, submitted, and rejected.
		t.State = domain.StateQueued
		t.Lease = nil
		t.LifecyclePhase = "changes_requested"
		t.Version = e.Version
		s.replaceTask(e.TaskID, t)
		return nil
	case domain.EventTaskSubmitted:
		var sp struct {
			ResultSHA string             `json:"result_sha"`
			RemoteRef string             `json:"remote_ref"`
			PR        domain.ExternalRef `json:"pr"`
			GateRef   string             `json:"gate_ref"`
			ReviewRef string             `json:"review_ref"`
			PacketRef string             `json:"packet_ref"`
		}
		if err := json.Unmarshal(e.Payload, &sp); err != nil {
			return err
		}
		t.Submission = &domain.SubmissionRef{ResultSHA: sp.ResultSHA, RemoteRef: sp.RemoteRef, PR: sp.PR, GateRef: sp.GateRef, ReviewRef: sp.ReviewRef, PacketRef: sp.PacketRef}
		// In review, not complete. The human owns what happens next, and the
		// lease is released because no agent is working on this any more.
		t.State = domain.StateInReview
		t.Lease = nil
		t.LifecyclePhase = "submitted"
		t.Version = e.Version
		s.replaceTask(e.TaskID, t)
		return nil
	case domain.EventReviewRecorded:
		var rp struct {
			ArtifactRef string `json:"artifact_ref"`
			ResultSHA   string `json:"result_sha"`
			Blocking    int    `json:"blocking"`
		}
		if err := json.Unmarshal(e.Payload, &rp); err != nil {
			return err
		}
		t.Review = &domain.ReviewRef{ArtifactRef: rp.ArtifactRef, ResultSHA: rp.ResultSHA, Blocking: rp.Blocking}
		t.Version = e.Version
		s.replaceTask(e.TaskID, t)
		return nil
	case domain.EventDeferredFindingRecorded:
		// Recorded in the log, projected onto nothing. A deferred finding must
		// not reach agent context, or deferring it would cost what acting on
		// it costs.
		t.Version = e.Version
		s.replaceTask(e.TaskID, t)
		return nil
	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
			}
		}
		// Retry recovery: the router treats Attempt as terminal once it
		// reaches MaxAttempts, and no other event lowers it. See
		// operations.RetryTask, which is the only intended producer.
		if v, ok := p["attempt"].(float64); ok {
			t.Attempt = int(v)
		}
		if v, ok := p["next_retry_at"].(string); ok {
			if v == "" {
				t.NextRetryAt = time.Time{}
			} else if d, err := time.Parse(time.RFC3339, v); err == nil {
				t.NextRetryAt = d
			}
		}
		if v, ok := p["failure_class"].(string); ok {
			t.FailureClass = v
		}
	}
	// A question only stands while the task is blocked on it. Afterwards the
	// answer is an ordinary standing decision and the log still holds the
	// question, so keeping it on the task would put it in every later context.
	if t.State != domain.StateBlocked {
		t.DecisionRequest = 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
		}
	}
	if e.Type == domain.EventHumanDecisionRecorded {
		var p struct {
			Source domain.HumanDecisionSource `json:"source"`
		}
		if err := json.Unmarshal(e.Payload, &p); err != nil {
			return err
		}
		// One external utterance yields one decision, forever. This is what
		// makes a lost cursor write harmless: the refetch is rejected here
		// instead of becoming a second copy of the same instruction.
		if p.Source.ExternalID != "" {
			if _, ok := s.decisionSource[p.Source.Provider+"\x00"+p.Source.ExternalID]; ok {
				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 == domain.EventWorkPhaseChanged {
		var p struct {
			Phase       domain.WorkPhase `json:"phase"`
			ArtifactRef string           `json:"artifact_ref"`
		}
		if err := json.Unmarshal(e.Payload, &p); err != nil {
			return err
		}
		if !taskExists {
			return domain.ErrNotFound
		}
		if !domain.CanTransitionPhase(t.WorkPhase, p.Phase) {
			return fmt.Errorf("%w: cannot move from work phase %q to %q", domain.ErrInvalid, t.WorkPhase, p.Phase)
		}
		// Leaving research or plan without sealing the artifact would hand the
		// next phase a conversation to reconstruct instead of a result to read.
		if (t.WorkPhase == domain.WorkPhaseResearch || t.WorkPhase == domain.WorkPhasePlan) && p.ArtifactRef == "" {
			return fmt.Errorf("%w: leaving work phase %q requires a sealed artifact_ref", domain.ErrInvalid, t.WorkPhase)
		}
		if p.ArtifactRef != "" {
			if _, err := s.Artifact(p.ArtifactRef); err != nil {
				return fmt.Errorf("%w: missing artifact %s", domain.ErrInvalid, p.ArtifactRef)
			}
		}
	}
	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 == domain.EventReviewRecorded {
		var p struct {
			ArtifactRef string `json:"artifact_ref"`
		}
		if err := json.Unmarshal(e.Payload, &p); err != nil {
			return err
		}
		if _, err := s.Artifact(p.ArtifactRef); err != nil {
			return fmt.Errorf("%w: missing artifact %s", domain.ErrInvalid, p.ArtifactRef)
		}
	}
	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", domain.EventTaskSubmitted:
		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.
// QuotaSince sums the receipts at or after `since`. The event log is
// Orchestra's complete accounting source, so a window holding no receipts is
// observable zero consumption, not missing data: it reports known. `known` is
// false only when a receipt inside the window said its own consumption was
// unknown. Reporting an empty window as unknown deadlocked admission — a
// harness with a configured quota limit and no receipt history could never be
// leased, and the only producer of a receipt is a completed lease. Found on
// burn-in run 2, 2026-08-26.
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, true
	}
	start := sort.Search(len(index.records), func(i int) bool { return !index.records[i].At.Before(since) })
	if start == len(index.records) {
		return 0, true
	}
	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
}

// EffectiveIntent reduces the task's contract plus its human-decision events
// into the standing authority for the task. It is the only sanctioned answer
// to "what has the human most recently decided?" — a caller must never read
// that from handoff prose. The reduction is over the whole log under one lock,
// so it cannot observe a decision appended without its task projection.
func (s *Store) EffectiveIntent(id string) (domain.EffectiveIntent, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	t, ok := s.tasks[id]
	if !ok {
		return domain.EffectiveIntent{}, domain.ErrNotFound
	}
	return domain.ReduceIntent(t, s.events)
}

// 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)
	}
	// Ownership begins here, so reconciliation happens here. Every launch and
	// every resume is downstream of a TaskLeased event, and Store.Lease is the
	// only place one is minted.
	if s.PreLease != nil {
		if err := s.PreLease(id); err != nil {
			return domain.Event{}, fmt.Errorf("reconcile human input: %w", err)
		}
	}
	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)
}