orchestra/cmd/orchestra-worker/phase_test.go

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 refusal is an answer. The agent is told why, in the same confirmed
// delivery path every other Orchestra-originated input uses, and the request
// is cleared so it can write a corrected one instead of resending the same
// rejected file at every boundary.
func TestRefusedPhaseRequestIsAnsweredAndCleared(t *testing.T) {
	w, backend, wt, done := phaseWorker(t, func(rw http.ResponseWriter, r *http.Request) {
		if r.URL.Path == "/v1/federation/phase" {
			http.Error(rw, "phase request refused: task may only move to \"research\", not \"implement\"", http.StatusConflict)
			return
		}
		rw.Write([]byte(`{"verdict":"continue"}`))
	})
	defer done()
	writeRequest(t, wt, domain.WorkPhaseFrame, domain.WorkPhaseImplement)

	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 len(backend.prompts) == 0 || !strings.Contains(backend.prompts[0], `may only move to "research"`) {
		t.Fatalf("the agent was not told why: %q", backend.prompts)
	}
	if _, err := os.Stat(filepath.Join(wt, ".orchestra", phaseRequestFile)); !os.IsNotExist(err) {
		t.Fatal("the refused request survived, so the agent will resend it")
	}
}

// A coordinator that cannot be reached has not refused anything. Telling the
// agent its request was rejected would be a lie, and dropping the file would
// lose a request that is still valid.
func TestTransientPhaseFailureKeepsTheRequest(t *testing.T) {
	w, backend, wt, done := phaseWorker(t, func(rw http.ResponseWriter, r *http.Request) {
		if r.URL.Path == "/v1/federation/phase" {
			http.Error(rw, "upstream down", http.StatusServiceUnavailable)
			return
		}
		rw.Write([]byte(`{"verdict":"continue"}`))
	})
	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 transient failure discarded the request")
	}
	for _, p := range backend.prompts {
		if strings.Contains(p, "refused") {
			t.Fatalf("a transient failure was reported to the agent as a refusal: %q", p)
		}
	}
}

// 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))
	}
}

// The bug that made run 4 stall exactly like run 3. rotationTick returned
// early for the claude harness before reaching the turn boundary, so
// federatedTurn had one call site that this harness never took. Phase requests
// were never read and human decisions were never delivered on the harness both
// burn-in runs actually used.
//
// Claude still skips the occupancy state machine below that branch, because it
// owns its own context rollover. A turn boundary is not a rotation.
func TestClaudeHarnessReachesTheTurnBoundary(t *testing.T) {
	reached := make(chan string, 4)
	w, backend, wt, done := phaseWorker(t, func(rw http.ResponseWriter, r *http.Request) {
		reached <- r.URL.Path
		if r.URL.Path == "/v1/federation/phase" {
			_ = json.NewEncoder(rw).Encode(map[string]string{"phase": "research"})
			return
		}
		rw.Write([]byte(`{"verdict":"continue"}`))
	})
	defer done()
	writeRequest(t, wt, domain.WorkPhaseFrame, domain.WorkPhaseResearch)

	// rotationTick, not federatedTurn: the dead path was the route in.
	w.rotationTick(context.Background(), "task", w.sessions["task"])

	var saw bool
	for len(reached) > 0 {
		if <-reached == "/v1/federation/phase" {
			saw = true
		}
	}
	if !saw {
		t.Fatal("the claude harness never reached the phase boundary")
	}
	if s := w.sessions["task"]; !s.HandoffRequested || s.HandoffReason != "phase_changed" {
		t.Fatalf("session did not rotate: %+v", s)
	}
	_ = backend
}