Give the phase brief a protocol, and end the session it advances

The brief told the agent to ask for a phase change and never carried the asking. The agent asked in prose, no code represented the request, and the session idled until its lease expired. That is what failed run 3. F21. The agent asks with .orchestra/phase-request.json, and seals research.json or plan.json where the phase it is leaving produces one. At a verified turn boundary the worker checks the phase belief, the transition and the artifact, then calls the coordinator with its lease epoch and a derived operation id. AdvanceWorkPhase is unchanged, so a request cannot reach a move the operator surface could not also make. Redelivery is idempotent. F22. A session now records the phase it was launched to run. One that no longer matches its task rotates with reason phase_changed, whether this worker asked for the change or an operator made it. F20. CLIAdapter.prompt sent handoff and rotation prompts without confirming them, which is the failure F20 exists to catch. Fixed at the shared call site. F23 needed no change. Issue comments already become decisions with no submission, through Reconciler.Reconcile at PreLease and at every turn boundary. The earlier finding searched internal/operations alone and was wrong. Tests now cover the boundary it turns on. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011xsXyr5J1RACo71YeKG3Pu
2026-08-27 16:35:37 +04:00
parent 15408a5463
commit 1f5bf7e66e
14 changed files with 1380 additions and 2 deletions
@@ -354,6 +354,9 @@ func (w *worker) start(ctx context.Context, t domain.Task, ref string) error {
 	if err != nil {
 		return err
 	}
+	// The phase this session was launched to run. A later phase change makes
+	// this session's context the wrong one, which is what rotates it (F22).
+	s.Phase = string(currentPhase(t))
 	s.TaskFileSHA = taskHash(t)
 	if w.harness == "claude" {
 		s.ContextHandoffSHA, _ = fileSHA256(filepath.Join(wt, "HANDOFF.md"))
@@ -1608,6 +1611,19 @@ func (w *worker) federatedTurn(ctx context.Context, id string, a herdr.Adapter,
 	if !at {
 		return
 	}
+	// A phase this session no longer runs ends it, whether this worker asked
+	// for the change or an operator made it (F22). Checked before the request
+	// below so a session cannot advance a phase twice.
+	if w.phaseChanged(id, s) {
+		w.rotateForPhase(ctx, id, a, s)
+		return
+	}
+	// The agent asks for a phase change here, at a boundary it has reached
+	// (F21). Orchestra decides, and an accepted change ends this session.
+	if w.requestPhase(ctx, id, s) {
+		w.rotateForPhase(ctx, id, a, s)
+		return
+	}
 	answer, err := w.api.Turn(ctx, id, l.Epoch, verdict, s.DeliveredDecisions)
 	if err != nil {
 		// Observable, not fatal. A coordinator that cannot be reached does not
@@ -1648,6 +1664,148 @@ func (w *worker) federatedTurn(ctx context.Context, id string, a herdr.Adapter,
 	_ = w.save()
 }

+// phaseRequestFile is the agent's bounded phase-change intent (F21). Prose in
+// the pane is not a request: matching on it would make the protocol depend on
+// wording the agent is free to vary, and on Orchestra reading its own echo.
+const phaseRequestFile = "phase-request.json"
+
+// phaseRequest is what the agent writes. It states the phase it believes it
+// is in as well as the one it wants, so a request written from a stale
+// context is refused rather than applied to whatever phase is current.
+type phaseRequest struct {
+	From domain.WorkPhase `json:"from"`
+	To   domain.WorkPhase `json:"to"`
+}
+
+// phaseArtifact names the sealed output each phase must produce before it may
+// be left. Phases absent from this table seal nothing.
+var phaseArtifact = map[domain.WorkPhase]string{
+	domain.WorkPhaseResearch: "research.json",
+	domain.WorkPhasePlan:     "plan.json",
+}
+
+func currentPhase(t domain.Task) domain.WorkPhase {
+	if t.WorkPhase == "" {
+		return domain.WorkPhaseFrame
+	}
+	return t.WorkPhase
+}
+
+// phaseChanged reports whether this session is running a phase the task has
+// since left. It covers a change this worker requested and one an operator
+// made through the coordinator equally, because both leave the same evidence:
+// a session whose context was built for a phase that is no longer current.
+func (w *worker) phaseChanged(id string, s herdr.Session) bool {
+	t, ok := w.tasks[id]
+	if !ok || s.Phase == "" {
+		return false
+	}
+	return string(currentPhase(t)) != s.Phase
+}
+
+// rotateForPhase ends the current cognitive session because the phase moved
+// (F22). A phase change is a change of context, not of instruction: leaving
+// the old agent running would either waste the lease waiting for it to idle
+// out, as run 3 did, or let it keep working under a brief that no longer
+// applies.
+func (w *worker) rotateForPhase(ctx context.Context, id string, a herdr.Adapter, s herdr.Session) {
+	if s.HandoffRequested {
+		return
+	}
+	requester, ok := a.(herdr.ReasonedHandoffRequester)
+	if !ok {
+		w.recordError(fmt.Errorf("phase rotation %s: adapter cannot state a reason", id))
+		return
+	}
+	if err := requester.RequestHandoffReason(ctx, s, "phase_changed", nil); err != nil {
+		w.recordError(fmt.Errorf("phase rotation %s: %w", id, err))
+		return
+	}
+	s.HandoffRequested, s.HandoffReason = true, "phase_changed"
+	w.sessions[id] = s
+	_ = w.save()
+	log.Printf("phase changed for %s: session rotating", id)
+}
+
+// requestPhase carries an agent's phase request to the coordinator (F21).
+// It reports whether the phase moved.
+//
+// Everything checkable locally is checked before the call, so an agent that
+// asked for the wrong thing learns it from a recorded error rather than from
+// a lease that quietly stops being renewed.
+func (w *worker) requestPhase(ctx context.Context, id string, s herdr.Session) bool {
+	path := filepath.Join(s.Worktree, ".orchestra", phaseRequestFile)
+	b, err := os.ReadFile(path)
+	if err != nil {
+		return false
+	}
+	var req phaseRequest
+	if err := json.Unmarshal(b, &req); err != nil {
+		w.recordError(fmt.Errorf("phase request %s: %w", id, err))
+		return false
+	}
+	t, ok := w.tasks[id]
+	if !ok {
+		return false
+	}
+	if req.From != currentPhase(t) {
+		w.recordError(fmt.Errorf("phase request %s: task is in work phase %q, not %q", id, currentPhase(t), req.From))
+		return false
+	}
+	if !domain.CanTransitionPhase(req.From, req.To) {
+		w.recordError(fmt.Errorf("phase request %s: %q to %q is not a legal transition", id, req.From, req.To))
+		return false
+	}
+	// The phase being left seals its result before it may be left. Decoding
+	// here means a malformed artifact is reported against the agent that
+	// wrote it, while its session is still alive to be told.
+	var artifact []byte
+	if name := phaseArtifact[req.From]; name != "" {
+		artifact, err = os.ReadFile(filepath.Join(s.Worktree, ".orchestra", name))
+		if err != nil {
+			w.recordError(fmt.Errorf("phase request %s: work phase %q must seal .orchestra/%s first: %w", id, req.From, name, err))
+			return false
+		}
+		switch req.From {
+		case domain.WorkPhaseResearch:
+			if _, decErr := workphase.DecodeResearch(artifact); decErr != nil {
+				w.recordError(fmt.Errorf("phase request %s: research artifact: %w", id, decErr))
+				return false
+			}
+		case domain.WorkPhasePlan:
+			if _, decErr := workphase.DecodePlan(artifact); decErr != nil {
+				w.recordError(fmt.Errorf("phase request %s: plan artifact: %w", id, decErr))
+				return false
+			}
+		}
+	}
+	l := w.leases[id]
+	// Derived, not random: a redelivery after a lost response must carry the
+	// same id so the coordinator recognises it instead of advancing twice.
+	op := "phase:" + id + ":" + l.Epoch + ":" + string(req.From) + ":" + string(req.To)
+	phase, err := w.api.AdvancePhase(ctx, id, l.Epoch, op, req.From, req.To, artifact)
+	if err != nil {
+		w.recordError(fmt.Errorf("phase request %s: %w", id, err))
+		return false
+	}
+	if phase == "" {
+		// Accepted but not advanced: the coordinator raised a trajectory gate
+		// and the human now owns the move. Keep the request so the same ask is
+		// re-sent, under the same operation id, once the gate clears.
+		return false
+	}
+	// Durable before the file is removed. A removal that raced the response
+	// would lose the request and leave the agent waiting on an answer that
+	// already arrived.
+	if err := os.Remove(path); err != nil {
+		w.recordError(fmt.Errorf("phase request %s: %w", id, err))
+	}
+	t.WorkPhase = phase
+	w.tasks[id] = t
+	log.Printf("phase request %s accepted: %s to %s", id, req.From, phase)
+	return true
+}
+
 // sendPrompt delivers Orchestra-originated input and confirms the harness took
 // it. A phase continuation or a decision notice whose Enter is lost strands the
 // session exactly as a lost launch does.
@@ -0,0 +1,292 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"orchestra/internal/domain"
+	"orchestra/internal/federation"
+	"orchestra/internal/herdr"
+	"orchestra/internal/workphase"
+)
+
+// phaseWorker builds a worker whose single session sits at a turn boundary,
+// with a worktree the agent can write its request into.
+func phaseWorker(t *testing.T, handler http.HandlerFunc) (*worker, *recordingBackend, string, func()) {
+	t.Helper()
+	api := httptest.NewServer(handler)
+	wt := t.TempDir()
+	if err := os.MkdirAll(filepath.Join(wt, ".orchestra"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	backend := &recordingBackend{status: "idle"}
+	w := &worker{
+		api:         federation.Client{BaseURL: api.URL, WorkerID: "h", Token: "t"},
+		backend:     backend,
+		harness:     "claude",
+		sessions:    map[string]herdr.Session{"task": {PaneID: "pane", Worktree: wt, Phase: string(domain.WorkPhaseFrame)}},
+		leases:      map[string]lease{"task": {Epoch: "e1", Version: 1, Until: time.Now().Add(time.Hour)}},
+		tasks:       map[string]domain.Task{"task": {ID: "task", WorkPhase: domain.WorkPhaseFrame}},
+		quarantined: map[string]bool{},
+		statePath:   filepath.Join(t.TempDir(), "state.json"),
+	}
+	return w, backend, wt, api.Close
+}
+
+func writeRequest(t *testing.T, wt string, from, to domain.WorkPhase) {
+	t.Helper()
+	b, _ := json.Marshal(phaseRequest{From: from, To: to})
+	if err := os.WriteFile(filepath.Join(wt, ".orchestra", phaseRequestFile), b, 0o644); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// The whole point of F21: the agent asks with a file, Orchestra answers, and
+// the session that asked is rotated rather than left idling until its lease
+// dies. Run 3 failed conformance because none of this existed.
+func TestPhaseRequestAdvancesAndRotatesTheSession(t *testing.T) {
+	var sent map[string]any
+	w, backend, wt, done := phaseWorker(t, func(rw http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/federation/phase" {
+			t.Errorf("unexpected %s %s", r.Method, r.URL.Path)
+			rw.WriteHeader(http.StatusNotFound)
+			return
+		}
+		_ = json.NewDecoder(r.Body).Decode(&sent)
+		_ = json.NewEncoder(rw).Encode(map[string]string{"phase": "research"})
+	})
+	defer done()
+	writeRequest(t, wt, domain.WorkPhaseFrame, domain.WorkPhaseResearch)
+
+	a := herdr.CLIAdapter{Backend: backend, Harness: "claude"}
+	w.federatedTurn(context.Background(), "task", a, "continue")
+
+	if sent["lease_epoch"] != "e1" || sent["from"] != "frame" || sent["to"] != "research" {
+		t.Fatalf("request = %+v", sent)
+	}
+	if op, _ := sent["operation_id"].(string); op == "" {
+		t.Fatal("request carried no operation id")
+	}
+	// Accepted and rotating. A phase change that left the old session running
+	// is the F22 bug, so this assertion is the test for it.
+	s := w.sessions["task"]
+	if !s.HandoffRequested || s.HandoffReason != "phase_changed" {
+		t.Fatalf("session did not rotate: %+v", s)
+	}
+	if len(backend.prompts) == 0 || !strings.Contains(backend.prompts[len(backend.prompts)-1], "next work phase") {
+		t.Fatalf("agent was not told why it is stopping: %q", backend.prompts)
+	}
+	// Consumed, so the same request is not replayed at the next boundary.
+	if _, err := os.Stat(filepath.Join(wt, ".orchestra", phaseRequestFile)); !os.IsNotExist(err) {
+		t.Fatal("the accepted request file survived")
+	}
+}
+
+// A phase an operator advanced through the coordinator ends the session too.
+// The session's context was built for a phase that is no longer current, and
+// nothing about that depends on who asked.
+func TestExternalPhaseChangeRotatesTheSession(t *testing.T) {
+	w, backend, _, done := phaseWorker(t, func(rw http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/v1/federation/phase" {
+			t.Errorf("an external phase change was re-requested by the worker")
+		}
+		rw.Write([]byte(`{"verdict":"continue"}`))
+	})
+	defer done()
+	task := w.tasks["task"]
+	task.WorkPhase = domain.WorkPhaseResearch
+	w.tasks["task"] = task
+
+	a := herdr.CLIAdapter{Backend: backend, Harness: "claude"}
+	w.federatedTurn(context.Background(), "task", a, "continue")
+
+	if s := w.sessions["task"]; !s.HandoffRequested || s.HandoffReason != "phase_changed" {
+		t.Fatalf("session did not rotate: %+v", s)
+	}
+}
+
+// A phase that seals an artifact must not leave without one. Caught locally so
+// the agent is told while its session is still alive to be told.
+func TestPhaseRequestRefusesAnUnsealedArtifact(t *testing.T) {
+	w, backend, wt, done := phaseWorker(t, func(rw http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/v1/federation/phase" {
+			t.Errorf("an unsealed phase request reached the coordinator")
+		}
+		rw.Write([]byte(`{"verdict":"continue"}`))
+	})
+	defer done()
+	task := w.tasks["task"]
+	task.WorkPhase = domain.WorkPhaseResearch
+	w.tasks["task"] = task
+	w.sessions["task"] = herdr.Session{PaneID: "pane", Worktree: wt, Phase: string(domain.WorkPhaseResearch)}
+	writeRequest(t, wt, domain.WorkPhaseResearch, domain.WorkPhasePlan)
+
+	a := herdr.CLIAdapter{Backend: backend, Harness: "claude"}
+	w.federatedTurn(context.Background(), "task", a, "continue")
+
+	if s := w.sessions["task"]; s.HandoffRequested {
+		t.Fatal("a refused request rotated the session")
+	}
+	if !strings.Contains(w.lastError, "research.json") {
+		t.Fatalf("lastError = %q", w.lastError)
+	}
+}
+
+// A sealed artifact travels with the request, so the next phase reads a result
+// instead of reconstructing a conversation.
+func TestPhaseRequestCarriesTheSealedArtifact(t *testing.T) {
+	var sent struct {
+		Artifact []byte `json:"artifact"`
+	}
+	w, backend, wt, done := phaseWorker(t, func(rw http.ResponseWriter, r *http.Request) {
+		_ = json.NewDecoder(r.Body).Decode(&sent)
+		_ = json.NewEncoder(rw).Encode(map[string]string{"phase": "plan"})
+	})
+	defer done()
+	task := w.tasks["task"]
+	task.WorkPhase = domain.WorkPhaseResearch
+	w.tasks["task"] = task
+	w.sessions["task"] = herdr.Session{PaneID: "pane", Worktree: wt, Phase: string(domain.WorkPhaseResearch)}
+
+	sealed, err := workphase.Encode(workphase.Research{Findings: []workphase.Finding{{Claim: "c", Evidence: "e"}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(wt, ".orchestra", "research.json"), sealed, 0o644); err != nil {
+		t.Fatal(err)
+	}
+	writeRequest(t, wt, domain.WorkPhaseResearch, domain.WorkPhasePlan)
+
+	a := herdr.CLIAdapter{Backend: backend, Harness: "claude"}
+	w.federatedTurn(context.Background(), "task", a, "continue")
+
+	if _, err := workphase.DecodeResearch(sent.Artifact); err != nil {
+		t.Fatalf("artifact did not arrive sealed: %v", err)
+	}
+}
+
+// A malformed artifact is the agent's mistake, and it must not be sealed into
+// the log as the phase's accepted result.
+func TestPhaseRequestRefusesAMalformedArtifact(t *testing.T) {
+	w, backend, wt, done := phaseWorker(t, func(rw http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/v1/federation/phase" {
+			t.Errorf("a malformed artifact reached the coordinator")
+		}
+		rw.Write([]byte(`{"verdict":"continue"}`))
+	})
+	defer done()
+	task := w.tasks["task"]
+	task.WorkPhase = domain.WorkPhaseResearch
+	w.tasks["task"] = task
+	w.sessions["task"] = herdr.Session{PaneID: "pane", Worktree: wt, Phase: string(domain.WorkPhaseResearch)}
+	if err := os.WriteFile(filepath.Join(wt, ".orchestra", "research.json"), []byte(`{"findings":[]}`), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	writeRequest(t, wt, domain.WorkPhaseResearch, domain.WorkPhasePlan)
+
+	a := herdr.CLIAdapter{Backend: backend, Harness: "claude"}
+	w.federatedTurn(context.Background(), "task", a, "continue")
+
+	if s := w.sessions["task"]; s.HandoffRequested {
+		t.Fatal("a malformed artifact rotated the session")
+	}
+}
+
+// A refused request keeps the pane's own state out of it: the file stays, so
+// the same ask is retried under the same operation id once the reason clears.
+func TestRefusedPhaseRequestIsRetried(t *testing.T) {
+	calls := 0
+	w, backend, wt, done := phaseWorker(t, func(rw http.ResponseWriter, r *http.Request) {
+		calls++
+		if calls == 1 {
+			http.Error(rw, "conflict", http.StatusConflict)
+			return
+		}
+		_ = json.NewEncoder(rw).Encode(map[string]string{"phase": "research"})
+	})
+	defer done()
+	writeRequest(t, wt, domain.WorkPhaseFrame, domain.WorkPhaseResearch)
+	a := herdr.CLIAdapter{Backend: backend, Harness: "claude"}
+
+	w.federatedTurn(context.Background(), "task", a, "continue")
+	if _, err := os.Stat(filepath.Join(wt, ".orchestra", phaseRequestFile)); err != nil {
+		t.Fatal("a refused request must survive for the retry")
+	}
+	w.federatedTurn(context.Background(), "task", a, "continue")
+	if s := w.sessions["task"]; !s.HandoffRequested {
+		t.Fatalf("the retry did not advance the phase: %+v", s)
+	}
+}
+
+// confirmingBackend fails its first confirmation and accepts the next, which
+// is the shape of a submit that never reached the harness.
+type confirmingBackend struct {
+	recordingBackend
+	failures int
+}
+
+func (b *confirmingBackend) ConfirmInput(context.Context, herdr.Session, string) (string, error) {
+	if b.failures > 0 {
+		b.failures--
+		return "", errPromptNotSubmitted
+	}
+	return "input editor cleared", nil
+}
+
+var errPromptNotSubmitted = &confirmError{}
+
+type confirmError struct{}
+
+func (*confirmError) Error() string { return "prompt_not_submitted" }
+
+// F20's guarantee is not that a send was attempted, it is that the harness
+// took it. A decision whose Enter was lost must stay undelivered, so the same
+// correction is sent again at the next boundary rather than being recorded as
+// shown to an agent that never saw it.
+func TestDecisionNoticeStaysUndeliveredUntilConfirmed(t *testing.T) {
+	api := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
+		var body struct {
+			Delivered []string `json:"delivered_decisions"`
+		}
+		_ = json.NewDecoder(r.Body).Decode(&body)
+		out := federation.TurnDecision{Verdict: "continue"}
+		if len(body.Delivered) == 0 {
+			out.Decisions = []domain.HumanDecision{{
+				ID: "d1", Kind: domain.HumanDecisionCorrection, Subject: "strategy", Value: "no, use b",
+			}}
+		}
+		_ = json.NewEncoder(rw).Encode(out)
+	}))
+	defer api.Close()
+
+	backend := &confirmingBackend{recordingBackend: recordingBackend{status: "idle"}, failures: 1}
+	w := &worker{
+		api:       federation.Client{BaseURL: api.URL, WorkerID: "h", Token: "t"},
+		backend:   backend,
+		harness:   "claude",
+		sessions:  map[string]herdr.Session{"task": {PaneID: "pane"}},
+		leases:    map[string]lease{"task": {Epoch: "e1", Version: 2}},
+		tasks:     map[string]domain.Task{"task": {ID: "task"}},
+		statePath: filepath.Join(t.TempDir(), "state.json"),
+	}
+
+	w.federatedTurn(context.Background(), "task", boundaryAdapter{at: true}, "continue")
+	if ids := w.sessions["task"].DeliveredDecisions; len(ids) != 0 {
+		t.Fatalf("an unconfirmed correction was recorded as delivered: %v", ids)
+	}
+
+	w.federatedTurn(context.Background(), "task", boundaryAdapter{at: true}, "continue")
+	if ids := w.sessions["task"].DeliveredDecisions; len(ids) != 1 || ids[0] != "d1" {
+		t.Fatalf("delivered ids = %v", ids)
+	}
+	if len(backend.prompts) != 2 {
+		t.Fatalf("sends = %d, want the correction retried once", len(backend.prompts))
+	}
+}