engine

package
v1.17.1 Latest Latest
Warning

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

 Go to latest Published: Jul 21, 2026 License: Apache-2.0 Imports: 16 Imported by: 0

Documentation

Overview

Package engine holds Harbor's shared TaskRegistry state machine: the task/group/patch lifecycle, idempotency dedup, cascade-cancel, group governance (seal/resolve/fail-fast), retain-turn waiters, and WatchGroup fan-out. It is the single implementation every TaskRegistry driver wraps; drivers differ ONLY in how records are persisted, expressed through the Backend seam.

The engine is persistence-agnostic: it never imports the StateStore. On construction it calls Backend.Hydrate to rebuild its in-memory indices, and on every mutation it writes through Backend.SaveTask / Backend.SaveGroup / Backend.SavePatch. The in-process driver supplies an ephemeral backend (no reload across restarts); the durable driver supplies a StateStore-backed backend whose records survive a restart. A backend implementation MAY block on I/O — the engine holds its lock across a Save, so a slow backend serializes mutations on that engine instance (acceptable for the single-instance durability the durable driver targets).

Internal model:

  • A primary `map[TaskID]*Task` holds the live task state.
  • A secondary `map[idempotencyKey]idempotencyRecord` resolves `(SessionID, IdempotencyKey)` lookups for `Spawn` dedup.
  • A children index `map[TaskID][]TaskID` powers cascade-cancel BFS without scanning the primary map.
  • A single `sync.RWMutex` guards every map.
  • Caller-controlled strings (Description, Query, Result.Value, Error.Message) are run through the `audit.Redactor` BEFORE the record is handed to the backend. The redactor's reflective walk returns a `map[string]any`; we read the redacted strings back and replace them on the Task before persisting.
  • `Close(ctx)` flips an atomic flag; subsequent calls return `ErrRegistryClosed`. There are no engine-owned goroutines to join, so Close is fast.

Index

Constants

View Source 
const RecoveryErrorCode = "runtime_restarted"

RecoveryErrorCode is the reserved TaskError.Code stamped on a task the recovery sweep transitions to Failed because it was left Running by a runtime restart. Consumers (the Console, lineage tooling) match on it to distinguish a restart-interrupted task from a task that failed during normal execution.

Variables

This section is empty.

Functions

This section is empty.

Types

type Backend 

type Backend interface {
	// SaveTask persists a task record (the task plus the engine's
	// idempotency content hash, which is not recoverable from the
	// post-redaction task alone).
	SaveTask(ctx context.Context, rec TaskRecord) error
	// DeleteTask removes a task's persisted record. The engine calls it
	// to COMPENSATE a half-completed spawn (a task persisted before its
	// group wiring failed), so a restart never resurrects a spawn the
	// caller was told failed. A backend that never reloads (ephemeral)
	// may no-op. Deleting an absent record is not an error.
	DeleteTask(ctx context.Context, t *tasks.Task) error
	// SaveGroup persists a task-group record.
	SaveGroup(ctx context.Context, g *tasks.TaskGroup) error
	// SavePatch persists a patch record.
	SavePatch(ctx context.Context, p *tasks.Patch) error
	// Hydrate returns every persisted record so [New] can rebuild the
	// engine's in-memory state. An ephemeral backend returns an empty
	// Snapshot.
	Hydrate(ctx context.Context) (Snapshot, error)
}

Backend is the persistence seam the engine writes through. It is the ONLY difference between TaskRegistry drivers: the engine owns the lifecycle state machine; a Backend owns where records live and whether they survive a restart.

Two implementations ship:

  • the in-process driver's ephemeral backend, whose Hydrate returns an empty Snapshot (records never reload across a restart); and
  • the durable driver's StateStore-backed backend, whose records are keyed per-record and whose Hydrate replays them on open.

Contract:

  • Save* MUST be idempotent on the record's identity: a re-save of the same task/group/patch overwrites in place (the engine re-persists the full record on every mutation, and the recovery sweep re-persists a recovered record). A backend that appended instead of overwriting would accumulate stale duplicates.
  • Save* MAY block on I/O. The engine calls Save* while holding its lock, so a slow backend serializes mutations on that engine instance — acceptable for single-instance durability.
  • A Save* that cannot serialize a record (e.g. a non-serializable payload) MUST return an error, never silently drop it (CLAUDE.md §5 fail-loud).
  • Hydrate is called exactly once, from New, before the engine is handed to any caller. It returns every persisted record so the engine can rebuild its indices. An error fails New loudly.

type Engine 

type Engine struct {
	// contains filtered or unexported fields
}

func New 

