orchestration

package
v0.3.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 8, 2026 License: MIT Imports: 9 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	ACPKindDispatch = "dispatch"
	ACPKindClaim    = "claim"
	ACPKindReport   = "report"
)

ACP event kinds. claim/report carry the worker-rigor fields (spec 10 R3); dispatch is the brain's own record.

Variables

View Source 
var ErrDuplicateMission = errors.New("duplicate mission id")

ErrDuplicateMission is returned when a dispatch with a mission id already in the ledger is appended again. It is the invariant that makes crash-recovery re-issue idempotent: the same deterministic mission id can appear at most once.

View Source 
var ErrSessionRevisionConflict = errors.New("session revision conflict")

Functions

func AppendACP 

func AppendACP(path string, event ACPEvent) error

AppendACP appends one event to the ledger. It is O(1): Seq is a read-time projection (ReadACP numbers events by position), so a write never re-reads the whole file — open, append, fsync. ponytail: the semantic appends (AppendClaim's attempt count, AppendDispatch's duplicate-mission guard) still scan the ledger, but that read is their actual job and the per-spec ledger is bounded by task count × attempts. Maintain an on-disk index only if a ledger ever grows past a few thousand events.

func AppendClaim 

func AppendClaim(path string, event ACPEvent) error

AppendClaim appends a claim event, filling the attempt number from the current ledger when the caller left it zero. The read-then-append happens under the caller's spec lock so the derived attempt is race-free.

func AppendDispatch 

func AppendDispatch(path string, event ACPEvent) error

AppendDispatch appends a dispatch event, refusing a duplicate mission id (spec 07 R3). The read-then-append runs under the caller's spec lock so the duplicate check is race-free.

func CheckpointPath 

func CheckpointPath(root, slug string) string

CheckpointPath is the per-spec write-ahead checkpoint file.

func HasLiveLease 

func HasLiveLease(leases []Lease, now time.Time) bool

HasLiveLease reports whether any lease is still within its TTL at now. A live lease means another controller is actively holding work and blocks a second start/resume; only expired/orphaned leases are recoverable by resume (R5).

func HasMission 

func HasMission(events []ACPEvent, missionID string) bool

HasMission reports whether the ledger already carries an event with missionID.

func MissionID 

func MissionID(sessionID string, step int, taskID string) string

MissionID is the deterministic identifier for a dispatch: session id + step + task. Determinism is the point — a dispatch re-issued after a crash on resume reuses the same mission id, so the ledger's duplicate guard makes re-issue idempotent (spec 07 R3).

func NextAttempt 

func NextAttempt(events []ACPEvent, taskID string) int

NextAttempt is the attempt number for a new claim on taskID: the count of prior claim events for that task plus one. It is a countable fact derived from the ledger under the spec lock, never a stored counter that can skew (spec 10 R3). It feeds spec 06's escalation counting.

func SaveCheckpoint 

func SaveCheckpoint(root, path string, cp Checkpoint) error

SaveCheckpoint writes the checkpoint durably (atomic write + fsync via core.AtomicWrite) under the spec lock. It must return before the caller makes the corresponding dispatch visible in the ledger (write-ahead ordering).

func SaveSessionCAS 

func SaveSessionCAS(root, path string, expectedRevision int64, next Session) error

Types

type ACPEvent 

type ACPEvent struct {
	Seq     int       `json:"seq"`
	Time    time.Time `json:"time"`
	Kind    string    `json:"kind"`
	TaskID  string    `json:"task_id,omitempty"`
	Payload string    `json:"payload,omitempty"`
	// MissionID is the deterministic dispatch identifier (session/step/task, spec
	// 07 R3). Optional so bare dispatch events and pre-spec-07 ledgers stay valid.
	// It is the key the resume path and the duplicate guard match on.
	MissionID string `json:"mission_id,omitempty"`

	// Worker-rigor fields (spec 10 R3), all optional so a bare dispatch event and
	// pre-telemetry ledgers stay valid. Attempt is monotonic per task (see
	// NextAttempt); GitHead pins the claim/report to a commit; ChangedFiles is the
	// worker-reported diff surface; VerifyRef points at the verify evidence
	// backing a report; Telemetry is verbatim worker cost.
	Attempt      int               `json:"attempt,omitempty"`
	GitHead      string            `json:"git_head,omitempty"`
	ChangedFiles []string          `json:"changed_files,omitempty"`
	VerifyRef    string            `json:"verify_ref,omitempty"`
	Telemetry    *core.Annotations `json:"telemetry,omitempty"`
}

func MissionEvent 

func MissionEvent(events []ACPEvent, missionID string) (ACPEvent, bool)

MissionEvent returns the first event carrying missionID, or false.

func ReadACP 

func ReadACP(path string) ([]ACPEvent, error)

ReadACP loads the ledger and numbers each event by position (Seq = 1-based index). It reads the whole file at once — no per-line size cap, so a large event (ChangedFiles/Payload/Telemetry) never trips the old 64KB bufio.Scanner limit.

type Action 

type Action string
const (
	ActionWait     Action = "wait"
	ActionDispatch Action = "dispatch"
	ActionHalt     Action = "halt"
	ActionTimeout  Action = "timeout"
	ActionEscalate Action = "escalate"
)

type Authority 

type Authority struct {
	Enabled bool
}

func (Authority) CanClearGate 

func (authority Authority) CanClearGate(severity GateSeverity) bool

func (Authority) CanDispatch 

func (authority Authority) CanDispatch() bool

type Checkpoint 

