diff --git a/cmd/orchestra-worker/main.go b/cmd/orchestra-worker/main.go index b7aef84..61e32ce 100644 --- a/cmd/orchestra-worker/main.go +++ b/cmd/orchestra-worker/main.go @@ -454,8 +454,8 @@ func (w *worker) start(ctx context.Context, t domain.Task, ref string) error { // Acknowledging a launch means the harness accepted the instruction, not // that the adapter call returned nil. Without this the worker reported a // started agent while the prompt sat unsubmitted in the input editor. - if c, ok := backend.(herdr.LaunchConfirmer); ok { - evidence, confirmErr := c.ConfirmLaunch(ctx, s, submitted) + if c, ok := backend.(herdr.InputConfirmer); ok { + evidence, confirmErr := c.ConfirmInput(ctx, s, submitted) if confirmErr != nil { // An unsubmitted prompt leaves a live pane that nothing owns, and // a retained session would make the retry skip this task @@ -673,6 +673,11 @@ func (w *worker) sendLine(ctx context.Context, s herdr.Session, line string) err if err := backend.SendKeys(ctx, s, []string{"ENTER"}); err != nil { return fmt.Errorf("submit %q: %w", line, err) } + // A lost Enter here leaves the session mid-rollover with /clear sitting in + // the editor, which is worse than a lost launch: nothing retries it. + if err := w.confirmInput(ctx, s, line); err != nil { + return fmt.Errorf("submit %q: %w", line, err) + } return nil } @@ -962,7 +967,9 @@ func (w *worker) renewLeases(ctx context.Context) { continue } adapter := herdr.CLIAdapter{Backend: w.executionBackend(), Harness: w.harness} - text, err := adapter.PaneCapture(ctx, s, "recent") + // Input lines are excluded: keystrokes arriving at a pane, from + // Orchestra or from anyone else, are not the agent doing work. + text, err := w.paneProgress(ctx, adapter, s) if err != nil { w.recordError(fmt.Errorf("validate lease %s: %w", taskID, err)) continue @@ -1641,10 +1648,41 @@ func (w *worker) federatedTurn(ctx context.Context, id string, a herdr.Adapter, _ = w.save() } +// 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. func (w *worker) sendPrompt(ctx context.Context, s herdr.Session, text string) error { backend := w.executionBackend() if backend == nil { return fmt.Errorf("execution backend is not configured") } - return backend.Prompt(ctx, s.PaneID, text, time.Minute) + if err := backend.Prompt(ctx, s.PaneID, text, time.Minute); err != nil { + return err + } + return w.confirmInput(ctx, s, text) +} + +// confirmInput is the one place Orchestra proves a write landed. A backend +// whose own protocol acknowledges input does not implement InputConfirmer and +// needs no second opinion. +func (w *worker) confirmInput(ctx context.Context, s herdr.Session, text string) error { + c, ok := w.executionBackend().(herdr.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 +} + +// paneProgress reports pane content with input lines removed where the backend +// can separate them, and falls back to the raw capture where it cannot. +func (w *worker) paneProgress(ctx context.Context, adapter herdr.CLIAdapter, s herdr.Session) (string, error) { + if p, ok := w.executionBackend().(herdr.PaneProgress); ok { + return p.PaneProgress(ctx, s) + } + return adapter.PaneCapture(ctx, s, "recent") } diff --git a/cmd/orchestra-worker/main_test.go b/cmd/orchestra-worker/main_test.go index d91955f..e12f065 100644 --- a/cmd/orchestra-worker/main_test.go +++ b/cmd/orchestra-worker/main_test.go @@ -370,10 +370,17 @@ func TestApprovalResponseOpenCodeAllowOnce(t *testing.T) { func mustJSON(v any) []byte { b, _ := json.Marshal(v); return b } type recordingBackend struct { - status string - capture string - calls []string - prompts []string + status string + capture string + progress string + calls []string + prompts []string +} + +// PaneProgress makes this stub the kind of backend that can separate harness +// output from input, which is what the worker must prefer over a raw capture. +func (b *recordingBackend) PaneProgress(context.Context, herdr.Session) (string, error) { + return b.progress, nil } func (b *recordingBackend) Kind() string { return "recording" } @@ -406,7 +413,7 @@ func TestRenewLeasesRequiresProgress(t *testing.T) { rw.Write([]byte(`{}`)) })) defer api.Close() - backend := &recordingBackend{status: "idle", capture: "same screen"} + backend := &recordingBackend{status: "idle", progress: "same screen"} w := &worker{ api: federation.Client{BaseURL: api.URL, WorkerID: "h", Token: "t"}, backend: backend, @@ -425,7 +432,7 @@ func TestRenewLeasesRequiresProgress(t *testing.T) { if renewals != 1 { t.Fatalf("busy agent renewals=%d, want 1", renewals) } - backend.status, backend.capture = "idle", "new output" + backend.status, backend.progress = "idle", "new output" l := w.leases["task"] l.Until = time.Now() w.leases["task"] = l @@ -764,7 +771,7 @@ type launchBackend struct { } func (b *launchBackend) LaunchTransport(string) herdr.LaunchTransport { return b.transport } -func (b *launchBackend) ConfirmLaunch(context.Context, herdr.Session, string) (string, error) { +func (b *launchBackend) ConfirmInput(context.Context, herdr.Session, string) (string, error) { if !b.confirmed { return "", fmt.Errorf("%w: editor still holds the prompt", herdr.ErrPromptNotSubmitted) } @@ -915,3 +922,36 @@ func TestReconcileLeasesKeepsWorkerLocalObservations(t *testing.T) { t.Fatalf("observations survived a new epoch: %+v", got) } } + +// The live shape of the run-3 stall: someone typed into the pane, the capture +// changed, and nothing the agent did changed at all. Renewal must refuse. +func TestRenewLeasesIgnoresPaneInput(t *testing.T) { + renewals := 0 + api := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) { + renewals++ + rw.Write([]byte(`{}`)) + })) + defer api.Close() + backend := &recordingBackend{status: "idle", capture: "screen\n\u276f ", progress: "screen"} + 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: "e", Version: 1, Until: time.Now(), ProgressSHA: domain.Hash([]byte("screen"))}}, + quarantined: map[string]bool{}, + statePath: filepath.Join(t.TempDir(), "state.json"), + } + // Only the input line changes. The agent has done nothing. + backend.capture = "screen\n\u276f go ahead and implement it" + w.renewLeases(context.Background()) + if renewals != 0 { + t.Fatalf("pane input renewed the lease %d times", renewals) + } + // Harness output moves the digest, and only then does the lease renew. + backend.progress = "screen\nedited the script" + w.renewLeases(context.Background()) + if renewals != 1 { + t.Fatalf("renewals=%d, want 1 after real progress", renewals) + } +} diff --git a/internal/herdr/backend.go b/internal/herdr/backend.go index 0173cc1..e98f436 100644 --- a/internal/herdr/backend.go +++ b/internal/herdr/backend.go @@ -113,12 +113,25 @@ type LaunchTransporter interface { // uncertain one: the lease must be released and retried rather than held. var ErrPromptNotSubmitted = errors.New("prompt_not_submitted") -// LaunchConfirmer is optional. A backend that does not implement it treats a +// InputConfirmer is optional. A backend that does not implement it treats a // successful Prompt as proof of submission, which is only sound where the // backend's own protocol acknowledges the prompt. // -// ConfirmLaunch returns the evidence that convinced it, or an error wrapping +// Every write Orchestra originates goes through this, not only the launch: a +// lost Enter on a phase continuation or a context reset strands the session +// exactly as a lost launch does, and burn-in run 3 showed the Enter is lost +// often enough to matter. +// +// ConfirmInput returns the evidence that convinced it, or an error wrapping // ErrPromptNotSubmitted when the submission cannot be observed. -type LaunchConfirmer interface { - ConfirmLaunch(ctx context.Context, s Session, submitted string) (string, error) +type InputConfirmer interface { + ConfirmInput(ctx context.Context, s Session, submitted string) (string, error) +} + +// PaneProgress is optional. It reports pane content with harness input lines +// removed, so that typing into a pane is not mistaken for the agent doing +// work. Orchestra does not guarantee delivery of input it did not originate, +// but it must never count that input as progress. +type PaneProgress interface { + PaneProgress(ctx context.Context, s Session) (string, error) } diff --git a/internal/herdr/tmux.go b/internal/herdr/tmux.go index 882353c..63b6506 100644 --- a/internal/herdr/tmux.go +++ b/internal/herdr/tmux.go @@ -27,7 +27,7 @@ type TmuxBackend struct { Command string // Binary is test/packaging override for tmux itself. Binary string - // LaunchConfirmTimeout and LaunchConfirmPoll bound ConfirmLaunch. They are + // LaunchConfirmTimeout and LaunchConfirmPoll bound ConfirmInput. They are // tunable because how fast a terminal harness visibly reacts is a property // of the host, not of this code. Zero values mean 10s and 250ms. LaunchConfirmTimeout time.Duration @@ -428,12 +428,32 @@ func (b *TmuxBackend) inputState(ctx context.Context, s Session) (InputState, er return InputState{Text: strings.TrimSpace(strings.Join(parts, " ")), Active: true}, nil } +// PaneProgress hashes what the harness produced, not what someone typed at it. +// The capture joins wrapped lines, so an input block is one line beginning +// with the prompt marker and dropping those lines removes it whole. Found live +// during burn-in run 3: unexplained keystrokes in a pane kept a stalled lease +// renewing, because the renewal check hashed the whole capture. +func (b *TmuxBackend) PaneProgress(ctx context.Context, s Session) (string, error) { + text, err := b.PaneCapture(ctx, s, "recent") + if err != nil { + return "", err + } + var kept []string + for _, line := range strings.Split(text, "\n") { + if promptLine.MatchString(line) { + continue + } + kept = append(kept, line) + } + return strings.Join(kept, "\n"), nil +} + // launchResubmitLimit bounds the intervention independently of the observation // deadline. A slow TUI must not receive a fortieth Enter after it accepted the // first: three exact-editor resubmits, then observation only. const launchResubmitLimit = 3 -// ConfirmLaunch drives the submit to a decision instead of assuming one Enter +// ConfirmInput drives the submit to a decision instead of assuming one Enter // landed. Burn-in run 3 proved the submit is not deterministic: the text // reached the editor on all three attempts and the following Enter never took // effect. So this resends Enter while the live editor still holds exactly what @@ -442,7 +462,7 @@ const launchResubmitLimit = 3 // // The returned evidence records how many submits it took, which is the only // way to tell a harness that needs a second Enter from one that needed none. -func (b *TmuxBackend) ConfirmLaunch(ctx context.Context, s Session, submitted string) (string, error) { +func (b *TmuxBackend) ConfirmInput(ctx context.Context, s Session, submitted string) (string, error) { timeout, poll := b.LaunchConfirmTimeout, b.LaunchConfirmPoll if timeout <= 0 { timeout = 10 * time.Second diff --git a/internal/herdr/tmux_test.go b/internal/herdr/tmux_test.go index 49900ba..ee45660 100644 --- a/internal/herdr/tmux_test.go +++ b/internal/herdr/tmux_test.go @@ -172,9 +172,9 @@ func TestInputStateReadsTheEditorOwningTheCursor(t *testing.T) { // Run 3 proved one Enter is not enough. A launch still owning the cursor must // be resubmitted, and the evidence must say how many submits it took. -func TestConfirmLaunchResubmitsUntilTheEditorClears(t *testing.T) { +func TestConfirmInputResubmitsUntilTheEditorClears(t *testing.T) { b, keys := submitTmux(t, 2) - evidence, err := b.ConfirmLaunch(context.Background(), Session{PaneID: "s:1.0"}, LaunchReference) + evidence, err := b.ConfirmInput(context.Background(), Session{PaneID: "s:1.0"}, LaunchReference) if err != nil { t.Fatalf("confirm: %v", err) } @@ -188,11 +188,11 @@ func TestConfirmLaunchResubmitsUntilTheEditorClears(t *testing.T) { // The intervention is bounded independently of the observation deadline: a TUI // that already accepted the prompt must not receive a fortieth Enter. -func TestConfirmLaunchBoundsResubmission(t *testing.T) { +func TestConfirmInputBoundsResubmission(t *testing.T) { b, keys := submitTmux(t, 99) // Generous deadline on purpose: the cap must be what stops the resends. b.LaunchConfirmTimeout = 3 * time.Second - if _, err := b.ConfirmLaunch(context.Background(), Session{PaneID: "s:1.0"}, LaunchReference); !errors.Is(err, ErrPromptNotSubmitted) { + if _, err := b.ConfirmInput(context.Background(), Session{PaneID: "s:1.0"}, LaunchReference); !errors.Is(err, ErrPromptNotSubmitted) { t.Fatalf("err=%v, want ErrPromptNotSubmitted", err) } if got := enterCount(t, keys); got != launchResubmitLimit { @@ -202,10 +202,10 @@ func TestConfirmLaunchBoundsResubmission(t *testing.T) { // Queued is submitted. Treating it as a stalled editor would kill a pane whose // harness had already accepted the instruction. -func TestConfirmLaunchAcceptsQueuedInput(t *testing.T) { +func TestConfirmInputAcceptsQueuedInput(t *testing.T) { pane := "\u2500\u2500\u2500\u2500\n\u276f Press up to edit queued messages\n" b := paneTmux(t, pane, 1) - evidence, err := b.ConfirmLaunch(context.Background(), Session{PaneID: "s:1.0"}, LaunchReference) + evidence, err := b.ConfirmInput(context.Background(), Session{PaneID: "s:1.0"}, LaunchReference) if err != nil { t.Fatalf("confirm: %v", err) } @@ -354,9 +354,9 @@ exit 0 return &TmuxBackend{Binary: bin, LaunchConfirmTimeout: 2 * time.Second, LaunchConfirmPoll: 5 * time.Millisecond} } -func TestConfirmLaunchAcceptsObservedActivity(t *testing.T) { +func TestConfirmInputAcceptsObservedActivity(t *testing.T) { b := fakeTmux(t, 1, false) - evidence, err := b.ConfirmLaunch(context.Background(), Session{PaneID: "s:1.0"}, LaunchReference) + evidence, err := b.ConfirmInput(context.Background(), Session{PaneID: "s:1.0"}, LaunchReference) if err != nil { t.Fatalf("confirm: %v", err) } @@ -368,10 +368,10 @@ func TestConfirmLaunchAcceptsObservedActivity(t *testing.T) { // Prompt returning nil is not proof. An editor that still holds the text at // the deadline is a launch failure, and it must be the one class that releases // the lease instead of holding it as uncertain. -func TestConfirmLaunchFailsWhileTheEditorStillHoldsThePrompt(t *testing.T) { +func TestConfirmInputFailsWhileTheEditorStillHoldsThePrompt(t *testing.T) { b := fakeTmux(t, 1000, false) b.LaunchConfirmTimeout = 60 * time.Millisecond - _, err := b.ConfirmLaunch(context.Background(), Session{PaneID: "s:1.0"}, LaunchReference) + _, err := b.ConfirmInput(context.Background(), Session{PaneID: "s:1.0"}, LaunchReference) if !errors.Is(err, ErrPromptNotSubmitted) { t.Fatalf("err=%v, want ErrPromptNotSubmitted", err) } @@ -381,9 +381,43 @@ func TestConfirmLaunchFailsWhileTheEditorStillHoldsThePrompt(t *testing.T) { } // A confirmation that cannot read the pane must not become an acknowledgement. -func TestConfirmLaunchSurfacesTransportErrors(t *testing.T) { +func TestConfirmInputSurfacesTransportErrors(t *testing.T) { b := fakeTmux(t, 0, true) - if _, err := b.ConfirmLaunch(context.Background(), Session{PaneID: "s:1.0"}, LaunchReference); err == nil { + if _, err := b.ConfirmInput(context.Background(), Session{PaneID: "s:1.0"}, LaunchReference); err == nil { t.Fatal("a failed pane capture must not confirm a launch") } } + +// F16 and F20 are separate concerns and this is the line between them: +// Orchestra confirms delivery of what it originates, but nothing typed at a +// pane counts as the agent doing work. A capture whose only difference is the +// input line must produce the same progress digest. +func TestPaneProgressIgnoresInputLines(t *testing.T) { + body := "────\n ran the checks, nothing to change\n────\n" + idle := paneTmux(t, body+"❯ \n", 3) + typed := paneTmux(t, body+"❯ go ahead and implement it\n", 3) + a, err := idle.PaneProgress(context.Background(), Session{PaneID: "s:1.0"}) + if err != nil { + t.Fatal(err) + } + b, err := typed.PaneProgress(context.Background(), Session{PaneID: "s:1.0"}) + if err != nil { + t.Fatal(err) + } + if a != b { + t.Fatalf("typing changed the progress digest:\n%q\n%q", a, b) + } + if !strings.Contains(a, "ran the checks") { + t.Fatalf("progress digest dropped harness output: %q", a) + } + + // Harness output still moves it. + worked := paneTmux(t, "────\n edited scripts/orchestra_e2e_healthcheck.sh\n────\n❯ \n", 3) + c, err := worked.PaneProgress(context.Background(), Session{PaneID: "s:1.0"}) + if err != nil { + t.Fatal(err) + } + if c == a { + t.Fatal("real harness output left the progress digest unchanged") + } +}