v3 workflow: intent, phases, review, submission, enforcement, burn-in

The v3 stack, previously an uncommitted working tree, plus this session's two units and the burn-in instrument. This commit is the burn-in build identity: coordinator and worker must both report this revision before a task is created. Workflow (earlier sessions, uncommitted until now): human decision events and reduction, source cursors and reconcile-before-launch, turn-boundary reconciliation, internal/agentctx as the single renderer, ace-fca phases with sealed artifacts, the trajectory gate, bounded grilling, independent review, task pr enforcement, and human review reflection. Capability restrictions at the agent boundary: an authz.Agent surface at GatedWrite may ask and may not act. It also fixes two bugs the unit exposed -- gated surfaces could not reach the two endpoints written for them, and RequestHumanDecision would block an unowned task while rejecting a question from the session that did own it. Turn-boundary reconcile-failure escalation: a streak of consecutive failures asks the session to hand off, fenced on the lease epoch, with reconcile_failure as a real handoff reason. The worker was dropping the coordinator's verdict on the floor; it now acts on it. Burn-in: herdr.WriteLaunchContext dumps the exact agentctx.Build result to <worktree>/.orchestra/launch.md at every launch, local and federated. BURNIN.md is the runbook. deploy/build.sh stamps both binaries from one commit. go build, go vet and go test ./... pass, 20 packages. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-26 18:31:20 +04:00
parent 97a9c65302
commit 7f12c7fc37
78 changed files with 16417 additions and 352 deletions
@@ -0,0 +1,128 @@
+package store
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+
+	"orchestra/internal/domain"
+)
+
+// SourceCursor is how far a task has been reconciled against one external
+// human-input source. Its meaning belongs to the provider: a Gitea comment
+// id, a Vikunja activity id, a web command sequence. Orchestra only requires
+// that the provider can resume from it.
+//
+// The cursor is an efficiency bound, never the correctness guarantee. A
+// cursor that fails to persist after a decision was appended must not create
+// a second decision, so provenance uniqueness on (provider, external_id) is
+// what actually prevents duplicates. See Store.DecisionForSource.
+type SourceCursor struct {
+	TaskID   string `json:"task_id"`
+	Provider string `json:"provider"`
+	Cursor   string `json:"cursor"`
+}
+
+func cursorKey(taskID, provider string) string { return taskID + "\x00" + provider }
+
+func (s *Store) SourceCursor(taskID, provider string) (SourceCursor, bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	v, ok := s.cursors[cursorKey(taskID, provider)]
+	if !ok {
+		return SourceCursor{TaskID: taskID, Provider: provider}, false
+	}
+	return SourceCursor{TaskID: taskID, Provider: provider, Cursor: v}, true
+}
+
+// SetSourceCursor persists the cursor before returning. A caller must only
+// advance it after every event it derived from that input is durable.
+func (s *Store) SetSourceCursor(c SourceCursor) error {
+	if strings.TrimSpace(c.TaskID) == "" || strings.TrimSpace(c.Provider) == "" {
+		return fmt.Errorf("%w: cursor needs task_id and provider", domain.ErrInvalid)
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	prior, had := s.cursors[cursorKey(c.TaskID, c.Provider)]
+	s.cursors[cursorKey(c.TaskID, c.Provider)] = c.Cursor
+	if err := s.writeCursorsLocked(); err != nil {
+		if had {
+			s.cursors[cursorKey(c.TaskID, c.Provider)] = prior
+		} else {
+			delete(s.cursors, cursorKey(c.TaskID, c.Provider))
+		}
+		return err
+	}
+	return nil
+}
+
+// DecisionForSource resolves the decision already recorded for one external
+// human utterance, so a refetch after a lost cursor write is a skip rather
+// than a second decision.
+func (s *Store) DecisionForSource(provider, externalID string) (string, bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	id, ok := s.decisionSource[provider+"\x00"+externalID]
+	return id, ok
+}
+
+func (s *Store) writeCursorsLocked() error {
+	keys := make([]string, 0, len(s.cursors))
+	for k := range s.cursors {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	out := make([]SourceCursor, 0, len(keys))
+	for _, k := range keys {
+		task, provider, _ := strings.Cut(k, "\x00")
+		out = append(out, SourceCursor{TaskID: task, Provider: provider, Cursor: s.cursors[k]})
+	}
+	b, err := json.Marshal(out)
+	if err != nil {
+		return err
+	}
+	tmp := s.cursorPath + ".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.cursorPath); err != nil {
+		return err
+	}
+	dir, err := os.Open(filepath.Dir(s.cursorPath))
+	if err != nil {
+		return err
+	}
+	defer dir.Close()
+	return dir.Sync()
+}
+
+func (s *Store) loadCursors() error {
+	b, err := os.ReadFile(s.cursorPath)
+	if os.IsNotExist(err) {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	var in []SourceCursor
+	if err := json.Unmarshal(b, &in); err != nil {
+		return fmt.Errorf("source cursors: %w", err)
+	}
+	for _, c := range in {
+		s.cursors[cursorKey(c.TaskID, c.Provider)] = c.Cursor
+	}
+	return nil
+}
@@ -0,0 +1,99 @@
+package store
+
+import (
+	"encoding/json"
+	"errors"
+	"testing"
+	"time"
+
+	"orchestra/internal/authz"
+	"orchestra/internal/domain"
+)
+
+func decisionPayload(t *testing.T, id, kind, subject, value string, supersedes ...string) []byte {
+	t.Helper()
+	p := map[string]any{
+		"decision_id": id, "kind": kind, "subject": subject, "value": value,
+		"source": map[string]any{"provider": "gitea", "external_id": "issue-1#c7"},
+	}
+	if len(supersedes) > 0 {
+		p["supersedes"] = supersedes
+	}
+	b, err := json.Marshal(p)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return b
+}
+
+// A decision must land while the task is leased — that is the whole point,
+// since the human corrects work already in flight — without touching task
+// state or the current lease.
+func TestDecisionAppendsUnderLiveLeaseWithoutDisturbingProjection(t *testing.T) {
+	dir := t.TempDir()
+	s, err := Open(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := s.Append(created("create")); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.Lease("task-1", "h1", time.Minute); err != nil {
+		t.Fatal(err)
+	}
+	before, _ := s.Task("task-1")
+
+	e := domain.Event{
+		ID: "e-d1", Type: domain.EventHumanDecisionRecorded, TaskID: "task-1",
+		Version: before.Version + 1, At: time.Now().UTC(),
+		Payload: decisionPayload(t, "d1", "correction", "strategy", "use b"),
+		Surface: string(authz.Web), SchemaVersion: domain.CurrentEventSchema,
+	}
+	if err := s.Append(e); err != nil {
+		t.Fatalf("decision rejected under live lease: %v", err)
+	}
+	after, _ := s.Task("task-1")
+	if after.State != before.State {
+		t.Fatalf("state changed %s -> %s", before.State, after.State)
+	}
+	if after.Lease == nil || *after.Lease != *before.Lease {
+		t.Fatalf("lease changed: %+v -> %+v", before.Lease, after.Lease)
+	}
+	if after.Description != before.Description || after.LifecyclePhase != before.LifecyclePhase {
+		t.Fatalf("contract fields changed: %+v", after)
+	}
+
+	intent, err := s.EffectiveIntent("task-1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(intent.Decisions) != 1 || intent.Decisions[0].Value != "use b" {
+		t.Fatalf("standing set = %+v", intent.Decisions)
+	}
+	if intent.Decisions[0].Source.ExternalID != "issue-1#c7" {
+		t.Fatalf("provenance lost: %+v", intent.Decisions[0].Source)
+	}
+
+	// Same answer after a restart replay, from the log alone.
+	reopened, err := Open(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	replayed, err := reopened.EffectiveIntent("task-1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(replayed.Decisions) != 1 || replayed.Decisions[0].ID != "d1" {
+		t.Fatalf("replayed standing set = %+v", replayed.Decisions)
+	}
+}
+
+func TestEffectiveIntentUnknownTask(t *testing.T) {
+	s, err := Open(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.EffectiveIntent("nope"); !errors.Is(err, domain.ErrNotFound) {
+		t.Fatalf("want ErrNotFound, got %v", err)
+	}
+}
@@ -43,13 +43,30 @@ type Store 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"), tasks: map[string]domain.Task{}, external: map[string]string{}, activeLeases: map[string]map[string]struct{}{}, quota: map[string]quotaIndex{}}
+	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
 	}
@@ -230,9 +247,53 @@ func (s *Store) apply(e domain.Event) error {
 		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 {
@@ -354,6 +415,60 @@ func (s *Store) apply(e domain.Event) error {
 		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
@@ -391,6 +506,12 @@ func (s *Store) apply(e domain.Event) error {
 			}
 		}
 	}
+	// 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
 	}
@@ -526,6 +647,22 @@ func (s *Store) Append(e domain.Event) error {
 			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
@@ -555,6 +692,31 @@ func (s *Store) Append(e domain.Event) error {
 			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)
@@ -574,6 +736,17 @@ func (s *Store) Append(e domain.Event) error {
 	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)
@@ -638,7 +811,7 @@ func (s *Store) validateTransition(e domain.Event, t domain.Task, exists bool, p
 		return nil
 	}
 	switch e.Type {
-	case "TaskLeaseRenewed", "TaskLaunchAcknowledged", "TaskReleased", "TaskPickupValidated", "TaskCompleted", "TaskBlocked", "TaskNeedsAttention", "TaskFailed":
+	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
@@ -824,6 +997,21 @@ func (s *Store) Task(id string) (domain.Task, bool) {
 	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) {
@@ -841,6 +1029,14 @@ func (s *Store) Lease(id, harness string, ttl time.Duration) (domain.Event, erro
 	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