orchestra/internal/operations/workphase.go

package operations

import (
	"encoding/json"
	"errors"
	"fmt"

	"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:
			if _, err := workphase.DecodePlan(artifact); 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
}