func New(bus events.EventBus, redactor audit.Redactor, backend Backend) (*Engine, error)

New constructs an Engine over the given event bus, redactor, and persistence backend, then hydrates its in-memory state from the backend. Bus, redactor, and backend are all mandatory; a nil argument fails loudly (CLAUDE.md §5).

Hydration uses context.Background(): New runs at boot, before the engine is handed to any caller, so there is no request context to thread and no concurrent access to race. A backend whose Hydrate returns an error fails New loudly rather than booting with partial state.

func (*Engine) AcknowledgeBackground 

func (e *Engine) AcknowledgeBackground(ctx context.Context, sessionID identity.Identity, ids []tasks.TaskID) (int, error)

AcknowledgeBackground implements tasks.TaskRegistry. Marks the given tasks as user-acknowledged. Returns the count of tasks that transitioned; emits one `task.background_acknowledged` event per transition. Unknown task IDs are silently skipped.

func (*Engine) AddMemberToGroup 

func (e *Engine) AddMemberToGroup(ctx context.Context, gid tasks.TaskGroupID, tid tasks.TaskID) error

AddMemberToGroup is a driver-internal (non-interface) helper the conformance suite uses to wire a freshly-spawned task into a group. The tool dispatcher will route this through the SpawnTool's GroupID parameter; until then the helper exposes the seam directly so the conformance subtests can exercise the resolve gate without waiting for the tool-dispatch wiring.

Returns `ErrGroupNotFound` on an unknown group or a cross-session access attempt; `ErrGroupSealed` on a sealed or terminal group; `ErrNotFound` on an unknown task.

func (*Engine) ApplyGroup 

func (e *Engine) ApplyGroup(ctx context.Context, id tasks.TaskGroupID, action tasks.GroupAction) error

ApplyGroup implements tasks.TaskRegistry. Convenience dispatch.

func (*Engine) ApplyPatch 

func (e *Engine) ApplyPatch(ctx context.Context, sessionID identity.Identity, patchID string, action tasks.PatchAction) (bool, error)

ApplyPatch implements tasks.TaskRegistry.

func (*Engine) Cancel 

func (e *Engine) Cancel(ctx context.Context, id tasks.TaskID, reason string) (bool, error)

