diff --git a/BURNIN.md b/BURNIN.md index 6c48ec0..c1c0b97 100644 --- a/BURNIN.md +++ b/BURNIN.md @@ -610,3 +610,324 @@ was. Both halves report `edff021`, built from a detached worktree of that revision. Worker sha256 `68ac265455acb0e007bfb0e898a91b317cd7289d7f2bf5da6df1224e1d6706a5` at `/usr/local/bin/orchestra-worker`. Coordinator image built 13:08:30 +0400. + +## Run 3: autonomous expiry and relaunch, 2026-08-27 11:02 UTC + +Task `06G44JZB80MZBEY97196EZN8EC`, issue `kami/test-e2e#3`. Observed live, no +operator action in the window. The stranded epoch was +`06G44ZBZX4YH6PRN7Y4ZH8GG3W`. + +| seq | at (UTC) | event | epoch | +|---|---|---|---| +| 358 | 10:32:09 | `TaskLeaseRenewed` v15 | `06G44ZBZX4YH6PRN7Y4ZH8GG3W` | +| 359 | 11:02:16 | `TaskReleased` v16, `reason: lease_expired` | `06G44ZBZX4YH6PRN7Y4ZH8GG3W` | +| 360 | 11:03:17 | `TaskLeased` v17 | `06G45RVKAW20H1D22NHBRYKMN8` | +| 361 | 11:03:22 | `TaskLaunchAcknowledged` v18, `lifecycle_phase: started` | `06G45RVKAW20H1D22NHBRYKMN8` | + +Expiry to acknowledged launch: 66 seconds, autonomous. Every event from seq 353 +onward carries `surface: system`. The last `tui` event is the `TaskCorrected` at +09:11:55, before the window. + +Launch receipt, worker journal, 15:03:22 +04: + +``` +launch 06G44JZB80MZBEY97196EZN8EC confirmed: confirmation=editor_cleared submit_attempts=2 first_submit_at=2026-08-27T11:03:22.203859355Z confirmed_at=2026-08-27T11:03:22.717657436Z +``` + +### F16 is partial, not closed + +The hard predicate passed: no renewal carried the stranded epoch after +10:45:26. The run did not exercise the progress comparison. Worker health +records why the renewal was skipped: + +``` +"last_error": "validate lease 06G44JZB80MZBEY97196EZN8EC: tmux capture-pane -p -J -S -200 -t =orchestra-06g44jzb80mzbey97196ezn8ec-e53607ea:1.0: no server running on /tmp/tmux-1000/orchestra: exit status 1", +"error_at": "2026-08-27T11:02:16.213318366Z" +``` + +That is the `paneProgress` error path at `cmd/orchestra-worker/main.go:973`, not +the `default` branch at line 992. The tmux server was gone, so no pane text +could be hashed. + +- **Proven:** dead or missing pane, no renewal, lease expires, autonomous + re-lease and relaunch. +- **Unproven:** live pane with an idle agent and unchanged output, renewal + refused. + +### Epoch reset, stated precisely + +The new epoch and its acknowledged launch establish the reset structurally. +`leases` in `/var/lib/orchestra/worker-state/workpc-claude.json` holds only +`{epoch, version, until}`, with `progress_sha` absent under `omitempty`. No +runtime value of `ProgressSHA`, `UsageBaseline` or `PickupAcknowledged` was +observed. + +### F6 closed + +Worker health reports `active_task_id: 06G44JZB80MZBEY97196EZN8EC` with +`active_pane_id` set, against a leased task. + +### F18 sharpened, non-blocking + +`recordError` (`cmd/orchestra-worker/main.go:70`) keeps one slot, so the earlier +renewal decision near 10:52 was overwritten and cannot be read. One +`last_error` slot cannot preserve a causal sequence. Later replacement: a +bounded recent-error ring or event-backed observations, roughly the last 8 +`{at, class, message}` entries. Do not land this during a live run. + +### Ledger after this checkpoint + +``` +F6 closed +F14 closed +F15 closed by detection, transport fix pending live proof +F16 partial: missing-pane branch proven, static-live-pane branch unproven +F17 fixed, isolated live proof +F18 open observability +F19 fixed +F20 awaiting first post-frame confirmed input +``` + +### Next decisive checkpoint, F20 + +``` +frame completes +→ Orchestra emits the next input +→ worker journal must contain: input to confirmed: confirmation= submit_attempts= +→ only then may the agent continue +``` + +Receipt appears and the model proceeds: let the run continue through phase +transitions. Input lands with no receipt: F20 fails. No input attempted after +`frame`: a different lifecycle or phase-advance bug. + +## Run 3 conformance result, 2026-08-27 12:00 UTC + +```text +run 3 conformance result: failed +cause: no autonomous phase-advance path +``` + +### F16 closed + +```text +missing-pane branch: proven +live-pane idle + unchanged output branch: proven +external editor input ignored as progress: proven +``` + +Second branch evidence, worker health at 11:53:46 with the pane alive and +`herdr_status: reachable`: + +``` +"last_error": "lease 06G44JZB80MZBEY97196EZN8EC not renewed: agent status idle and pane unchanged since the last renewal", +"error_at": "2026-08-27T11:53:46.195553513Z" +``` + +That is the `default` branch at `cmd/orchestra-worker/main.go:992`. + +### F21, lifecycle: the phase brief promises a protocol that does not exist + +`.orchestra/launch.md:62` tells the agent `Orchestra decides when this phase +ends. Ask for a phase change, do not declare one.` The agent complied and +printed `Nothing blocks. Ready for a phase change to implement.` + +```text +phase brief: "ask for a phase change" +agent: asks +worker: has no representation of that request +coordinator: AdvanceWorkPhase exists, but nothing invokes it autonomously +``` + +`operations.AdvanceWorkPhase` is reachable only from the HTTP handler at +`cmd/orchestra/main.go:883` and from `internal/operations/review.go:101`. The +worker's only post-frame send is a decision notice at +`cmd/orchestra-worker/main.go:1639`, gated on `len(answer.Decisions) == 0` +returning early. + +The fix is a bounded agent intent, not pane-text matching: + +```go +type PhaseAdvanceRequest struct { + From WorkPhase + To WorkPhase +} +``` + +```text +verified turn boundary +→ obtain bounded phase intent from harness +→ validate requested transition +→ submit to coordinator +→ operations.AdvanceWorkPhase +→ if transition requires sealed artifact: + collect/seal artifact first +→ rotate/start successor if phase policy requires fresh context +``` + +`frame` needs no artifact. `research` and `plan` take the same path but must +seal their artifact before `AdvanceWorkPhase` accepts them. + +### F22, lifecycle: nothing reacts to `WorkPhaseChanged` at runtime + +`t.WorkPhase` is read only when a launch context is built, at +`cmd/orchestra-worker/main.go:369` and +`internal/orchestrator/orchestrator.go:1218`. No code path rotates, relaunches +or notifies a live session when the phase changes. A manual advance therefore +takes effect only at the next launch, and leaves the current idle pane idle. + +### F23, lifecycle: a decision cannot exist before a submission + +`answer.Decisions` is the only input `sendPrompt` ever carries post-frame. +Decisions come from `operations.ReflectSubmission`, and +`human.PullRequestState.FeedbackAfter` returns input only strictly after a +submission. A task still in `frame` has no submission, so no decision can be +recorded for it. + +Consequence: **F20 cannot be exercised on this task in its current state.** A +manual phase advance does not produce a post-frame input, because nothing +reacts to the phase change and no decision can exist yet. + +### Observation kept separate from F20 + +`❯ go ahead and implement it` appeared in the editor again on attempt 2, cursor +at `cursor_x=2`, never submitted. The agent transcript at +`~/.claude/projects/-tmp-test-e2e-worktrees-06G44JZB80MZBEY97196EZN8EC/97b30e4d-90a7-49bc-a9b8-7d9e450831e4.jsonl` +holds exactly one user message, the launch prompt at 11:55:52.236Z. No worker +send and no receipt exist for it. Origin unknown, external to Orchestra. F20 +does not absorb it. + +### Ledger after this checkpoint + +``` +F6 closed +F14 closed +F15 closed by detection, transport fix pending live proof +F16 closed, both branches live-proven +F17 fixed, isolated live proof +F18 open observability +F19 fixed +F20 blocked, not merely awaiting: see F23 +F21 open lifecycle, no autonomous phase-advance path +F22 open lifecycle, no runtime reaction to WorkPhaseChanged +F23 open lifecycle, no decision path before a submission +``` + +## F21, F22, F23 implemented, 2026-08-27 + +Run 3 was not shepherded further. Attempt 3 was left unspent: F16 is closed on +both branches, and another idle phase would have proven nothing new. + +### F23 was already implemented, and the earlier entry was wrong + +`human.Reconciler.Reconcile` imports issue comments from `task.Source` and +records `HumanDecisionRecorded` with no submission involved. It is wired at +two points in `cmd/orchestra/main.go`: `Store.PreLease`, and +`Coordinator.ReconcileHumanInput`, which `RemoteTurn` calls at every verified +turn boundary. `provider.GiteaComments.FetchAfter` reads the task's own issue. + +The earlier F23 entry traced `EventHumanDecisionRecorded` through +`internal/operations` only and concluded decisions required a submission. That +was a scoping error in the search, not a gap in the code. `ReflectSubmission` +is the pull-request path and stays narrow; it was never the general decision +source. + +Consequence for the record: **F20 had a legitimate post-launch send path +throughout run 3.** A comment on `kami/test-e2e#3` would have produced a +decision, a `DecisionNotice` at the next boundary, and a receipt. + +F23 is closed as already-implemented, with tests added for the boundary it +turns on. + +### F21, the phase-request protocol + +The agent asks with a bounded file. Prose is not a protocol, so nothing +matches on pane text. + +``` +.orchestra/phase-request.json {"from": "frame", "to": "research"} +.orchestra/research.json sealed before leaving research +.orchestra/plan.json sealed before leaving plan +``` + +At a verified turn boundary the worker validates what it can see locally: the +phase the agent believes it is in, the legality of the transition, and the +presence and decode of the artifact the phase must seal. It then calls +`POST /v1/federation/phase` with the lease epoch and a derived operation id. +The coordinator calls the existing `operations.AdvanceWorkPhase`. + +- `internal/operations/workphase.go`: `RequestWorkPhase`, `ErrPhaseRequest`. + Fences on lease epoch, refuses a stale phase belief, refuses any target but + the project's next phase, idempotent per operation id. +- `internal/federation/client.go`: `Client.AdvancePhase`. +- `cmd/orchestra/main.go`: the `/v1/federation/phase` route. +- `cmd/orchestra-worker/main.go`: `requestPhase`, `phaseArtifact`, + `phaseRequestFile`. +- `internal/agentctx/agentctx.go`: `phaseRequestBrief` renders the protocol + under the sentence that used to promise it. + +The operation id is derived, not random: `phase::::`. A +redelivery after a lost response carries the same id, so the coordinator +returns the first event instead of advancing twice. + +### F22, a phase change ends that cognitive session + +`herdr.Session` now carries `Phase`, the phase the session was launched to +run. At a turn boundary a session whose task has moved on is rotated with +reason `phase_changed`, which is added to the closed handoff-reason set in +`internal/herdr/adapter.go` and given its own wording in +`RequestHandoffReason`. + +One comparison covers both cases: a change this worker requested, and one an +operator made through `POST /v1/tasks/{id}/phase`. Both leave the same +evidence, a session built for a phase that is no longer current. + +### F20 hole found and closed while implementing F22 + +`CLIAdapter.prompt` called `backend.Prompt` and returned. Every rotation and +handoff prompt therefore went out unconfirmed, which is precisely the failure +F20 exists to catch. `RequestHandoffReason` is one of its callers, so the +phase rotation would have inherited it. + +Fixed at the shared call site rather than per caller: `prompt` now routes +through `InputConfirmer` and logs the same `input to confirmed:` +receipt the worker logs. + +### Tests + +`go build ./...`, `go vet ./...` and `go test ./...` all pass. + +- `internal/operations/phase_request_test.go`: full path with each artifact + sealed, skipped phase refused, stale phase belief refused, stale epoch + refused, missing operation id refused, redelivery idempotent, operation id + recorded. +- `cmd/orchestra-worker/phase_test.go`: request accepted then session rotates, + external phase change rotates, unsealed artifact refused locally, malformed + artifact refused locally, sealed artifact travels with the request, refused + request retried, decision notice stays undelivered until confirmed. +- `internal/integration/phase_protocol_test.go`: a pre-submission issue + comment steers a live leased session and is not re-sent once delivered; + the phase-request path seals and fences at the coordinator. +- `internal/human/pullrequest_window_test.go`: pull-request feedback ignores + anything not strictly after the submission, and applies trust. +- `internal/agentctx/agentctx_test.go`: the brief names the request file, the + request shape, and the artifact to seal. + +### Ledger + +``` +F6 closed +F14 closed +F15 closed by detection, transport fix pending live proof +F16 closed, both branches live-proven +F17 fixed, isolated live proof +F18 open observability, non-blocking +F19 fixed +F20 fixed, adapter hole closed; live proof pending run 4 +F21 fixed, tests only +F22 fixed, tests only +F23 closed, already implemented; tests added +``` + +Run 4 starts from here. Both halves must be rebuilt and redeployed before it +begins, and the deployment boundary recorded as usual. diff --git a/cmd/orchestra-worker/main.go b/cmd/orchestra-worker/main.go index 61e32ce..4c5873f 100644 --- a/cmd/orchestra-worker/main.go +++ b/cmd/orchestra-worker/main.go @@ -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. diff --git a/cmd/orchestra-worker/phase_test.go b/cmd/orchestra-worker/phase_test.go new file mode 100644 index 0000000..9f9c459 --- /dev/null +++ b/cmd/orchestra-worker/phase_test.go @@ -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)) + } +} diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index f4ab3a4..dfb73f6 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -1222,6 +1222,58 @@ func main() { } json.NewEncoder(w).Encode(federation.TurnDecision{Verdict: verdict, Decisions: decisions}) }) + // F21. The worker's report of an agent's bounded phase-change request. + // The agent asks by writing a file; Orchestra decides here, through the + // same AdvanceWorkPhase the operator surface uses. + mux.HandleFunc("/v1/federation/phase", func(w http.ResponseWriter, r *http.Request) { + if _, err := workerAuth(r); err != nil { + http.Error(w, err.Error(), http.StatusUnauthorized) + return + } + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", 405) + return + } + var body struct { + TaskID string `json:"task_id"` + LeaseEpoch string `json:"lease_epoch"` + OperationID string `json:"operation_id"` + From domain.WorkPhase `json:"from"` + To domain.WorkPhase `json:"to"` + Artifact []byte `json:"artifact"` + } + if json.NewDecoder(r.Body).Decode(&body) != nil || body.TaskID == "" || body.LeaseEpoch == "" { + http.Error(w, "task_id and lease_epoch are required", http.StatusBadRequest) + return + } + t, ok := s.Task(body.TaskID) + if !ok { + http.Error(w, "task not found", 404) + return + } + project, ok := rr.Project(t.Project) + if !ok { + http.Error(w, "unknown project "+t.Project, 409) + return + } + e, err := operations.RequestWorkPhase(s, project, body.TaskID, body.LeaseEpoch, body.OperationID, body.From, body.To, body.Artifact) + if errors.Is(err, operations.ErrTrajectoryGate) { + // The human is being asked. Not a failure, and not a phase change + // the worker should rotate on yet. + w.WriteHeader(http.StatusAccepted) + json.NewEncoder(w).Encode(map[string]string{"status": "trajectory_gate", "detail": err.Error()}) + return + } + if err != nil { + http.Error(w, err.Error(), http.StatusConflict) + return + } + var p struct { + Phase domain.WorkPhase `json:"phase"` + } + _ = json.Unmarshal(e.Payload, &p) + json.NewEncoder(w).Encode(map[string]string{"phase": string(p.Phase)}) + }) mux.HandleFunc("/v1/federation/commands", func(w http.ResponseWriter, r *http.Request) { wid, err := workerAuth(r) if err != nil { diff --git a/internal/agentctx/agentctx.go b/internal/agentctx/agentctx.go index 486f74b..12261af 100644 --- a/internal/agentctx/agentctx.go +++ b/internal/agentctx/agentctx.go @@ -104,6 +104,41 @@ var phaseBrief = map[domain.WorkPhase]string{ domain.WorkPhaseReview: "Check the implementation against the goal, the decisions, and the accepted plan. Report findings with evidence. Do not rewrite the work under review.", } +// phaseRequestBrief states the mechanism behind the sentence above it. The +// brief used to tell an agent to ask for a phase change while nothing carried +// the asking: the agent asked in prose, no code represented the request, and +// the session idled until its lease expired. The request is a file because +// prose in a pane is not a protocol. +func phaseRequestBrief(phase domain.WorkPhase) string { + next := domain.NextPhases(phase) + if len(next) == 0 { + return "" + } + var b strings.Builder + b.WriteString("\nAsk by writing .orchestra/phase-request.json at the end of a turn:\n\n") + fmt.Fprintf(&b, " {\"from\": %q, \"to\": %q}\n", string(phase), string(next[0])) + if len(next) > 1 { + var names []string + for _, p := range next { + names = append(names, string(p)) + } + fmt.Fprintf(&b, "\nLegal values for \"to\" from here: %s. This project may allow fewer, and a request outside its path is refused with the phase you may ask for.\n", strings.Join(names, ", ")) + } + if artifact := phaseSealFile[phase]; artifact != "" { + fmt.Fprintf(&b, "\nSeal .orchestra/%s before you ask. The request is refused without it.\n", artifact) + } + b.WriteString("\nAn accepted request ends this session and starts the next phase with your sealed result. Saying you are ready in the pane is not a request and nothing reads it.\n") + return b.String() +} + +// phaseSealFile is the artifact a phase must seal before it may be left. It +// mirrors the worker's table; both exist because the agent needs to be told +// and the worker needs to check. +var phaseSealFile = map[domain.WorkPhase]string{ + domain.WorkPhaseResearch: "research.json", + domain.WorkPhasePlan: "plan.json", +} + // askingBrief narrows step 4 per phase. The bar is not the same everywhere: a // research phase that has not looked yet has no standing to ask, and an // implementation phase asks only when a discovery invalidates the trajectory @@ -204,6 +239,7 @@ func renderTask(in Input) string { fmt.Fprintf(&b, "\nAsking the human, in this phase: %s\n", brief) } b.WriteString("\nOrchestra decides when this phase ends. Ask for a phase change, do not declare one.\n") + b.WriteString(phaseRequestBrief(in.Phase)) if len(in.Policy) > 0 { b.WriteString("\n## Operating policy\n\n") diff --git a/internal/agentctx/agentctx_test.go b/internal/agentctx/agentctx_test.go index acc468d..9c0dfcf 100644 --- a/internal/agentctx/agentctx_test.go +++ b/internal/agentctx/agentctx_test.go @@ -306,3 +306,48 @@ func TestAcceptanceCriteriaRenderInOrder(t *testing.T) { t.Fatalf("absent acceptance did not render Not stated:\n%s", got.Task) } } + +// The brief used to tell an agent to ask for a phase change while nothing +// carried the asking. Whatever else the wording says, it has to name the file +// the worker actually reads, or the instruction is a promise again. +func TestPhaseBriefNamesTheRequestFile(t *testing.T) { + for _, phase := range []domain.WorkPhase{ + domain.WorkPhaseFrame, domain.WorkPhaseResearch, domain.WorkPhasePlan, domain.WorkPhaseImplement, + } { + out, err := Build(Input{ + Task: domain.Task{ID: "t1", Title: "demo"}, + Phase: phase, + Git: GitState{Worktree: "/w", Branch: "orchestra/t1"}, + }) + if err != nil { + t.Fatalf("%s: %v", phase, err) + } + if !strings.Contains(out.Task, ".orchestra/phase-request.json") { + t.Fatalf("%s brief does not name the request file", phase) + } + if !strings.Contains(out.Task, `"from"`) || !strings.Contains(out.Task, `"to"`) { + t.Fatalf("%s brief does not state the request shape", phase) + } + } +} + +// A phase that seals an artifact must say so where it says how to ask, +// because the request is refused without it. +func TestPhaseBriefNamesTheArtifactToSeal(t *testing.T) { + for phase, file := range map[domain.WorkPhase]string{ + domain.WorkPhaseResearch: "research.json", + domain.WorkPhasePlan: "plan.json", + } { + out, err := Build(Input{ + Task: domain.Task{ID: "t1", Title: "demo"}, + Phase: phase, + Git: GitState{Worktree: "/w", Branch: "orchestra/t1"}, + }) + if err != nil { + t.Fatalf("%s: %v", phase, err) + } + if !strings.Contains(out.Task, ".orchestra/"+file) { + t.Fatalf("%s brief does not name %s", phase, file) + } + } +} diff --git a/internal/domain/workphase.go b/internal/domain/workphase.go index 4408871..4043aba 100644 --- a/internal/domain/workphase.go +++ b/internal/domain/workphase.go @@ -88,3 +88,13 @@ func ValidateWorkPhaseChanged(p map[string]any) error { } return nil } + +// NextPhases returns the phases Orchestra may move to from this one. A +// project's declared path narrows this further, so it is what an agent may +// legally ask for rather than what it will certainly be granted. +func NextPhases(from WorkPhase) []WorkPhase { + if from == "" { + from = WorkPhaseFrame + } + return append([]WorkPhase(nil), legalPhaseTransitions[from]...) +} diff --git a/internal/federation/client.go b/internal/federation/client.go index 2345fc3..ddc17ff 100644 --- a/internal/federation/client.go +++ b/internal/federation/client.go @@ -144,6 +144,30 @@ func (c Client) Turn(ctx context.Context, taskID, epoch, verdict string, deliver return out, nil } +// AdvancePhase carries an agent's bounded phase-change request to the +// coordinator, which decides. The accepted phase comes back so the worker +// knows the session it owns has been superseded and must rotate. +// +// artifact is the sealed output of the phase being left, and is empty for a +// phase that produces none. +func (c Client) AdvancePhase(ctx context.Context, taskID, epoch, operationID string, from, to domain.WorkPhase, artifact []byte) (domain.WorkPhase, error) { + resp, err := c.request(ctx, http.MethodPost, "/v1/federation/phase", map[string]any{ + "task_id": taskID, "lease_epoch": epoch, "operation_id": operationID, + "from": string(from), "to": string(to), "artifact": artifact, + }) + if err != nil { + return "", err + } + defer resp.Body.Close() + var out struct { + Phase domain.WorkPhase `json:"phase"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return "", err + } + return out.Phase, nil +} + // Intent fetches the reduced authority for one task: its contract plus the // human decisions still standing. A worker renders its launch instruction // from this, never from handoff prose. diff --git a/internal/herdr/adapter.go b/internal/herdr/adapter.go index aa9639a..732d38e 100644 --- a/internal/herdr/adapter.go +++ b/internal/herdr/adapter.go @@ -6,6 +6,7 @@ import ( "encoding/hex" "encoding/json" "fmt" + "log" "orchestra/internal/continuity" "os" "os/exec" @@ -173,7 +174,23 @@ func (a CLIAdapter) prompt(ctx context.Context, s Session, text string, wait tim if client, ok := backend.(*Client); ok { client.BindAgent(s.PaneID, s.AgentName) } - return backend.Prompt(ctx, s.PaneID, text, wait) + if err := backend.Prompt(ctx, s.PaneID, text, wait); err != nil { + return err + } + // Every Orchestra-originated pane write confirms (F20). This is the + // shared path for handoff and rotation prompts, so leaving it unconfirmed + // left the exact failure F20 exists to catch: a prompt sitting unsubmitted + // in the editor while Orchestra waits for a reply that cannot come. + c, ok := backend.(InputConfirmer) + if !ok { + return nil + } + evidence, err := c.ConfirmInput(ctx, s, text) + if err != nil { + return err + } + log.Printf("input to %s confirmed: %s", s.PaneID, evidence) + return nil } const handoffPrompt = `Orchestra is about to rotate this task. Write ONLY the following labelled answers to ` + HandoffReportFile + `, then stop. Output nothing else. @@ -216,6 +233,8 @@ func (a CLIAdapter) RequestHandoffReason(ctx context.Context, s Session, reason sb.WriteString("Signs of thrashing were detected (repeated failing test runs, repeated edits to the same file, or the same tool call repeated back to back). Stop the current approach rather than trying it again.\n") case "milestone": sb.WriteString("A coherent unit of work looks complete (a successful commit). If the next step is independent of what you just did, this is a good point to hand off.\n") + case "phase_changed": + sb.WriteString("This task has moved to its next work phase, so this session's context is no longer the right one for it. This is not a judgement about your work: the next phase starts fresh with the result you sealed. Stop at a clean point and hand off.\n") case "reconcile_failure": sb.WriteString("Orchestra cannot currently read the human input for this task, so it can no longer guarantee your instructions are current. Stop at a clean point and hand off. This is not a judgement about your work.\n") } @@ -528,7 +547,7 @@ func (a CLIAdapter) lastObservedCommand(s Session) string { func handoffReason(s Session) string { switch s.HandoffReason { - case "threshold", "milestone", "thrash", "manual", "reconcile_failure": + case "threshold", "milestone", "thrash", "manual", "reconcile_failure", "phase_changed": return s.HandoffReason default: return "threshold" diff --git a/internal/herdr/herdr.go b/internal/herdr/herdr.go index 840ea1a..6765049 100644 --- a/internal/herdr/herdr.go +++ b/internal/herdr/herdr.go @@ -160,6 +160,11 @@ type Session struct { // session's lease was created — the immutable-spec hash continuity's // pickup validation compares against on the next rotation (§6.2). TaskFileSHA string `json:"task_file_sha,omitempty"` + // Phase is the work phase this session was launched in. Orchestra may + // advance the phase while the session runs; a session that no longer + // matches its task's phase is finished, because a phase change is a + // change of cognitive context and not a change of instruction. + Phase string `json:"phase,omitempty"` // DeliveredDecisions holds the ids of the human decisions this session has // already been shown. A decision recorded while the lease is live is // delivered at the next verified turn boundary, and recording it here is diff --git a/internal/human/pullrequest_window_test.go b/internal/human/pullrequest_window_test.go new file mode 100644 index 0000000..5c27c12 --- /dev/null +++ b/internal/human/pullrequest_window_test.go @@ -0,0 +1,42 @@ +package human + +import ( + "testing" + "time" +) + +// The two concepts must stay separate. Pull-request feedback is a response to +// a submission, so anything written at or before the submission was already +// visible when it was made and cannot be a response to it. Steering written +// earlier is issue input, reconciled through the task's own source, and it +// stays valid in every phase. +func TestPullRequestFeedbackIgnoresAnythingNotAfterTheSubmission(t *testing.T) { + submitted := time.Date(2026, 8, 27, 12, 0, 0, 0, time.UTC) + state := PullRequestState{ + Comments: []Input{ + {Provider: "gitea", ExternalID: "1", Author: "kami", At: submitted.Add(-time.Hour), Body: "before"}, + {Provider: "gitea", ExternalID: "2", Author: "kami", At: submitted, Body: "at"}, + {Provider: "gitea", ExternalID: "3", Author: "kami", At: submitted.Add(time.Hour), Body: "after"}, + }, + Reviews: []ReviewObservation{ + {Actor: "kami", State: "changes_requested", At: submitted.Add(-time.Minute), Body: "early review"}, + }, + } + got := state.FeedbackAfter("gitea", submitted, Trust{}) + if len(got) != 1 || got[0].Body != "after" { + t.Fatalf("feedback = %+v", got) + } +} + +// An untrusted actor's words never move a task, whenever they arrive. +func TestPullRequestFeedbackAppliesTrust(t *testing.T) { + submitted := time.Date(2026, 8, 27, 12, 0, 0, 0, time.UTC) + state := PullRequestState{Comments: []Input{ + {Provider: "gitea", ExternalID: "1", Author: "bot", At: submitted.Add(time.Hour), Body: "merged by automation"}, + {Provider: "gitea", ExternalID: "2", Author: "kami", At: submitted.Add(time.Hour), Body: "change this"}, + }} + got := state.FeedbackAfter("gitea", submitted, Trust{Ignored: []string{"bot"}}) + if len(got) != 1 || got[0].Author != "kami" { + t.Fatalf("feedback = %+v", got) + } +} diff --git a/internal/integration/phase_protocol_test.go b/internal/integration/phase_protocol_test.go new file mode 100644 index 0000000..bb88daf --- /dev/null +++ b/internal/integration/phase_protocol_test.go @@ -0,0 +1,132 @@ +package integration + +import ( + "context" + "testing" + + "orchestra/internal/domain" + "orchestra/internal/human" + "orchestra/internal/operations" + "orchestra/internal/orchestrator" + "orchestra/internal/registry" + "orchestra/internal/router" + "orchestra/internal/workphase" +) + +// F23. Human steering must work throughout the task, not only after a pull +// request exists. Pull-request feedback is a separate concept with a narrower +// window; conflating them would leave every pre-submission phase unsteerable, +// which is the state run 3 was diagnosed in. +// +// The task here never submits anything. It sits in its first phase, leased and +// running, and a comment on its own issue still becomes a standing decision +// that the live session is handed at its next verified turn boundary. +func TestPreSubmissionCommentSteersALiveSession(t *testing.T) { + s, reg, _ := setup(t) + task := ingest(t, s, "381") + src := &tracingSource{tr: &trace{}} + rec := &human.Reconciler{Store: s, Sources: map[string]human.Source{"gitea": src}} + s.PreLease = func(id string) error { return rec.Reconcile(context.Background(), id) } + + c := &orchestrator.Coordinator{Store: s, Worktrees: worktrees{}, Adapters: adapters{&harness{occupancy: .1}}, StatePath: t.TempDir() + "/sessions.json", Hard: .8} + c.ReconcileHumanInput = rec.Reconcile + rt := router.Router{Store: s, Registry: reg, Reachability: alwaysReachable{}, OnLease: func(e domain.Event) error { + return c.Start(context.Background(), e) + }} + if leased, err := rt.AssignPending(); err != nil || len(leased) != 1 { + t.Fatalf("leased=%d err=%v", len(leased), err) + } + leasedTask, _ := s.Task(task.ID) + if leasedTask.Lease == nil { + t.Fatal("task is not leased") + } + // Still in the first phase, and nothing has been submitted. + if leasedTask.WorkPhase != "" && leasedTask.WorkPhase != domain.WorkPhaseFrame { + t.Fatalf("phase = %q", leasedTask.WorkPhase) + } + epoch := leasedTask.Lease.Epoch + + // No human input yet: the boundary answers with nothing to deliver. + _, decisions, err := c.RemoteTurn(context.Background(), task.ID, epoch, orchestrator.TurnContinue, nil) + if err != nil { + t.Fatal(err) + } + if len(decisions) != 0 { + t.Fatalf("decisions before any comment = %+v", decisions) + } + + // The human comments on the issue while the session runs. + src.next = "918" + src.inputs = []human.Input{{Provider: "gitea", ExternalID: "918", Author: "kami", Body: "no, use b"}} + + _, decisions, err = c.RemoteTurn(context.Background(), task.ID, epoch, orchestrator.TurnContinue, nil) + if err != nil { + t.Fatal(err) + } + if len(decisions) != 1 || decisions[0].Value != "no, use b" { + t.Fatalf("decisions = %+v", decisions) + } + + // Reported as delivered, so the same correction is not re-sent every turn. + _, decisions, err = c.RemoteTurn(context.Background(), task.ID, epoch, orchestrator.TurnContinue, []string{decisions[0].ID}) + if err != nil { + t.Fatal(err) + } + if len(decisions) != 0 { + t.Fatalf("a delivered decision repeated: %+v", decisions) + } +} + +// F21 and F22 at the coordinator boundary: an accepted request moves the phase +// and seals what the phase produced, and a stale session's request is refused. +func TestPhaseRequestPathSealsAndFences(t *testing.T) { + s, reg, _ := setup(t) + task := ingest(t, s, "381") + c := &orchestrator.Coordinator{Store: s, Worktrees: worktrees{}, Adapters: adapters{&harness{occupancy: .1}}, StatePath: t.TempDir() + "/sessions.json", Hard: .8} + rt := router.Router{Store: s, Registry: reg, Reachability: alwaysReachable{}, OnLease: func(e domain.Event) error { + return c.Start(context.Background(), e) + }} + if _, err := rt.AssignPending(); err != nil { + t.Fatal(err) + } + leased, _ := s.Task(task.ID) + epoch := leased.Lease.Epoch + project := registry.Project{ID: "p"} + + if _, err := operations.RequestWorkPhase(s, project, task.ID, epoch, "op-1", domain.WorkPhaseFrame, domain.WorkPhaseResearch, nil); err != nil { + t.Fatal(err) + } + got, _ := s.Task(task.ID) + if got.WorkPhase != domain.WorkPhaseResearch { + t.Fatalf("phase = %q", got.WorkPhase) + } + + // A request carrying a lease epoch that no longer owns the task is a stale + // opinion from a session that has been superseded. + if _, err := operations.RequestWorkPhase(s, project, task.ID, "stale-epoch", "op-2", domain.WorkPhaseResearch, domain.WorkPhasePlan, sealed(t, researchArtifact)); err == nil { + t.Fatal("a stale epoch advanced the phase") + } + if got, _ := s.Task(task.ID); got.WorkPhase != domain.WorkPhaseResearch { + t.Fatalf("phase moved on a stale request: %q", got.WorkPhase) + } + + // The real owner's request seals the research the next phase will read. + if _, err := operations.RequestWorkPhase(s, project, task.ID, epoch, "op-3", domain.WorkPhaseResearch, domain.WorkPhasePlan, sealed(t, researchArtifact)); err != nil { + t.Fatal(err) + } + got, _ = s.Task(task.ID) + if got.WorkPhase != domain.WorkPhasePlan || got.ResearchRef == "" { + t.Fatalf("task = %+v", got) + } +} + +var researchArtifact = workphase.Research{Findings: []workphase.Finding{{Claim: "runs per figure", Evidence: "attr.go:88"}}} + +func sealed(t *testing.T, v interface{ Validate() error }) []byte { + t.Helper() + b, err := workphase.Encode(v) + if err != nil { + t.Fatal(err) + } + return b +} diff --git a/internal/operations/phase_request_test.go b/internal/operations/phase_request_test.go new file mode 100644 index 0000000..ff4de72 --- /dev/null +++ b/internal/operations/phase_request_test.go @@ -0,0 +1,167 @@ +package operations + +import ( + "encoding/json" + "errors" + "testing" + + "orchestra/internal/domain" + "orchestra/internal/registry" + "orchestra/internal/store" +) + +// epochOf returns the epoch of the task's current lease, which fences every +// worker-driven call. +func epochOf(t *testing.T, s *store.Store, id string) string { + t.Helper() + task, ok := s.Task(id) + if !ok || task.Lease == nil { + t.Fatal("task has no lease") + } + return task.Lease.Epoch +} + +func TestRequestWorkPhaseAdvancesAndSealsEachArtifact(t *testing.T) { + s, id := phaseStore(t) + project := registry.Project{ID: "p"} + lease(t, s, id) + epoch := epochOf(t, s, id) + + // frame -> research seals nothing: framing produces no artifact. + if _, err := RequestWorkPhase(s, project, id, epoch, "op-1", domain.WorkPhaseFrame, domain.WorkPhaseResearch, nil); err != nil { + t.Fatal(err) + } + if got, _ := s.Task(id); got.WorkPhase != domain.WorkPhaseResearch { + t.Fatalf("phase = %q", got.WorkPhase) + } + + // research -> plan is refused until the research is sealed. + if _, err := RequestWorkPhase(s, project, id, epoch, "op-2", domain.WorkPhaseResearch, domain.WorkPhasePlan, nil); !errors.Is(err, domain.ErrInvalid) { + t.Fatalf("leaving research unsealed must fail, got %v", err) + } + if _, err := RequestWorkPhase(s, project, id, epoch, "op-3", domain.WorkPhaseResearch, domain.WorkPhasePlan, sealed(t, research)); err != nil { + t.Fatal(err) + } + got, _ := s.Task(id) + if got.WorkPhase != domain.WorkPhasePlan || got.ResearchRef == "" { + t.Fatalf("task = %+v", got) + } + + // plan -> implement is refused until the plan is sealed. + if _, err := RequestWorkPhase(s, project, id, epoch, "op-4", domain.WorkPhasePlan, domain.WorkPhaseImplement, nil); !errors.Is(err, domain.ErrInvalid) { + t.Fatalf("leaving plan unsealed must fail, got %v", err) + } + if _, err := RequestWorkPhase(s, project, id, epoch, "op-5", domain.WorkPhasePlan, domain.WorkPhaseImplement, sealed(t, plan)); err != nil { + t.Fatal(err) + } + got, _ = s.Task(id) + if got.WorkPhase != domain.WorkPhaseImplement || got.PlanRef == "" || got.ResearchRef == "" { + t.Fatalf("task = %+v", got) + } +} + +func TestRequestWorkPhaseRefusesASkippedPhase(t *testing.T) { + s, id := phaseStore(t) + lease(t, s, id) + // frame -> implement is a legal domain transition, but not the next step + // on this project's declared path. The agent is refused rather than + // silently corrected. + _, err := RequestWorkPhase(s, registry.Project{ID: "p"}, id, epochOf(t, s, id), "op-1", domain.WorkPhaseFrame, domain.WorkPhaseImplement, nil) + if !errors.Is(err, ErrPhaseRequest) { + t.Fatalf("err = %v", err) + } + if got, _ := s.Task(id); got.WorkPhase != "" { + t.Fatalf("phase moved to %q", got.WorkPhase) + } +} + +func TestRequestWorkPhaseRefusesAStalePhaseBelief(t *testing.T) { + s, id := phaseStore(t) + lease(t, s, id) + epoch := epochOf(t, s, id) + if _, err := RequestWorkPhase(s, registry.Project{ID: "p"}, id, epoch, "op-1", domain.WorkPhaseFrame, domain.WorkPhaseResearch, nil); err != nil { + t.Fatal(err) + } + // The agent still believes it is framing. Acting on this would advance a + // phase it never ran. + _, err := RequestWorkPhase(s, registry.Project{ID: "p"}, id, epoch, "op-2", domain.WorkPhaseFrame, domain.WorkPhaseResearch, nil) + if !errors.Is(err, ErrPhaseRequest) { + t.Fatalf("err = %v", err) + } +} + +func TestRequestWorkPhaseRefusesAStaleLeaseEpoch(t *testing.T) { + s, id := phaseStore(t) + lease(t, s, id) + _, err := RequestWorkPhase(s, registry.Project{ID: "p"}, id, "not-the-epoch", "op-1", domain.WorkPhaseFrame, domain.WorkPhaseResearch, nil) + if !errors.Is(err, domain.ErrConflict) { + t.Fatalf("err = %v", err) + } + if got, _ := s.Task(id); got.WorkPhase != "" { + t.Fatalf("phase moved to %q", got.WorkPhase) + } +} + +func TestRequestWorkPhaseRequiresAnOperationID(t *testing.T) { + s, id := phaseStore(t) + lease(t, s, id) + _, err := RequestWorkPhase(s, registry.Project{ID: "p"}, id, epochOf(t, s, id), "", domain.WorkPhaseFrame, domain.WorkPhaseResearch, nil) + if !errors.Is(err, domain.ErrInvalid) { + t.Fatalf("err = %v", err) + } +} + +// A lost response is the ordinary case, not the exotic one: the worker resends +// the same request and must not advance the phase a second time. +func TestRequestWorkPhaseIsIdempotentPerOperationID(t *testing.T) { + s, id := phaseStore(t) + project := registry.Project{ID: "p"} + lease(t, s, id) + epoch := epochOf(t, s, id) + + first, err := RequestWorkPhase(s, project, id, epoch, "op-1", domain.WorkPhaseFrame, domain.WorkPhaseResearch, nil) + if err != nil { + t.Fatal(err) + } + second, err := RequestWorkPhase(s, project, id, epoch, "op-1", domain.WorkPhaseFrame, domain.WorkPhaseResearch, nil) + if err != nil { + t.Fatal(err) + } + if second.ID != first.ID { + t.Fatalf("redelivery produced a new event: %s then %s", first.ID, second.ID) + } + if got, _ := s.Task(id); got.WorkPhase != domain.WorkPhaseResearch { + t.Fatalf("phase = %q", got.WorkPhase) + } + var changes int + for _, e := range s.Events(0) { + if e.TaskID == id && e.Type == domain.EventWorkPhaseChanged { + changes++ + } + } + if changes != 1 { + t.Fatalf("recorded %d phase changes, want 1", changes) + } +} + +// The operation id is what makes redelivery safe, so it has to survive into +// the event the next redelivery reads. +func TestRequestWorkPhaseRecordsTheOperationID(t *testing.T) { + s, id := phaseStore(t) + lease(t, s, id) + e, err := RequestWorkPhase(s, registry.Project{ID: "p"}, id, epochOf(t, s, id), "op-1", domain.WorkPhaseFrame, domain.WorkPhaseResearch, nil) + if err != nil { + t.Fatal(err) + } + var p struct { + OperationID string `json:"operation_id"` + From string `json:"from"` + Phase string `json:"phase"` + } + if err := json.Unmarshal(e.Payload, &p); err != nil { + t.Fatal(err) + } + if p.OperationID != "op-1" || p.From != "frame" || p.Phase != "research" { + t.Fatalf("payload = %+v", p) + } +} diff --git a/internal/operations/workphase.go b/internal/operations/workphase.go index 19fd57d..97b6401 100644 --- a/internal/operations/workphase.go +++ b/internal/operations/workphase.go @@ -2,6 +2,7 @@ package operations import ( "encoding/json" + "errors" "fmt" "orchestra/internal/authz" @@ -94,3 +95,77 @@ func current(t domain.Task) domain.WorkPhase { } return t.WorkPhase } + +// ErrPhaseRequest reports that an agent's phase request was refused. It is +// distinct from ErrTrajectoryGate: a gate is the human being asked, this is +// the request itself being wrong. +var ErrPhaseRequest = errors.New("phase request refused") + +// RequestWorkPhase is the agent-initiated half of a phase change (F21). +// +// The phase brief tells the agent to ask for a phase change rather than +// declare one, and until this existed nothing carried the request. The agent +// asked, the worker had no representation of the asking, and the session sat +// idle until its lease expired. That is what made run 3 fail conformance. +// +// The agent asks; Orchestra still decides. Everything the agent supplies is +// checked here: the phase it believes it is in, the phase it wants, the +// artifact the phase it is leaving must seal. The transition itself is +// AdvanceWorkPhase, unchanged, so a request can never reach a move the +// operator surface could not also make. +// +// leaseEpoch fences the request the way every other worker-driven call is +// fenced: a request written by a session whose lease has since been +// reassigned is a stale opinion, not an instruction. +// +// operationID makes redelivery idempotent. A lost response must not advance +// the phase twice, so a request already recorded under the same id returns +// its event rather than moving again. +func RequestWorkPhase(s *store.Store, project registry.Project, taskID, leaseEpoch, operationID string, from, to domain.WorkPhase, artifact []byte) (domain.Event, error) { + if operationID == "" { + return domain.Event{}, fmt.Errorf("%w: operation_id required", domain.ErrInvalid) + } + if e, ok := phaseOperation(s, taskID, operationID); ok { + return e, nil + } + t, ok := s.Task(taskID) + if !ok { + return domain.Event{}, domain.ErrNotFound + } + if t.Lease == nil || leaseEpoch == "" || t.Lease.Epoch != leaseEpoch { + return domain.Event{}, domain.ErrConflict + } + // The agent states the phase it believes it is in. Disagreeing with the + // log means it is working from a stale context, and acting on its request + // would advance a phase it never actually ran. + if from != current(t) { + return domain.Event{}, fmt.Errorf("%w: task %s is in work phase %q, not %q", ErrPhaseRequest, taskID, current(t), from) + } + next, ok := project.NextPhase(current(t)) + if !ok { + return domain.Event{}, fmt.Errorf("%w: work phase %q is the end of project %s's path", ErrPhaseRequest, current(t), project.ID) + } + // Only the next phase on the project's declared path. An agent that asks + // to skip one is refused rather than quietly corrected, because a silent + // correction would teach it the wrong protocol. + if to != next { + return domain.Event{}, fmt.Errorf("%w: task %s may only move to %q, not %q", ErrPhaseRequest, taskID, next, to) + } + return advanceWorkPhase(s, project, taskID, artifact, map[string]any{"operation_id": operationID}) +} + +// phaseOperation finds a phase change already recorded under this operation id. +func phaseOperation(s *store.Store, taskID, operationID string) (domain.Event, bool) { + for _, e := range s.Events(0) { + if e.TaskID != taskID || e.Type != domain.EventWorkPhaseChanged { + continue + } + var p struct { + OperationID string `json:"operation_id"` + } + if json.Unmarshal(e.Payload, &p) == nil && p.OperationID != "" && p.OperationID == operationID { + return e, true + } + } + return domain.Event{}, false +}