diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index e1b1c41..1c98773 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -1831,6 +1831,25 @@ func main() { coordinator.ReconcileFailureHandoff = v } } + // The third reconciliation point, and the same reasoning the submitted + // work loop above already uses: a blocked task cannot be leased, so a + // pre-lease hook can never see the reply that should unblock it (F64). + // Orchestra stopped and asked the human a question; nothing else here + // will go and read the answer. + go func() { + ticker := time.NewTicker(time.Minute) + defer ticker.Stop() + for range ticker.C { + for _, t := range s.Tasks() { + if t.State != domain.StateBlocked || !domain.BlockReasonAwaitsReply(t.BlockReason) { + continue + } + if err := reconciler.Reconcile(context.Background(), t.ID); err != nil { + log.Printf("reconcile blocked task %s: %v", t.ID, err) + } + } + } + }() } if path := os.Getenv("ORCHESTRA_JSONL"); path != "" { sup := &provider.Supervisor{Name: "jsonl", Run: func(ctx context.Context) error { diff --git a/internal/domain/domain.go b/internal/domain/domain.go index 55bbfb9..dd17a3e 100644 --- a/internal/domain/domain.go +++ b/internal/domain/domain.go @@ -82,6 +82,23 @@ const ( BlockReasonUnknown BlockReason = "unknown" ) +// BlockReasonAwaitsReply reports whether this stop is a question waiting on a +// human, rather than a fault or a budget. Every reason listed here resumes on +// a reply, so both the loop that goes and reads replies and the loop that +// returns answered tasks to the queue must agree on the set. They did not: +// plan_mismatch blocked for an answer that nothing ever went to fetch, and +// nothing would have resumed it if it had (F64). +// +// operator_required is deliberately absent. A task that has spent its question +// budget continues when an operator says so, not when someone replies. +func BlockReasonAwaitsReply(r BlockReason) bool { + switch r { + case BlockReasonHumanDecision, BlockReasonTrajectoryGate, BlockReasonPlanMismatch: + return true + } + return false +} + func (r BlockReason) Valid() bool { switch r { case BlockReasonLeaseFailure, BlockReasonWorkerOffline, BlockReasonLeaseExpired, @@ -168,19 +185,19 @@ type Task struct { QualityGate string `json:"quality_gate,omitempty"` // Block evidence is projected from TaskBlocked so terminal records remain // diagnosable after the live coordinator mapping is gone. - Blocker string `json:"blocker,omitempty"` - BlockReason BlockReason `json:"block_reason,omitempty"` - BlockedAt time.Time `json:"blocked_at,omitempty"` - LastPaneID string `json:"last_pane_id,omitempty"` - LastHarness string `json:"last_harness_id,omitempty"` + Blocker string `json:"blocker,omitempty"` + BlockReason BlockReason `json:"block_reason,omitempty"` + BlockedAt time.Time `json:"blocked_at,omitempty"` + LastPaneID string `json:"last_pane_id,omitempty"` + LastHarness string `json:"last_harness_id,omitempty"` // LastLeaseEpoch is the fencing token of the lease that most recently // ended. A worker can push its release anchor and only then discover the // lease expired; the finished work is durable in git but the commit can // never land. Retaining the epoch lets exactly that owner still commit // while the task sits unleased. - LastLeaseEpoch string `json:"last_lease_epoch,omitempty"` - PaneState string `json:"pane_state,omitempty"` // open, closed, unreachable, unknown - LastSession SessionEvidence `json:"last_session,omitempty"` + LastLeaseEpoch string `json:"last_lease_epoch,omitempty"` + PaneState string `json:"pane_state,omitempty"` // open, closed, unreachable, unknown + LastSession SessionEvidence `json:"last_session,omitempty"` // Recovery state is part of the durable projection, never process-local // router memory. This makes retry and operator diagnostics survive a // coordinator restart. @@ -217,7 +234,7 @@ type Task struct { // current one, oldest first. A superseded plan stays queryable: the // verification recorded against it is provenance, not garbage. PlanHistory []string `json:"plan_history,omitempty"` - LastError string `json:"last_error,omitempty"` + LastError string `json:"last_error,omitempty"` } // ReviewRef binds a sealed review artifact to one commit. diff --git a/internal/operations/human_decision.go b/internal/operations/human_decision.go index a17ff03..ca2224c 100644 --- a/internal/operations/human_decision.go +++ b/internal/operations/human_decision.go @@ -107,11 +107,9 @@ func ResumeAnsweredBlockers(s *store.Store) ([]domain.Event, error) { if t.State != domain.StateBlocked { continue } - switch t.BlockReason { - case domain.BlockReasonHumanDecision, domain.BlockReasonTrajectoryGate: - default: - // operator_required is deliberately not resumed by a reply. An - // operator decides when a task that spent its budget continues. + // operator_required is deliberately not resumed by a reply. An + // operator decides when a task that spent its budget continues. + if !domain.BlockReasonAwaitsReply(t.BlockReason) { continue } if !blockerAnswered(s, t.ID, t.BlockReason) { diff --git a/internal/operations/planmismatch_test.go b/internal/operations/planmismatch_test.go index 388fcf6..8e5d658 100644 --- a/internal/operations/planmismatch_test.go +++ b/internal/operations/planmismatch_test.go @@ -215,7 +215,16 @@ func TestHumanAnswerResumesTheSamePlanWithoutResealing(t *testing.T) { if !PlanMismatchAnswered(s, id) { t.Fatal("the human answered and the task is still waiting") } + // F64: asserting the predicate is not asserting the resume. This test + // passed for as long as the predicate had no caller, while a task blocked + // on a mismatch stayed blocked forever however the human replied. + if events, err := ResumeAnsweredBlockers(s); err != nil || len(events) != 1 { + t.Fatalf("an answered mismatch did not return to the queue: events=%v err=%v", events, err) + } after, _ := s.Task(id) + if after.State != domain.StateQueued { + t.Fatalf("state = %s after the human answered, want queued", after.State) + } if after.PlanRef != planRef { t.Fatal("answering the question replaced the plan") }