package operations import ( "encoding/json" "errors" "fmt" "strings" "orchestra/internal/authz" "orchestra/internal/domain" "orchestra/internal/registry" "orchestra/internal/store" "orchestra/internal/workphase" ) // AdvanceWorkPhase moves a task to the next phase on its project's declared // path and seals the artifact the phase produced. // // Only Orchestra changes phase. An agent that believes the phase should // change says so through the approval surface, and this is what acts on that // belief. The artifact is validated before the transition is recorded, so a // phase can never be left with an artifact the next phase cannot read. // // Review is the end of the path. Its only move is back to implement, because // a review that passes ends the task through the lifecycle, not the phase. func AdvanceWorkPhase(s *store.Store, project registry.Project, taskID string, artifact []byte) (domain.Event, error) { return advanceWorkPhase(s, project, taskID, artifact, nil) } // advanceWorkPhase carries extra payload fields a specific transition needs, // such as the commit a review phase is entered against. func advanceWorkPhase(s *store.Store, project registry.Project, taskID string, artifact []byte, extra map[string]any) (domain.Event, error) { t, ok := s.Task(taskID) if !ok { return domain.Event{}, domain.ErrNotFound } next, ok := project.NextPhase(t.WorkPhase) if !ok { return domain.Event{}, fmt.Errorf("%w: work phase %q is the end of project %s's path", domain.ErrInvalid, current(t), project.ID) } // The gate sits between the sealed artifact and the next phase, so the // human confirms a direction that is already written down. if project.GateRequired(current(t), next) { switch { case trajectoryGateOpen(s, taskID): cleared, err := clearTrajectoryGate(s, t) if err != nil { return domain.Event{}, err } t = cleared case t.State == domain.StateBlocked && t.BlockReason == domain.BlockReasonTrajectoryGate: // Already waiting. Re-raising would spam the human and reset the // position the open check depends on. return domain.Event{}, fmt.Errorf("%w (task %s, %s to %s)", ErrTrajectoryGate, taskID, current(t), next) default: // The artifact is not sealed yet, so the packet reads the proposal // from the bytes in hand. The caller retries this same advance with // the same artifact once the human has answered. return domain.Event{}, raiseTrajectoryGate(s, t, current(t), next, artifact) } } payload := map[string]any{"phase": string(next), "from": string(current(t))} for k, v := range extra { payload[k] = v } if len(artifact) > 0 { // Validate against the phase being left, which is the phase that // produced this artifact. switch current(t) { case domain.WorkPhaseResearch: if _, err := workphase.DecodeResearch(artifact); err != nil { return domain.Event{}, err } case domain.WorkPhasePlan: doc, err := workphase.ParsePlan(artifact) if err != nil { return domain.Event{}, err } // Citations resolve here and nowhere else: the coordinator holds // ResearchRef, so this is the only party that can tell whether a // cited finding exists. A plan resting on a finding nobody // recorded fails on the planner, while its session is still alive // to be told, rather than on the implementer later. if err := resolvePlanReferences(s, t, doc); err != nil { return domain.Event{}, err } // The project decides what a plan command may execute, and the // brief promises the planner it learns that at seal time. Run 9 // sealed ["bash", "scripts/test_healthcheck.sh"] against a policy // that allows neither shape: the only check lived in // PlanPhaseCommands, which runs when the implementer asks to // verify, one phase and one session too late. if err := resolvePlanCommands(project, doc); err != nil { return domain.Event{}, err } } ref, err := s.PutArtifact(artifact) if err != nil { return domain.Event{}, err } payload["artifact_ref"] = ref } b, err := json.Marshal(payload) if err != nil { return domain.Event{}, err } e := domain.Event{ID: domain.NewID(), Type: domain.EventWorkPhaseChanged, TaskID: taskID, Version: t.Version + 1, Payload: b, Surface: string(authz.System)} return e, s.Append(e) } func current(t domain.Task) domain.WorkPhase { if t.WorkPhase == "" { return domain.WorkPhaseFrame } 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 } // resolvePlanCommands refuses a plan whose automated checks fall outside the // project's verification policy. It fails on the planner, whose session is // still alive to correct it, rather than on the implementer that inherits it. func resolvePlanCommands(project registry.Project, doc workphase.PlanDoc) error { for _, phase := range doc.Phases { for _, argv := range phase.Automated { if allowed, why := project.Verification.Allows(argv); !allowed { return fmt.Errorf("%w: %s verification: %s", domain.ErrInvalid, phase.ID, why) } } } return nil } // resolvePlanReferences refuses a plan that cites research the task never // sealed. Its cost is one CAS read against a ref the coordinator already // holds. func resolvePlanReferences(s *store.Store, t domain.Task, doc workphase.PlanDoc) error { if len(doc.References) == 0 { return nil } if t.ResearchRef == "" { return fmt.Errorf("%w: the plan cites %s but this task sealed no research", domain.ErrInvalid, strings.Join(doc.References, ", ")) } raw, err := s.Artifact(t.ResearchRef) if err != nil { return fmt.Errorf("resolve plan references: %w", err) } r, err := workphase.DecodeStoredResearch(raw) if err != nil { return fmt.Errorf("resolve plan references: %w", err) } if missing := doc.ResolveReferences(r); len(missing) > 0 { return fmt.Errorf("%w: the plan cites research:%s, which the accepted research does not contain", domain.ErrInvalid, strings.Join(missing, ", research:")) } return nil }