Files
orchestra/internal/domain/domain.go
T
kami 6565b9fce2 Let an expired lease still commit the anchor it pushed
A release pushes the anchor first and commits second. When the lease
expired in between, the commit could never land: the worker sent the
epoch from w.leases, which the expiry replay had already deleted, and
the coordinator refused a /handoff without a live owned lease. Run 10
lost a finished task this way, its work sitting in the worktree until
retry_limit.

The epoch now belongs to the release transaction, so it survives the
lease. TaskReleased retains the ending epoch as Task.LastLeaseEpoch,
and lateHandoffAccepted lets exactly that owner commit while the task
is queued, unleased, and carrying no handoff of its own. A successor
that has already re-leased the task holds the lease, so the late
handoff loses. Store.Append's version fence settles the race, which is
why the late path skips the version check the worker cannot satisfy.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVbaKucEYBjMqVeUgJUsc1
2026-08-28 20:57:19 +04:00

590 lines
23 KiB
Go

package domain
import (
"crypto/rand"
"crypto/sha256"
"encoding/base32"
"encoding/binary"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"strings"
"time"
)
var ErrConflict = errors.New("task version conflict")
var ErrNotFound = errors.New("task not found")
var ErrInvalid = errors.New("invalid event")
// ErrDuplicate is returned by Store.Append for a TaskCreated event whose
// (source, external_id) pair was already ingested. The caller already has a
// task for this content; nothing was appended.
var ErrDuplicate = errors.New("duplicate task ingestion")
// CurrentEventSchema is 3: schema 2 requires every event to declare its
// authorizing Surface; schema 3 adds lease fencing epochs. Older events stay
// readable so a deployment can recover its existing log before new writes
// are emitted (the store derives a non-renewable legacy epoch on replay).
// LeaseTTL is how long a lease survives without renewal, and LeaseRenewAt is
// how much remaining time makes the worker renew. Both halves read these, so a
// coordinator granting one TTL while a worker assumes another is not possible.
// Short leases are the recovery mechanism for a stalled agent: a stalled pane
// is only reclaimed when its lease runs out, and 30 minutes per window made
// run 5's review stall unbounded in practice.
const (
LeaseTTL = 5 * time.Minute
LeaseRenewAt = LeaseTTL / 2
)
const CurrentEventSchema = 3
type TaskState string
const (
StateQueued TaskState = "queued"
StateLeased TaskState = "leased"
StateCompleted TaskState = "completed"
StateFailed TaskState = "failed"
StateBlocked TaskState = "blocked"
// StateNeedsAttention records a recoverable fault without abandoning the
// current fenced lease. The owning worker may still reconcile a late
// completion, explicitly release it, or renew it while an operator
// investigates; expiry remains the only automatic reclaim.
StateNeedsAttention TaskState = "needs_attention"
// StateInReview is a submitted change waiting on the human. It is not
// completion: an agent never decides that a change shipped.
StateInReview TaskState = "in_review"
)
// BlockReason is the machine-readable diagnosis for a TaskBlocked event.
// Blocker remains the operator-facing detail; this field lets projections
// group attention without repeatedly parsing prose at read time.
type BlockReason string
const (
BlockReasonLeaseFailure BlockReason = "lease_failure"
BlockReasonWorkerOffline BlockReason = "worker_offline"
BlockReasonLeaseExpired BlockReason = "lease_expired"
BlockReasonApproval BlockReason = "approval"
BlockReasonHandoffValidation BlockReason = "handoff_validation"
BlockReasonOperator BlockReason = "operator_block"
BlockReasonSystem BlockReason = "system_error"
// BlockReasonTrajectoryGate is a deliberate stop, not a fault: the plan is
// sealed and Orchestra is waiting for the human to confirm the direction.
BlockReasonTrajectoryGate BlockReason = "trajectory_gate"
// BlockReasonHumanDecision is a bounded question the repository could not
// answer. BlockReasonOperatorRequired is what a task becomes when it has
// spent its question budget: an operator looks at it rather than the
// agent asking again.
BlockReasonHumanDecision BlockReason = "human_decision"
BlockReasonOperatorRequired BlockReason = "operator_required"
BlockReasonUnknown BlockReason = "unknown"
)
func (r BlockReason) Valid() bool {
switch r {
case BlockReasonLeaseFailure, BlockReasonWorkerOffline, BlockReasonLeaseExpired,
BlockReasonApproval, BlockReasonHandoffValidation, BlockReasonOperator,
BlockReasonSystem, BlockReasonUnknown, BlockReasonTrajectoryGate,
BlockReasonHumanDecision, BlockReasonOperatorRequired, BlockReasonPlanMismatch:
return true
}
return false
}
// InferBlockReason supplies a stable category for older events which only
// recorded a prose blocker. New producers should send block_reason directly.
func InferBlockReason(blocker string) BlockReason {
v := strings.ToLower(blocker)
switch {
case strings.Contains(v, "handoff"):
return BlockReasonHandoffValidation
case strings.Contains(v, "approval") || strings.Contains(v, "permission"):
return BlockReasonApproval
case strings.Contains(v, "expired") && strings.Contains(v, "lease"):
return BlockReasonLeaseExpired
case strings.Contains(v, "worker") && (strings.Contains(v, "offline") || strings.Contains(v, "unreachable")):
return BlockReasonWorkerOffline
case strings.Contains(v, "lease") || strings.Contains(v, "agent.start") || strings.Contains(v, "pane"):
return BlockReasonLeaseFailure
default:
return BlockReasonSystem
}
}
type Estimate struct {
Value float64 `json:"value"`
Who string `json:"who"`
Confidence float64 `json:"confidence"`
}
// SessionEvidence is captured by the machine that owns a pane immediately
// before it drops its mapping. It is deliberately observation-only: it never
// claims that a pane is still live after the worker has closed it.
type SessionEvidence struct {
PaneID string `json:"pane_id,omitempty"`
HarnessID string `json:"harness_id,omitempty"`
PaneState string `json:"pane_state,omitempty"`
Source string `json:"source,omitempty"`
CapturedAt time.Time `json:"captured_at,omitempty"`
CheckedAt time.Time `json:"checked_at,omitempty"`
}
type Lease struct {
HarnessID string `json:"harness_id"`
// Epoch is an opaque fencing token minted for every assignment. Versions
// change for ordinary lifecycle events; an epoch changes only when
// ownership changes, so an old pane can never become current again after
// a release/re-lease cycle.
Epoch string `json:"epoch"`
Until time.Time `json:"until"`
}
type Task struct {
ID string `json:"id"`
Source string `json:"source"`
ExternalID string `json:"external_id"`
Project string `json:"project"`
Capability []string `json:"capability"`
Parent string `json:"parent,omitempty"`
InherentPriority int `json:"inherent_priority"`
Due *time.Time `json:"due,omitempty"`
Estimate *Estimate `json:"estimate,omitempty"`
State TaskState `json:"state"`
Lease *Lease `json:"lease,omitempty"`
// HandoffRef survives the queued interval between TaskReleased and the
// next router-owned TaskLeased event; it is the only artifact the worker
// may use for local pickup validation.
HandoffRef string `json:"handoff_ref,omitempty"`
// ReleaseTransaction and ReleaseAnchor bind successor pickup to the exact
// durable predecessor checkpoint. They survive queueing and re-lease.
ReleaseTransaction string `json:"release_transaction,omitempty"`
ReleaseAnchor string `json:"release_anchor,omitempty"`
PickupTransaction string `json:"pickup_transaction,omitempty"`
PickupLeaseVersion int `json:"pickup_lease_version,omitempty"`
Version int `json:"version"`
Title string `json:"title,omitempty"`
Description string `json:"description,omitempty"`
Acceptance []string `json:"acceptance,omitempty"`
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"`
// 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"`
// Recovery state is part of the durable projection, never process-local
// router memory. This makes retry and operator diagnostics survive a
// coordinator restart.
Attempt int `json:"attempt,omitempty"`
NextRetryAt time.Time `json:"next_retry_at,omitempty"`
FailureClass string `json:"failure_class,omitempty"`
LifecyclePhase string `json:"lifecycle_phase,omitempty"`
// WorkPhase is the cognitive phase (frame/research/plan/implement/review),
// orthogonal to State and LifecyclePhase. Empty means frame.
WorkPhase WorkPhase `json:"work_phase,omitempty"`
// DecisionRequest is the question this task is currently blocked on. It is
// cleared when the task leaves the blocked state, because the answer then
// stands on its own as a decision and the log still holds the question.
DecisionRequest *DecisionRequest `json:"decision_request,omitempty"`
// ReviewTargetSHA is the commit the current review phase was entered
// against. A review of any other commit is not a review of this work.
ReviewTargetSHA string `json:"review_target_sha,omitempty"`
// Review is the last independent review, bound to the commit it was
// performed against. A review is never a free-floating pass: when the code
// moves, ResultSHA no longer matches and the review describes a tree that
// does not exist any more.
Review *ReviewRef `json:"review,omitempty"`
// Submission is the durable record of the change handed to the human.
Submission *SubmissionRef `json:"submission,omitempty"`
// ResearchRef and PlanRef are the sealed artifacts of the phases already
// finished. The next phase reads these, never the session that wrote them.
ResearchRef string `json:"research_ref,omitempty"`
PlanRef string `json:"plan_ref,omitempty"`
// PlanProgress is what Orchestra established about the accepted plan's
// phases. Read it through PlanPhases, which discards records belonging to
// a superseded plan.
PlanProgress *PlanProgress `json:"plan_progress,omitempty"`
// PlanHistory holds the plan refs this task worked from before the
// 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"`
}
// ReviewRef binds a sealed review artifact to one commit.
type ReviewRef struct {
ArtifactRef string `json:"artifact_ref"`
ResultSHA string `json:"result_sha"`
// Blocking is the count of blocker and important findings, projected so a
// completion check does not have to read the artifact to know the answer.
Blocking int `json:"blocking"`
}
// EventReviewRecorded seals one independent review. Orchestra emits it; the
// reviewing session only supplies the findings.
const EventReviewRecorded = "ReviewRecorded"
// ReviewSatisfied reports whether this task holds an accepted review of the
// exact commit named. It is the mechanical half of completion eligibility.
func (t Task) ReviewSatisfied(resultSHA string) bool {
return t.Review != nil && t.Review.ResultSHA == resultSHA && t.Review.Blocking == 0
}
type Event struct {
SchemaVersion int `json:"schema_version,omitempty"`
Seq uint64 `json:"seq"`
ID string `json:"id"`
Type string `json:"type"`
TaskID string `json:"task_id"`
Version int `json:"version"`
At time.Time `json:"at"`
Payload json.RawMessage `json:"payload"`
// Surface identifies the bus capability the emitter is authorized under
// (see internal/authz). It is required on every event so authorization is
// enforced once, at the store append boundary, regardless of whether the
// emitter reached the store over HTTP, from the router, from a harness
// adapter, or from a provider.
Surface string `json:"surface"`
}
func Hash(v []byte) string { h := sha256.Sum256(v); return hex.EncodeToString(h[:]) }
// NewID returns a sortable, 128-bit ULID-like identifier using the canonical
// 48-bit millisecond timestamp plus 80 bits of cryptographic randomness.
var ulidEncoding = base32.NewEncoding("0123456789ABCDEFGHJKMNPQRSTVWXYZ").WithPadding(base32.NoPadding)
func NewID() string {
b := make([]byte, 16)
binary.BigEndian.PutUint64(b[:8], uint64(time.Now().UnixMilli())<<16)
_, _ = rand.Read(b[6:])
return ulidEncoding.EncodeToString(b)
}
func ValidateEvent(e Event) error {
if e.SchemaVersion > CurrentEventSchema || e.Type == "" || e.TaskID == "" || len(e.Payload) == 0 || len(e.Payload) > 64*1024 {
return ErrInvalid
}
if e.SchemaVersion >= 2 && strings.TrimSpace(e.Surface) == "" {
return fmt.Errorf("%w: surface required", ErrInvalid)
}
allowed := map[string]bool{"TaskCreated": true, "TaskLeased": true, "TaskLeaseRenewed": true, "TaskReleased": true, "TaskLaunchAcknowledged": true, "TaskPickupValidated": true, "TaskCompleted": true, "TaskFailed": true, "TaskBlocked": true, "TaskNeedsAttention": true, "ApprovalRequested": true, "ApprovalGranted": true, "ApprovalDenied": true, "TaskAmended": true, "TaskCorrected": true, "QuotaReported": true, "StandupAdvisory": true, EventHumanDecisionRecorded: true, EventHumanDecisionSuperseded: true, EventWorkPhaseChanged: true, EventDeferredFindingRecorded: true, EventReviewRecorded: true, EventTaskSubmitted: true, EventTaskChangesRequested: true, EventPlanPhaseVerified: true, EventPlanMismatchRecorded: true}
if !allowed[e.Type] {
return fmt.Errorf("%w: unknown type %q", ErrInvalid, e.Type)
}
var p map[string]any
if err := json.Unmarshal(e.Payload, &p); err != nil {
return fmt.Errorf("%w: payload is not JSON", ErrInvalid)
}
if p == nil {
return fmt.Errorf("%w: payload must be an object", ErrInvalid)
}
if err := ValidatePayload(e.Type, p); err != nil {
return err
}
if e.SchemaVersion >= 3 {
switch e.Type {
case "TaskLeased", "TaskLeaseRenewed", "TaskLaunchAcknowledged", "TaskPickupValidated":
if v, ok := p["lease_epoch"].(string); !ok || strings.TrimSpace(v) == "" {
return fmt.Errorf("%w: lease_epoch required", ErrInvalid)
}
}
}
return nil
}
func ValidateCreated(p map[string]any) error {
for _, k := range []string{"source", "external_id", "project"} {
if s, ok := p[k].(string); !ok || strings.TrimSpace(s) == "" {
return fmt.Errorf("%w: %s required", ErrInvalid, k)
}
}
return nil
}
func ValidatePayload(typ string, p map[string]any) error {
requiredString := func(key string) error {
v, ok := p[key].(string)
if !ok || strings.TrimSpace(v) == "" {
return fmt.Errorf("%w: %s required", ErrInvalid, key)
}
return nil
}
switch typ {
case "TaskCreated":
return ValidateCreated(p)
case "TaskLeased":
if err := requiredString("harness_id"); err != nil {
return err
}
// Validation of a durable event must not depend on the current clock.
// Comparing until_ns against time.Now() here made every lease event
// fail validation once it expired, so replaying the log after a
// restart refused the store's own history and the coordinator could
// not start at all. Freshness is a lease question, answered by
// Store.Lease and Store.ExpireLeases; well-formedness is this
// function's question.
until, untilOK := p["until_ns"].(float64)
if ttl, ok := p["ttl"].(float64); ok {
if ttl <= 0 {
return fmt.Errorf("%w: ttl invalid", ErrInvalid)
}
} else if !untilOK || until <= 0 {
return fmt.Errorf("%w: ttl required", ErrInvalid)
}
if v, ok := p["expected_version"].(float64); !ok || v < 0 || v != float64(int(v)) {
return fmt.Errorf("%w: expected_version invalid", ErrInvalid)
}
case "TaskLeaseRenewed":
if err := requiredString("harness_id"); err != nil {
return err
}
// Time-independent for the same reason as TaskLeased above.
until, ok := p["until_ns"].(float64)
if !ok || until <= 0 {
return fmt.Errorf("%w: until_ns required", ErrInvalid)
}
if v, ok := p["expected_version"].(float64); !ok || v < 0 || v != float64(int(v)) {
return fmt.Errorf("%w: expected_version invalid", ErrInvalid)
}
case "TaskReleased":
if err := requiredString("handoff_ref"); err != nil && p["reason"] == nil {
return err
}
case "TaskPickupValidated":
for _, key := range []string{"transaction_id", "handoff_ref", "anchor_sha", "harness_id"} {
if err := requiredString(key); err != nil {
return err
}
}
if err := requiredHash(p, "handoff_ref"); err != nil {
return err
}
if v, ok := p["anchor_sha"].(string); !ok || len(v) != 40 {
return fmt.Errorf("%w: anchor_sha invalid", ErrInvalid)
}
if v, ok := p["lease_version"].(float64); !ok || v < 1 || v != float64(int(v)) {
return fmt.Errorf("%w: lease_version invalid", ErrInvalid)
}
if _, ok := p["handoff_ref"]; ok {
if err := requiredHash(p, "handoff_ref"); err != nil {
return err
}
v, ok := p["anchor_sha"].(string)
if !ok || len(v) != 40 || strings.TrimSpace(v) != v {
return fmt.Errorf("%w: anchor_sha invalid", ErrInvalid)
}
}
case "TaskCompleted":
if err := requiredString("report_ref"); err != nil {
return err
}
if err := requiredHash(p, "report_ref"); err != nil {
return err
}
if receipt, ok := p["receipt"].(map[string]any); !ok || len(receipt) == 0 {
return fmt.Errorf("%w: receipt required", ErrInvalid)
}
if v, ok := p["result_sha"]; ok {
if s, ok := v.(string); !ok || len(s) != 40 {
return fmt.Errorf("%w: result_sha invalid", ErrInvalid)
}
if err := requiredString("branch"); err != nil {
return err
}
if err := requiredString("remote"); err != nil {
return err
}
}
case "TaskFailed":
if err := requiredString("reason"); err != nil {
return err
}
case "TaskLaunchAcknowledged":
if err := requiredString("harness_id"); err != nil {
return err
}
case "TaskBlocked", "TaskNeedsAttention":
if err := requiredString("blocker"); err != nil {
return err
}
if v, ok := p["block_reason"]; ok {
s, ok := v.(string)
if !ok || !BlockReason(s).Valid() {
return fmt.Errorf("%w: block_reason invalid", ErrInvalid)
}
}
if _, ok := p["handoff_ref"]; ok {
if err := requiredHash(p, "handoff_ref"); err != nil {
return err
}
}
if v, ok := p["pane_state"]; ok && v != "open" && v != "closed" && v != "unreachable" && v != "unknown" {
return fmt.Errorf("%w: pane_state invalid", ErrInvalid)
}
if v, ok := p["decision_request"]; ok {
m, ok := v.(map[string]any)
if !ok {
return fmt.Errorf("%w: decision_request must be an object", ErrInvalid)
}
if err := decodeDecisionRequest(m).Validate(); err != nil {
return err
}
}
case "TaskAmended":
if len(p) == 0 {
return fmt.Errorf("%w: amendment cannot be empty", ErrInvalid)
}
case "TaskCorrected":
// §3.1: "a wrong event is never edited; a compensating event is
// appended and replay sees both." `corrects` names the event this one
// reverses/repairs — existence against the log is checked in
// Store.Append, where the log is visible; ValidatePayload only knows
// shape.
if err := requiredString("corrects"); err != nil {
return err
}
if v, ok := p["state"]; ok {
s, ok := v.(string)
if !ok {
return fmt.Errorf("%w: state must be a string", ErrInvalid)
}
switch TaskState(s) {
case StateQueued, StateLeased, StateCompleted, StateFailed, StateBlocked, StateNeedsAttention, StateInReview:
default:
return fmt.Errorf("%w: state invalid", ErrInvalid)
}
}
if v, ok := p["attempt"]; ok {
n, ok := v.(float64)
if !ok || n < 0 || n != float64(int(n)) {
return fmt.Errorf("%w: attempt must be a non-negative whole number", ErrInvalid)
}
}
if v, ok := p["next_retry_at"]; ok {
s, ok := v.(string)
if !ok {
return fmt.Errorf("%w: next_retry_at must be a string", ErrInvalid)
}
if s != "" {
if _, err := time.Parse(time.RFC3339, s); err != nil {
return fmt.Errorf("%w: next_retry_at must be RFC3339 or empty", ErrInvalid)
}
}
}
if len(p) < 2 {
return fmt.Errorf("%w: correction must change at least one field", ErrInvalid)
}
case "ApprovalRequested":
for _, k := range []string{"subject_ref", "options"} {
if _, ok := p[k]; !ok {
return fmt.Errorf("%w: %s required", ErrInvalid, k)
}
}
case "ApprovalGranted", "ApprovalDenied":
if err := requiredString("subject_ref"); err != nil {
return err
}
case "QuotaReported":
if err := requiredString("harness_id"); err != nil {
return err
}
if v, ok := p["consumed"].(float64); !ok || v < 0 {
return fmt.Errorf("%w: consumed required", ErrInvalid)
}
case "StandupAdvisory":
if _, ok := p["items"]; !ok {
return fmt.Errorf("%w: items required", ErrInvalid)
}
case EventHumanDecisionRecorded:
for _, k := range []string{"decision_id", "kind", "subject", "value"} {
if err := requiredString(k); err != nil {
return err
}
}
if kind, _ := p["kind"].(string); !HumanDecisionKind(kind).Valid() {
return fmt.Errorf("%w: kind invalid", ErrInvalid)
}
src, ok := p["source"].(map[string]any)
if !ok {
return fmt.Errorf("%w: source required", ErrInvalid)
}
if v, ok := src["provider"].(string); !ok || strings.TrimSpace(v) == "" {
return fmt.Errorf("%w: source.provider required", ErrInvalid)
}
if v, ok := p["supersedes"]; ok {
list, ok := v.([]any)
if !ok {
return fmt.Errorf("%w: supersedes must be an array", ErrInvalid)
}
for _, item := range list {
if s, ok := item.(string); !ok || strings.TrimSpace(s) == "" {
return fmt.Errorf("%w: supersedes entries must be decision ids", ErrInvalid)
}
}
}
case EventHumanDecisionSuperseded:
if err := requiredString("decision_id"); err != nil {
return err
}
case EventWorkPhaseChanged:
return ValidateWorkPhaseChanged(p)
case EventTaskSubmitted:
return ValidateTaskSubmitted(p)
case EventTaskChangesRequested:
if v, ok := p["submitted_sha"].(string); !ok || len(v) != 40 {
return fmt.Errorf("%w: submitted_sha invalid", ErrInvalid)
}
if v, ok := p["submission_event"].(string); !ok || strings.TrimSpace(v) == "" {
return fmt.Errorf("%w: submission_event required", ErrInvalid)
}
ids, ok := p["decision_ids"].([]any)
if !ok || len(ids) == 0 {
return fmt.Errorf("%w: decision_ids required", ErrInvalid)
}
for _, id := range ids {
if s, ok := id.(string); !ok || strings.TrimSpace(s) == "" {
return fmt.Errorf("%w: decision_ids entries must be ids", ErrInvalid)
}
}
case EventPlanPhaseVerified:
return ValidatePlanPhaseVerified(p)
case EventPlanMismatchRecorded:
return ValidatePlanMismatchRecorded(p)
case EventReviewRecorded:
if err := requiredHash(p, "artifact_ref"); err != nil {
return err
}
if v, ok := p["result_sha"].(string); !ok || len(v) != 40 {
return fmt.Errorf("%w: result_sha invalid", ErrInvalid)
}
if v, ok := p["blocking"].(float64); !ok || v < 0 || v != float64(int(v)) {
return fmt.Errorf("%w: blocking invalid", ErrInvalid)
}
case EventDeferredFindingRecorded:
f := DeferredFinding{}
f.Summary, _ = p["summary"].(string)
f.Why, _ = p["why"].(string)
return f.Validate()
}
return nil
}
func requiredHash(p map[string]any, key string) error {
v, ok := p[key].(string)
if !ok || len(v) != 64 {
return fmt.Errorf("%w: %s must be sha256", ErrInvalid, key)
}
if _, err := hex.DecodeString(v); err != nil {
return fmt.Errorf("%w: %s must be sha256", ErrInvalid, key)
}
return nil
}