Cancel implements tasks.TaskRegistry. The target transitions unconditionally (a direct cancel is never gated on the target's own PropagateOnCancel); the descendant walk then depends on the target's PropagateOnCancel:

  • "cascade" (default): BFS through descendants, transitioning each non-terminal descendant to StatusCancelled — EXCEPT an isolate-marked descendant, whose whole subtree detaches from the cascade and keeps running (see cascadeCancelDescendantsLocked).
  • "isolate": only the target transitions; no descendant walk.

So an isolate-marked task survives an ANCESTOR's cascade but is always reachable by a direct Cancel (the operator's last word, and the spawning run's own CancelTask).

Already-terminal target → (false, nil). Missing target → ErrNotFound.

func (*Engine) CancelGroup 

func (e *Engine) CancelGroup(ctx context.Context, id tasks.TaskGroupID, reason string, propagate bool) error

CancelGroup implements tasks.TaskRegistry.

func (*Engine) Close 

func (e *Engine) Close(_ context.Context) error

Close implements tasks.TaskRegistry. Idempotent.

Drains every still-open `WatchGroup` subscriber and every `RegisterRetainTurnWaiter` channel before flipping `closed`. A caller blocked on either channel observes the close-on-shutdown instead of leaking forever — the public contract is "channel closes exactly once: either on resolve or on registry teardown." Identity is irrelevant here: we close everything we own.

func (*Engine) CreatePendingPatch 

func (e *Engine) CreatePendingPatch(ctx context.Context, sessionID identity.Identity, patchID string, bytesPayload []byte) (*tasks.Patch, error)

CreatePendingPatch is a driver-internal helper the conformance suite uses to seed a pending patch record. Planner code will land patches through a typed interface; Harbor ships the transition surface (ApplyPatch) and the helper that creates the pending record the planner would normally create.

func (*Engine) Get 

func (e *Engine) Get(ctx context.Context, id tasks.TaskID) (*tasks.Task, error)

Get implements tasks.TaskRegistry.

Identity-mandatory: the ctx Identity is read and verified against the stored task's Identity. Cross-tenant / cross-session reads are rejected with ErrNotFound (the task is invisible from outside its scope; we do NOT leak existence-without-access).

func (*Engine) IncrementToolCount 

func (e *Engine) IncrementToolCount(ctx context.Context, id tasks.TaskID) error

IncrementToolCount implements tasks.TaskRegistry.

Atomically increments `t.ToolCount` by 1 under the FSM lock and persists the updated record. The lock ensures N concurrent calls against the same task yield the correct final count — N — without torn writes.

Does NOT advance the FSM status and does NOT emit a bus event: per-tool-dispatch traffic is already covered by `task.spawned` / `tool.*` events; ToolCount is the cheap rollup the Console reads without subscribing to the per-tool stream.

func (*Engine) List 

func (e *Engine) List(ctx context.Context, sessionID identity.Identity, f tasks.TaskFilter) ([]tasks.TaskSummary, error)

List implements tasks.TaskRegistry. Filters by session and the optional fields on `f`. Returns task summaries in arbitrary order does not promise iteration order; may.

func (*Engine) ListGroups 

func (e *Engine) ListGroups(ctx context.Context, sessionID identity.Identity, status *tasks.TaskGroupStatus) ([]tasks.TaskGroup, error)

ListGroups implements tasks.TaskRegistry.

func (*Engine) ListTenant added in v1.11.0 

func (e *Engine) ListTenant(_ context.Context, tenantID string, f tasks.TaskFilter) ([]*tasks.Task, error)

ListTenant implements tasks.TaskRegistry — the admin-widened fleet read. It scans the engine's in-memory task map and returns FULL Task copies for every task whose tenant matches `tenantID`, across ALL (user, session) scopes, honouring the optional `f` facets.

The scan is COMPLETE without a StateStore scan: the engine's in-memory `tasks` map holds every live task in this process (both drivers hydrate their backend into it on open), so a tenant filter over the map yields the whole per-runtime fleet view. Cross-RUNTIME federation is the coordinator's job over per-runtime reads (the same division as sessions / events) — Harbor ships the per-runtime widened read.

Unlike List, ListTenant takes an EXPLICIT tenant argument rather than reading a mandatory triple from ctx: it is the admin-gated fleet read and the Protocol service is its only production caller (it gates on the verified `auth.ScopeAdmin` claim before invoking). An empty `tenantID` fails loud with tasks.ErrInvalidRequest — a fleet read never dumps the whole store.

func (*Engine) MarkComplete 

func (e *Engine) MarkComplete(ctx context.Context, id tasks.TaskID, result tasks.TaskResult) error

MarkComplete implements tasks.TaskRegistry. Persists `result` on the Task record (after caller-side redaction; the caller is responsible for redacting Value before passing it in, but we run it through the redactor again as a safety net).

func (*Engine) MarkFailed 

func (e *Engine) MarkFailed(ctx context.Context, id tasks.TaskID, taskErr tasks.TaskError) error

MarkFailed implements tasks.TaskRegistry. Persists `err` on the Task record (after caller-side redaction on the message).

func (*Engine) MarkPaused 

func (e *Engine) MarkPaused(ctx context.Context, id tasks.TaskID) error

MarkPaused implements tasks.TaskRegistry.

func (*Engine) MarkResumed 

func (e *Engine) MarkResumed(ctx context.Context, id tasks.TaskID) error

MarkResumed implements tasks.TaskRegistry. Distinct method (vs MarkRunning) so the bus event can be `task.resumed` rather than `task.started`. The FSM transition is identical to MarkRunning's Paused → Running edge.

func (*Engine) MarkRunning 

func (e *Engine) MarkRunning(ctx context.Context, id tasks.TaskID) error

MarkRunning implements tasks.TaskRegistry.

func (*Engine) OldestRetainedAt added in v1.14.0 

func (e *Engine) OldestRetainedAt(_ context.Context) (time.Time, bool, error)

OldestRetainedAt reports the OBSERVED oldest task-creation instant across the WHOLE retained task set — every (tenant, user, session) in this runtime — as the identity-free runtime-wide retention horizon a fleet-observe caller reads on `runtime.health`. It is the runtime-wide analogue of events.RetentionReporter, discovered by type assertion at the posture wiring seam (no `Supports*` capability protocol — CLAUDE.md §4.4). The returned instant carries no per-task content — only the bare oldest CreatedAt — so it is safe to expose runtime-wide, never a cross-tenant enumeration.

Returns (zero, false, nil) when the engine retains no tasks (present is false), and (zero, false, ErrRegistryClosed) after Close. The scan is COMPLETE without a StateStore scan: both drivers hydrate their backend into the in-memory `tasks` map on open, so the map holds every live task in this process. The read is an O(N) oldest-head fold under the RLock, never a wire-crossing enumeration.

func (*Engine) Prioritize 

func (e *Engine) Prioritize(ctx context.Context, id tasks.TaskID, priority int) (bool, error)

Prioritize implements tasks.TaskRegistry. Stores the new value; does NOT preempt or reorder execution (the "scheduling is runtime engine's concern" line). Emits task.prioritised.

func (*Engine) RecoverInterruptedTasks 

func (e *Engine) RecoverInterruptedTasks(ctx context.Context) (int, error)

RecoverInterruptedTasks is the open-time recovery sweep a durable driver runs immediately after New, before the registry is handed to any caller. It transitions every task left at StatusRunning by a crash to StatusFailed with RecoveryErrorCode, emitting one task.failed event per recovered task.

Scope. Only StatusRunning tasks are swept. StatusPaused is a durable wait state by design (HITL / tool-side OAuth survive a restart via the unified pause/resume primitive) and is left untouched. StatusPending tasks never began executing and remain discoverable for a future re-drive (auto-re-drive of recovered work is a runloop/steering concern and is out of scope here — the record is recovered, execution is not).

Idempotency. The sweep persists each recovered task as Failed before returning, so a second open of the same store observes it as Failed (no longer Running) and does not re-sweep or re-emit — exactly-once in the normal case, and never a double-write/double-emit. A bus failure mid-sweep leaves the task correctly persisted as Failed but drops that one recovery event (at-most-once), surfaced in the returned error rather than silently swallowed.

Group interaction. Recovery reuses the normal terminal-transition path, so a recovered (failed) group member drives the group resolve gate: a FailFast group cancels its remaining members and resolves, and a sealed group whose last member just recovered resolves to Completed. The returned count covers task recoveries; cascade cancellations triggered by fail-fast are reflected in the persisted group/member records.

func (*Engine) RegisterRetainTurnWaiter 

func (e *Engine) RegisterRetainTurnWaiter(sessionID identity.Identity) (<-chan tasks.TaskGroupID, func())

RegisterRetainTurnWaiter implements tasks.TaskRegistry. Returns a channel that the driver closes when the session's earliest-active retain-turn group resolves. Buffered size 1 + close-once invariant.

func (*Engine) ResolveOrCreateGroup 

func (e *Engine) ResolveOrCreateGroup(ctx context.Context, req tasks.GroupRequest) (*tasks.TaskGroup, error)

ResolveOrCreateGroup implements tasks.TaskRegistry. Idempotent on (SessionID, GroupID): if a group with the same ID already exists AND belongs to the ctx session, returns the existing record unchanged. Otherwise creates a fresh group.

func (*Engine) SealGroup 

func (e *Engine) SealGroup(ctx context.Context, id tasks.TaskGroupID) error

SealGroup implements tasks.TaskRegistry.

func (*Engine) Spawn 

func (e *Engine) Spawn(ctx context.Context, req tasks.SpawnRequest) (tasks.TaskHandle, error)

Spawn implements tasks.TaskRegistry.

func (*Engine) SpawnTool 

func (e *Engine) SpawnTool(ctx context.Context, req tasks.SpawnToolRequest) (tasks.TaskHandle, error)

SpawnTool implements tasks.TaskRegistry. It reshapes the tool request onto the Spawn path so the FSM, idempotency, and parent-graph bookkeeping share one code path; the task persists at StatusPending and emits task.spawned. Tool arguments are not carried on the Task (that field is reserved for the tool-catalog wiring); the description captures the intent for the lifecycle log. A run-loop driver advances the task (MarkRunning / MarkComplete / MarkFailed) once it dispatches the tool.

func (*Engine) WatchGroup 

func (e *Engine) WatchGroup(sessionID identity.Identity, groupID tasks.TaskGroupID) (<-chan tasks.GroupCompletion, func(), error)

WatchGroup implements tasks.TaskRegistry. Returns a buffered (size 1) channel that the driver primes with `GroupCompletion` either now (resolved-but-still-tracked group) or at resolve time.

type Snapshot 

type Snapshot struct {
	Tasks   []TaskRecord
	Groups  []*tasks.TaskGroup
	Patches []*tasks.Patch
}

Snapshot is the full set of persisted records a Backend returns from Hydrate. The engine rebuilds its task / group / patch maps and the derived indices (idempotency, children, member→group) from it.

type TaskRecord 

type TaskRecord struct {
	Task        *tasks.Task
	ContentHash [32]byte
}

TaskRecord is the durable unit for a task: the task itself plus the engine's idempotency content hash. The hash is the SHA-256 of the task's PRE-redaction content (see spawnRequestContentHash); it is carried out-of-band because it cannot be re-derived from the post-redaction Task fields a backend would otherwise persist.

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.