type Checkpoint struct {
	SessionID string    `json:"session_id"`
	Step      int       `json:"step"`
	Wave      int       `json:"wave"`
	Decision  string    `json:"decision"`
	MissionID string    `json:"mission_id,omitempty"`
	TaskID    string    `json:"task_id,omitempty"`
	Lease     *Lease    `json:"lease,omitempty"`
	Time      time.Time `json:"time"`
}

Checkpoint is the write-ahead record the controller makes durable BEFORE a dispatch becomes visible in the ledger (spec 07 R1). If the process dies after the checkpoint but before the ACP dispatch, resume finds the checkpoint's mission id absent from the ledger and re-issues exactly that dispatch; if it dies after the ledger append, resume finds the mission id present and does not re-issue — this is what makes crash recovery converge with zero double-dispatch.

func LoadCheckpoint 

func LoadCheckpoint(path string) (Checkpoint, bool, error)

LoadCheckpoint reads the checkpoint. The bool is false when none exists yet.

type Decision 

type Decision struct {
	Action Action
	TaskID string
	Reason string
}

func Decide 

func Decide(snapshot Snapshot, limits DecisionLimits) Decision

func DispatchFrontier 

func DispatchFrontier(snapshot Snapshot, limits DecisionLimits, dispatcher Dispatcher) (Decision, error)

func EvaluateBrakes 

func EvaluateBrakes(snapshot Snapshot, limits DecisionLimits) Decision

type DecisionLimits 

type DecisionLimits struct {
	MaxCost       int
	Deadline      time.Time
	MaxRetries    int
	AllowDispatch bool
}

func DecisionLimitsForAuthority 

func DecisionLimitsForAuthority(authority Authority, limits DecisionLimits) DecisionLimits

type Dispatcher 

type Dispatcher interface {
	Dispatch(task core.FrontierTask) error
}

type GateSeverity 

type GateSeverity string
const (
	GateLow      GateSeverity = "low"
	GateMedium   GateSeverity = "medium"
	GateHigh     GateSeverity = "high"
	GateCritical GateSeverity = "critical"
)

type Lease 

type Lease struct {
	TaskID    string    `json:"task_id"`
	WorkerID  string    `json:"worker_id"`
	ExpiresAt time.Time `json:"expires_at"`
	Retries   int       `json:"retries"`
}

type Reclaim 

type Reclaim struct {
	TaskID string
	Retry  bool
	Reason string
}

func Escalation 

func Escalation(leases []Lease, maxRetries int, now time.Time) Reclaim

func ReclaimExpired 

func ReclaimExpired(leases []Lease, now time.Time, maxRetries int) []Reclaim

type ResumePlan 

type ResumePlan struct {
	Reissue  bool
	Mission  Checkpoint
	Conflict string
}

ResumePlan is the reconciliation of the write-ahead checkpoint against the ACP ledger (spec 07 R3/R4). Exactly one of the outcomes holds:

  • Conflict != "": the checkpoint and ledger disagree irreconcilably; resume must refuse (exit 1) rather than guess.
  • Reissue == true: the checkpoint's mission never reached the ledger (crash between checkpoint and dispatch); resume re-issues exactly that dispatch.
  • both zero: the checkpoint's mission is already in the ledger (or there is no outstanding mission); resume only reconstructs session state.

func PlanResume 

func PlanResume(cp Checkpoint, cpExists bool, events []ACPEvent) ResumePlan

PlanResume computes the resume plan from the checkpoint and the ledger. It is pure: the caller supplies the loaded checkpoint (with existence flag) and the ledger events.

type Session 

type Session struct {
	// ID is a stable identifier minted at `brain start`, used to derive
	// deterministic mission ids (session/step/task) that survive a resume.
	ID       string       `json:"id,omitempty"`
	Revision int64        `json:"revision"`
	State    SessionState `json:"state,omitempty"`
	// Step is the controller tick counter, incremented per dispatch. It feeds the
	// mission id so a re-issued dispatch after resume reuses the same id.
	Step   int     `json:"step,omitempty"`
	Leases []Lease `json:"leases,omitempty"`
}

func LoadSession 

func LoadSession(path string) (Session, error)

func (Session) IsTerminal 

func (s Session) IsTerminal() bool

IsTerminal reports whether the session refuses further step/run/resume.

func (Session) Status 

func (s Session) Status() SessionState

Status returns the effective lifecycle state, treating the empty zero value as running (pre-spec-07 compatibility).

type SessionState 

type SessionState string

SessionState is the controller lifecycle (spec 07 R2). The persisted state machine is running → {cancelled | complete}; both terminal states refuse further step/run. An empty state decodes as running so pre-spec-07 sessions stay valid. "crashed" is never persisted — it is derived by `brain status` from a checkpoint that outran the ledger (see DeriveStatus). 

const (
	SessionRunning   SessionState = "running"
	SessionCancelled SessionState = "cancelled"
	SessionComplete  SessionState = "complete"
	SessionCrashed   SessionState = "crashed"
)

func DeriveStatus 

func DeriveStatus(session Session, cp Checkpoint, cpExists bool, events []ACPEvent) SessionState

DeriveStatus is the effective status reported by `brain status`. A terminal session reports its persisted state. A running session whose checkpoint names a mission missing from the ledger is reported as crashed: the process died mid-dispatch and a resume is needed to converge (spec 07 R6).

type Snapshot 

type Snapshot struct {
	Revision int64
	Phase    core.Phase
	Records  map[string]json.RawMessage
	Frontier []core.FrontierTask
	Leases   []Lease
	Now      time.Time
	Cost     int
}

func Sense 

func Sense(state core.State, frontier []core.FrontierTask, leases []Lease, now time.Time) Snapshot

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.