incident

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Sep 7, 2026 License: Apache-2.0 Imports: 32 Imported by: 0

Documentation

Overview

Package incident implements the Phase-0 incident substrate for agent-in-the-loop remediation: a deterministic failure classifier, a leader-gated dedupe subscriber, the incident store and status machine, a free-text log scrubber, and a durable-timer supervisor. Nothing in this package invokes an LLM or takes an autonomous action — it diagnoses failures and records incidents.

Index

Constants

View Source 
const (
	// Tier 1 — default autonomous.
	ActionTypeQuarantineReplay = "quarantine_replay"
	ActionTypeSnoozeRetry      = "snooze_retry"
	ActionTypeRetryFromFailure = "retry_from_failure"
	ActionTypeRetryCallbacks   = "retry_callbacks"
	ActionTypeNotify           = "notify"
	ActionTypeEscalate         = "escalate"

	// Tier 2 — autonomous only if explicitly allowed by the playbook.
	ActionTypeRerunWithParams          = "rerun_with_params"
	ActionTypePauseJob                 = "pause_job"
	ActionTypeUnpauseJob               = "unpause_job"
	ActionTypeClearCacheEntry          = "clear_cache_entry"
	ActionTypeSuppressDownstreamAlerts = "suppress_downstream_alerts"
	ActionTypeExtendSLAOnce            = "extend_sla_once"

	// Tier 3 — always approval-gated, never auto-executed. Execute() routes them
	// to an ApprovalRequest; ExecuteApproved() is the only path that dispatches
	// them, and only after a human decision (trust-the-substrate C4/C7).
	ActionTypeSkipTask           = "skip_task"
	ActionTypeOverrideSchemaGate = "override_schema_gate"
	ActionTypeApplyJobdefPatch   = "apply_jobdef_patch"
)

Typed action catalog (design-agent-in-the-loop.md). The agent never gets shell, SQL, or generic HTTP; it selects one of these typed actions and the executor validates + dispatches it server-side onto machinery that already exists.

View Source 
const (
	TierReadOnly   = 0
	TierAutonomous = 1
	TierGated      = 2
	TierApproval   = 3
)

Action tiers (design-agent-in-the-loop.md, "Action catalog: typed, server-enforced, tiered"). Tier semantics: tier 0/1 default autonomous, tier 2 autonomous only if explicitly allowed by the playbook, tier 3 always produces an ApprovalRequest and is never auto-executed in v1 regardless of config.

View Source 
const (
	// RuleAutoRetryBackoff re-runs a failed run after a fixed backoff — the
	// cheap path for transient-infra failures a plain retry clears.
	RuleAutoRetryBackoff = "auto_retry_backoff"
	// RuleSnoozeUntilCron defers a retry (e.g. wait for a late vendor file) via a
	// durable snooze timer rather than an in-process delay.
	RuleSnoozeUntilCron = "snooze_until_cron"
)

Deterministic Phase-0 rule names (design-agent-in-the-loop.md, "Playbook match"): if a failure class maps to a deterministic rule, the incident manager executes it directly and records it as an AgentAction with actor=policy — same audit trail, no container launch.

View Source 
const AgentActionTypeNote = "note"

AgentActionTypeNote is the AgentAction.Type used for free-text agent findings appended to the incident timeline. A note is a tier-0, actor=agent, status=executed row — it mutates nothing, it is evidence.

View Source 
const Redacted = "[REDACTED]"

Redacted is the placeholder substituted for a scrubbed secret value or token.

View Source 
const TimerKindSnoozeRetry = "snooze_retry"

TimerKindSnoozeRetry identifies snooze_retry durable timers.

Variables

View Source 
var (
	// ErrGlobalSessionCap is returned when the global concurrent-session cap is
	// already reached. The incident stays open and queues for later triage.
	ErrGlobalSessionCap = errors.New("incident: global agent-session cap reached")
	// ErrJobSessionCap is returned when the per-job concurrent-session cap is
	// already reached. A different-key failure on a job with an active session
	// still opens its own incident, which queues rather than being dropped.
	ErrJobSessionCap = errors.New("incident: per-job agent-session cap reached")
	// ErrNoProfile is returned when a session is requested without a profile.
	ErrNoProfile = errors.New("incident: agent session requires a profile")
)

Errors the dispatcher returns when a session cannot be launched under the configured caps. They are sentinel values so callers (and tests) can branch.

View Source 
var ErrActionNotApproved = errors.New("incident: action is not approved")

ErrActionNotApproved is returned by ExecuteApproved when the action is not in the `approved` state — it was never approved, was rejected, or has already run. It is the executor-side half of the once-only guarantee whose other half is the approvals service's conditional `decision = pending` update.

View Source 
var ErrActionNotPermitted = errors.New("incident: action not permitted by playbook")

ErrActionNotPermitted is returned when the effective playbook denies an action (not autonomously allowed and not routed to approval).

View Source 
var ErrAuthModeNone = errors.New("incident: apply_jobdef_patch refused: an auth mode is required (CAESIUM_AUTH_MODE=none)")

ErrAuthModeNone refuses apply_jobdef_patch under CAESIUM_AUTH_MODE=none (arc convention 4). Without an auth mode the approve route is an UNAUTHENTICATED POST that the agent container itself — which has network reach to the API — could call to approve its own jobdef rewrite. The master gate in Environment.Validate only refuses turning remediation ON without an auth mode; it does not cover a deployment that enabled remediation and later flipped auth off, which is exactly the window this check closes. Recorded as a failed AgentAction with this reason: not a panic, not a silent no-op.

View Source 
var ErrCrossBoundaryTarget = errors.New("incident: action target crosses the incident boundary")

ErrCrossBoundaryTarget is returned when an action names a run or job that does not belong to the incident's own job. The incident boundary is the security boundary: an action allowed for incident X must never reach across it to retry, pause, rerun, or clear the cache of an unrelated job/run — even if the playbook allows that action type. Recorded as a failed AgentAction.

View Source 
var ErrIncidentNotApprovable = errors.New("incident: cannot park incident awaiting approval")

N-3 (not built): a SECOND pending approval on one incident becomes unlistable once the first is decided — deciding moves the incident out of awaiting_approval, and the approvals feed lists what is parked there. Today nothing creates two (a proposal parks the incident and the session ends), but a future concurrent-session cap above 1 would. Filed as a follow-up.

ErrIncidentNotApprovable is returned when a tier-3 proposal is made against an incident that cannot be parked in awaiting_approval — it is terminal, or a human already advanced it past the point where an agent proposal is meaningful. The proposal is refused whole: no ApprovalRequest is created, because an approval whose incident is not parked is one no feed lists and no human can decide.

View Source 
var ErrInvalidTransition = errors.New("incident: invalid status transition")

ErrInvalidTransition is returned when a status transition is not permitted by the incident status machine.

View Source 
var ErrPatchAltersRemediation = errors.New("incident: apply_jobdef_patch may not change metadata.remediation: an agent may not edit the policy that governs it")

ErrPatchAltersRemediation refuses any jobdef patch that would change the job's own `metadata.remediation` block.

This is the security boundary that makes a job-level playbook trustworthy at all. Since the block is persisted and IS the input to the effective-playbook resolver, a patch that edits it is the agent rewriting the policy that governs the agent — the design's "the agent may not modify playbooks, profiles" rule, one indirection out. The attack is quiet: propose a byte-identical definition plus a permissive `metadata.remediation`, and a human approving what looks like a no-op hands the agent a wider allowlist for every later proposal.

The refusal is unconditional and covers BOTH provenance routes, so it holds whether the patch is applied directly or (for a git-synced job) rendered and escalated. A human editing the block through `caesium job apply` is unaffected; only the agent's own action surface is refused.

View Source 
var ErrPatchDefinitionRequired = errors.New("incident: apply_jobdef_patch requires a definition")

ErrPatchDefinitionRequired is returned when apply_jobdef_patch carries no proposed definition.

View Source 
var ErrRetryDeferred = errors.New("incident: retry deferred; retry once the condition clears")

ErrRetryDeferred marks a retry refused by a transient, retryable admission condition — the job is paused, or a concurrency slot is not yet free — that will clear on its own. ActionOps.RetryFromFailure implementations return it (wrapping the underlying cause) so a fired snooze_retry timer re-arms instead of being consumed and lost. A non-ErrRetryDeferred error is treated as a permanent failure.

View Source 
var ErrUnknownAction = errors.New("incident: unknown action type")

ErrUnknownAction is returned when the action type is not in the catalog.

Functions

func ActionTier 

func ActionTier(actionType string) (int, bool)

ActionTier returns the tier for an action type and whether it is in the catalog.

func CanTransition 

func CanTransition(from, to models.IncidentStatus) bool

CanTransition reports whether the status machine permits from → to.

func DedupeKey 

func DedupeKey(jobID uuid.UUID, taskName string, class FailureClass) string

DedupeKey is the stable correlation key for an incident: (job_id, task_name, failure_class). Failures sharing a key fold into one incident rather than opening twins.

func DefaultEngineFactory 

func DefaultEngineFactory(ctx context.Context, engineType models.AtomEngine) (atom.Engine, error)

DefaultEngineFactory selects the concrete engine for a profile, mirroring the worker's runtime executor.

func FreezeAllowlist 

func FreezeAllowlist(ctx context.Context, db *gorm.DB, jobID uuid.UUID, failingRunID *uuid.UUID) []string

FreezeAllowlist computes the static job allowlist that scopes an agent's read surface for one incident. It is the failing job itself PLUS every job transitively downstream of the failing job's output datasets in the lineage graph — but the seed datasets deliberately EXCLUDE any output rows produced by the failing run itself. That exclusion is the security property: a failing task can emit arbitrary `##caesium::output` markers, which materialize as lineage dataset rows; if those poisoned edges seeded the impact walk, an attacker who controls the failing task could widen the agent's read scope to jobs it should never see. Seeding only from the job's TRUSTED historical outputs (other runs) closes that hole.

The incident manager (an unscoped, server-side principal) calls this at incident open; the result is frozen onto the incident and every agent-session token minted for the incident carries a copy. The agent can never widen it.

FreezeAllowlist is best-effort with respect to lineage availability: if the lineage graph is empty or unavailable (e.g. OpenLineage disabled), it returns just the failing job's own alias rather than failing incident open.

func RecordNote 

func RecordNote(ctx context.Context, db *gorm.DB, incidentID uuid.UUID, sessionID *uuid.UUID, namespace *string, text string) (*models.AgentAction, error)

RecordNote appends a free-text finding to the incident timeline as an AgentAction row. sessionID is optional (nil for a note recorded outside a session). It returns the recorded row.

func ResolveProfile 

func ResolveProfile(ctx context.Context, db *gorm.DB, incidentID uuid.UUID, defaultProfile string) *models.AgentProfile

ResolveProfile returns the AgentProfile an incident's job names, falling back to the deployment default. It is what supplies the session IMAGE and limits — distinct from ResolvePlaybook, which supplies the enforced policy — and is best-effort: a nil profile means no session can be launched, not that the policy is unknown.

func SecretValuesFromEnv 

func SecretValuesFromEnv(raw, resolved map[string]string) []string

SecretValuesFromEnv extracts the resolved values of the env keys whose raw value was a secret:// reference. raw is the pre-resolution env (key -> raw value, some carrying the secret:// scheme); resolved is the post-resolution env (key -> actual value). Only keys that were secret refs contribute, so a literal env value that merely looks sensitive is not scrubbed here.

func SetSessionSupervisor 

func SetSessionSupervisor(s *Supervisor)

SetSessionSupervisor registers the process-wide session supervisor.

Types

type ActionOps 

type ActionOps interface {
	// RetryFromFailure re-runs a failed run through the admit-aware retry entry
	// point (run.Store.RetryFromFailureAdmitted — the B2 safety valves).
	RetryFromFailure(ctx context.Context, runID uuid.UUID) error
	// RetryCallbacks re-runs a run's failed callbacks (Dispatcher.RetryFailed).
	RetryCallbacks(ctx context.Context, runID uuid.UUID) error
	// RerunWithParams starts a new run with whitelisted param overrides, stamped
	// with the incident (new-run semantics: params feed cache identity).
	RerunWithParams(ctx context.Context, jobID uuid.UUID, params map[string]string) (uuid.UUID, error)
	// QuarantineReplay runs a side-effect-free what-if replay with --set params.
	QuarantineReplay(ctx context.Context, runID uuid.UUID, set map[string]string) (json.RawMessage, error)
	// Notify posts a structured update to a notification channel.
	Notify(ctx context.Context, channel, message string) error
	// Escalate pages a channel with an RCA summary, reporting whether the
	// escalation was actually ROUTED to a notification channel. An escalation
	// that reached nobody is still recorded (the event is persisted and
	// queryable), but it must never be reported as delivered — routed=false is
	// how the action row says "raised, but no policy carried it".
	Escalate(ctx context.Context, incidentID uuid.UUID, channel, summary string) (routed bool, err error)
	// SetJobPaused pauses/unpauses a job (Job.Paused).
	SetJobPaused(ctx context.Context, jobID uuid.UUID, paused bool) error
	// ClearCacheEntry deletes a task's cache entry.
	ClearCacheEntry(ctx context.Context, jobID uuid.UUID, taskName string) error
	// SuppressDownstreamAlerts suppresses downstream alerts until a deadline.
	SuppressDownstreamAlerts(ctx context.Context, incidentID uuid.UUID, until time.Time) error
	// ExtendSLAOnce writes a durable per-run SLA override.
	ExtendSLAOnce(ctx context.Context, runID uuid.UUID, extend time.Duration) error

	// SkipTask marks a task in a run skipped. The adapter goes through the
	// shipped run.Store.SkipTask, whose skipTaskAndDescendantsTx honours the
	// successors' trigger rules (design Open Question 3: skip interacts with
	// all_success/all_done and with cache identity — the result payload records
	// that caveat rather than hiding it).
	SkipTask(ctx context.Context, runID, taskID uuid.UUID, reason string) error
	// OverrideSchemaGateOnce records a ONE-RUN output-schema validation bypass on
	// the run row, which both ValidateTaskOutputSchema call sites read.
	OverrideSchemaGateOnce(ctx context.Context, runID uuid.UUID) error
	// ApplyJobdefPatch renders the proposed definition against the live job as a
	// diff and, unless dryRun, applies it through the shipped jobdefs importer
	// (the same in-process entry point POST /v1/jobdefs/apply uses). It returns
	// the rendered diff as JSON so the action row and the escalation carry the
	// exact change a human approved. Provenance routing is NOT its decision —
	// the executor derives the route and calls it with dryRun accordingly.
	ApplyJobdefPatch(ctx context.Context, jobID uuid.UUID, definition json.RawMessage, dryRun bool) (json.RawMessage, error)
}

ActionOps is the server-side operations surface the tier-1/2 catalog dispatches onto. It is an interface so the executor is unit-testable with a fake and so the incident package does not hard-depend on the run/callback/notification/ replay/cache subsystems — Stream C wires the concrete adapters. All methods map onto machinery that already exists (design action-catalog table).

type ActionParams 

type ActionParams struct {
	// RunID targets a specific run (retry, snooze, replay, extend_sla). Defaults
	// to the incident's remediation-target run when unset.
	RunID *uuid.UUID `json:"run_id,omitempty"`
	// JobID targets a specific job (pause/unpause, clear_cache, rerun). Defaults
	// to the incident's job when unset.
	JobID *uuid.UUID `json:"job_id,omitempty"`
	// TaskName targets a task (clear_cache_entry). Defaults to the incident task.
	TaskName string `json:"task_name,omitempty"`
	// Channel names a notification channel (notify, escalate).
	Channel string `json:"channel,omitempty"`
	// Message is a notify body.
	Message string `json:"message,omitempty"`
	// Summary is an escalation RCA summary.
	Summary string `json:"summary,omitempty"`
	// Overrides carries whitelisted param overrides (rerun_with_params) or replay
	// --set values (quarantine_replay).
	Overrides map[string]string `json:"overrides,omitempty"`
	// DelaySeconds defers a snooze_retry / bounds a suppress_downstream_alerts
	// window.
	DelaySeconds int64 `json:"delay_seconds,omitempty"`
	// ExtendSeconds extends a per-run SLA once (extend_sla_once).
	ExtendSeconds int64 `json:"extend_seconds,omitempty"`
	// Reason is the operator-facing justification recorded on a skip_task (it
	// becomes the skipped task row's error text, so the DAG explains itself).
	Reason string `json:"reason,omitempty"`
	// Definition carries the FULL desired job definition for apply_jobdef_patch,
	// in the same schema `caesium job apply` sends (pkg/jobdef.Definition). A
	// whole-document proposal rather than a field patch is deliberate: it is what
	// the shipped diff/apply path consumes, so the human approves exactly the
	// document that will be applied and the rendered diff is the real one.
	Definition json.RawMessage `json:"definition,omitempty"`
}

ActionParams is the union of typed parameters across the action catalog. Each handler reads and validates only the fields it needs.

type ActionRequest 

type ActionRequest struct {
	IncidentID uuid.UUID
	// SessionID links the action to an agent session (nil for actor=policy|human).
	SessionID *uuid.UUID
	// Actor originates the row (agent|human; deterministic rules use ExecutePolicy
	// which stamps policy).
	Actor models.AgentActionActor
	// Type is the catalog action type (e.g. "retry_from_failure").
	Type string
	// Params carries the typed action parameters.
	Params ActionParams
	// Playbook is the effective policy the action is validated against.
	Playbook Playbook
}

ActionRequest is one typed action to validate, record, and (when permitted) execute against an incident.

type AgentCredentialManager 

type AgentCredentialManager interface {
	MintAgentSessionKey(incidentID uuid.UUID, allowlist []string, ttl time.Duration) (*auth.CreateKeyResponse, error)
	RevokeKey(id uuid.UUID) error
}

AgentCredentialManager mints and revokes the scoped, short-lived credential a session runs with. auth.Service satisfies it. It is an interface so the supervisor can be unit-tested without the full auth service.

type ApprovalRedriver 

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

ApprovalRedriver re-dispatches approved-but-unexecuted tier-3 actions.

func NewApprovalRedriver 

func NewApprovalRedriver(db *gorm.DB, exec ApprovedExecutor, leaderCheck LeaderCheck, interval, grace time.Duration) *ApprovalRedriver

NewApprovalRedriver constructs the sweeper. A zero interval or grace takes the package default; a nil leaderCheck means "always act" (single-node, tests), matching TimerSupervisor.

func (*ApprovalRedriver) Run 

func (r *ApprovalRedriver) Run(ctx context.Context)

Run drives the sweep loop until ctx is cancelled.

It sweeps once BEFORE entering the ticker loop, unlike TimerSupervisor: the stranded-action case this exists for is created by a process death, so the rows most in need of redriving are the ones already on disk at startup.

func (*ApprovalRedriver) SweepOnce 

func (r *ApprovalRedriver) SweepOnce(ctx context.Context) error

SweepOnce dispatches every approved action that has been waiting longer than the grace period. Leader-gated so an N-node cluster redrives each action once.

Only `approved` rows are picked up. A row left `executing` by a process death is deliberately NOT redriven: the dispatch had already begun, and re-running a half-applied tier-3 mutation — a jobdef patch, a skipped task, a bypassed schema gate — unattended is worse than leaving a visible stuck row for a human.

type ApprovedExecutor 

type ApprovedExecutor interface {
	ExecuteApproved(ctx context.Context, actionID uuid.UUID) (*models.AgentAction, error)
}

ApprovedExecutor is the post-approval dispatch entry point the sweeper drives. *Executor satisfies it; the interface keeps the sweeper testable against a fake without constructing an executor's ActionOps and event sink.

type Bundle 

type Bundle struct {
	Incident       BundleIncident   `json:"incident"`
	Classification BundleClass      `json:"classification"`
	Failure        BundleFailure    `json:"failure"`
	Job            BundleJob        `json:"job"`
	RunHistory     []BundleRun      `json:"run_history"`
	LineageImpact  BundleImpact     `json:"lineage_impact"`
	Playbook       datatypes.JSON   `json:"playbook,omitempty"`
	Notes          []BundleNoteHint `json:"notes,omitempty"`
	GeneratedAt    time.Time        `json:"generated_at"`
}

Bundle is the JSON triage document an agent fetches once at session startup (GET /v1/agent/incidents/:id/bundle). Env injection cannot carry it — log tails alone exceed the ~128 KiB per-variable limit — so it is served over the scoped tool surface. It bundles everything the agent needs to plan within policy: the incident + classification, the failing task's error/scrubbed log/violations, the job + DAG, recent run history, the FROZEN lineage-impact allowlist, and the effective playbook. All attacker-influenced free text (the log tail) is scrubbed before it enters the bundle.

func BuildBundle 

func BuildBundle(ctx context.Context, db *gorm.DB, incidentID uuid.UUID, playbook *Playbook) (*Bundle, error)

BuildBundle assembles the triage bundle for an incident. playbook is the EFFECTIVE resolved policy (ResolvePlaybook), surfaced so the agent plans within the policy that will actually be enforced on its proposals.

It takes the resolved Playbook rather than an AgentProfile on purpose: the bundle used to surface the raw profile document while the executor enforced the job-scoped resolution, so the brief and the enforcement disagreed — an agent could plan correctly from its brief and still be denied, or believe it was constrained when it was not. The failing task's log tail is scrubbed by the A5 scrubber before it enters the bundle.

type BundleClass 

type BundleClass struct {
	Class    string         `json:"class"`
	Evidence datatypes.JSON `json:"evidence,omitempty"`
}

BundleClass carries the deterministic classification + its evidence.

type BundleDAGEdge 

type BundleDAGEdge struct {
	From string `json:"from"`
	To   string `json:"to"`
}

BundleDAGEdge is one from→to dependency edge (by task name).

type BundleFailedPartition 

type BundleFailedPartition struct {
	Partition string    `json:"partition"`
	Index     int       `json:"partition_index"`
	TaskRunID uuid.UUID `json:"task_run_id"`
	Attempt   int       `json:"attempt"`
	Error     string    `json:"error,omitempty"`
	ExitCode  *int      `json:"exit_code,omitempty"`
	Result    string    `json:"result,omitempty"`
}

BundleFailedPartition is one failed fan-out instance. It carries the error and exit code but NOT a log tail: N scrubbed log tails would blow the bundle far past the size an agent session can hold, and the primary LogTail above already carries the representative failure. The task_run_id is the handle for fetching any specific instance's log from the logs endpoint.

type BundleFailure 

type BundleFailure struct {
	Error            string         `json:"error,omitempty"`
	LogTail          string         `json:"log_tail,omitempty"`
	LogTailScrubbed  bool           `json:"log_tail_scrubbed"`
	SchemaViolations datatypes.JSON `json:"schema_violations,omitempty"`
	ExitCode         *int           `json:"exit_code,omitempty"`
	Image            string         `json:"image,omitempty"`
	Result           string         `json:"result,omitempty"`
	// Partition is the fan-out instance the fields above describe. Empty for an
	// unfanned task.
	Partition string `json:"partition,omitempty"`
	// PartitionCount is the size of the fan-out group, 0 when unfanned.
	PartitionCount int `json:"partition_count,omitempty"`
	// Partitions lists every FAILED instance of the group (capped at
	// bundleFailedPartitionCap). Nil for an unfanned task.
	Partitions []BundleFailedPartition `json:"failed_partitions,omitempty"`
	// PartitionsTruncated reports that Partitions was capped.
	PartitionsTruncated bool `json:"failed_partitions_truncated,omitempty"`
}

BundleFailure carries the failing task's diagnostic signals. LogTail is scrubbed; SchemaViolations and ExitCode come straight from the TaskRun.

For a FANNED step the primary fields describe the first failed instance (the one the incident was classified from) and Partitions enumerates every failed sibling — an agent triaging "the extract step failed" must be able to see that 3 of 12 partitions failed and which ones, not one arbitrary row.

type BundleImpact 

type BundleImpact struct {
	AllowedJobs []string `json:"allowed_jobs"`
	Frozen      bool     `json:"frozen"`
}

BundleImpact is the FROZEN lineage-impact snapshot: the static job allowlist the incident manager computed at open (excluding the failing run's own outputs). The agent reads this instead of the live /v1/lineage/impact route, from which its scoped token is 403'd.

type BundleIncident 

type BundleIncident struct {
	ID              uuid.UUID  `json:"id"`
	JobID           uuid.UUID  `json:"job_id"`
	RunID           *uuid.UUID `json:"run_id,omitempty"`
	TaskName        string     `json:"task_name,omitempty"`
	Status          string     `json:"status"`
	OccurrenceCount int        `json:"occurrence_count"`
	Attempt         int        `json:"attempt"`
	OpenedAt        time.Time  `json:"opened_at"`
}

BundleIncident is the incident header the agent triages.

type BundleJob 

type BundleJob struct {
	Alias            string          `json:"alias"`
	Paused           bool            `json:"paused"`
	SchemaValidation string          `json:"schema_validation,omitempty"`
	Tasks            []BundleTask    `json:"tasks"`
	Edges            []BundleDAGEdge `json:"edges"`
}

BundleJob is the job definition + DAG topology.

type BundleNoteHint 

type BundleNoteHint struct {
	CreatedAt time.Time `json:"created_at"`
	Text      string    `json:"text"`
}

BundleNoteHint surfaces prior timeline notes so a resumed session has context.

type BundleRun 

type BundleRun struct {
	ID          uuid.UUID  `json:"id"`
	Status      string     `json:"status"`
	Error       string     `json:"error,omitempty"`
	StartedAt   time.Time  `json:"started_at"`
	CompletedAt *time.Time `json:"completed_at,omitempty"`
	DurationMS  int64      `json:"duration_ms,omitempty"`
}

BundleRun is one recent run with its duration for history-based reasoning (e.g. "this vendor has been late 4 of the last 30 days").

type BundleTask 

type BundleTask struct {
	Name        string `json:"name"`
	TriggerRule string `json:"trigger_rule,omitempty"`
	Retries     int    `json:"retries"`
	ReplaySafe  bool   `json:"replay_safe"`
}

BundleTask is one step in the DAG.

type CapExceeded 

type CapExceeded int

CapExceeded distinguishes which concurrent-session cap a reservation failed. 

const (
	// CapNone means the reservation succeeded (no cap exceeded).
	CapNone CapExceeded = iota
	// CapGlobal means the global concurrent-session cap was already reached.
	CapGlobal
	// CapPerJob means the per-job concurrent-session cap was already reached.
	CapPerJob
)

type Classifier 

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

Classifier maps a Signal to a FailureClass using a fixed precedence of structured signals (event type, schema violations, engine result) followed by a configurable exit-code table and log-tail regex table. It holds no state beyond its rule tables and is safe for concurrent use.

func NewClassifier 

func NewClassifier() *Classifier

NewClassifier returns a Classifier seeded with sane default rules.

func (*Classifier) Classify 

func (c *Classifier) Classify(sig Signal) FailureClass

Classify maps a Signal to a FailureClass. Precedence, top to bottom:

  1. run_timed_out / sla_missed → sla_risk
  2. schema_violation event / violations → schema_violation
  3. StartupFailure / ResourceFailure → transient_infra
  4. log-tail regex table → data_unavailable|auth_failure|oom|quota
  5. exit-code table → (default 137 → oom)
  6. fallback → unknown

func (*Classifier) WithExitCodeRule 

func (c *Classifier) WithExitCodeRule(code int, class FailureClass) *Classifier

WithExitCodeRule overrides or adds an exit-code rule (configuration hook).

func (*Classifier) WithLogRule 

func (c *Classifier) WithLogRule(pattern string, class FailureClass) (*Classifier, error)

WithLogRule appends a log-tail rule (configuration hook). Appended rules are evaluated after the defaults.

type DeterministicRule 

type DeterministicRule struct {
	// Name is the operator-facing rule name recorded on the incident timeline.
	Name string
	// Class is the failure class this rule fires for.
	Class FailureClass
	// ActionType is the concrete catalog action the rule performs.
	ActionType string
	// Delay is the backoff before a retry (auto_retry_backoff) or the snooze
	// window (snooze_until_cron). Zero means immediate.
	Delay time.Duration
}

DeterministicRule maps a failure class to a server-side remediation that runs without an agent container. Delay is the backoff/snooze applied before the concrete action.

func DefaultRules 

func DefaultRules() []DeterministicRule

DefaultRules ships the two Phase-0 deterministic rules the design names.

type EngineFactory 

type EngineFactory func(ctx context.Context, engineType models.AtomEngine) (atom.Engine, error)

EngineFactory resolves an atom.Engine for an engine type. Injectable so tests can supply a fake engine without a container runtime.

type Executor 

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

Executor is the typed, server-enforced action layer. The agent never gets shell, SQL, or generic HTTP: every mutation arrives as a typed action, is validated against the effective playbook, executed server-side through the injected ActionOps, and recorded as an AgentAction audit row with the right actor/tier/status. Deterministic Phase-0 rules run through the same recording path with actor=policy and no container launch.

func NewExecutor 

func NewExecutor(store *Store, ops ActionOps) *Executor

NewExecutor constructs an executor over the incident store and the action operations surface (implemented by Stream C's concrete adapters; a fake in tests). A nil ops is tolerated for actions that never dispatch through it (deny/approve paths), but executing a dispatching action then panics — callers must supply ops in any path that executes.

func (*Executor) ApplyDeterministicRule 

func (e *Executor) ApplyDeterministicRule(ctx context.Context, inc *models.Incident, rules *Rules) (*models.AgentAction, bool, error)

ApplyDeterministicRule runs the deterministic rule for the incident's class (if any) as an actor=policy action, returning the recorded action and true. It returns (nil, false, nil) when no rule matches the class — the caller then dispatches an agent session instead. The rule's concrete action is executed through the same recording/metric/audit path as an agent action, so the timeline reconstructs uniformly.

func (*Executor) Execute 

func (e *Executor) Execute(ctx context.Context, req ActionRequest) (*models.AgentAction, error)

Execute validates a typed action against the effective playbook, records it as an AgentAction row, and — when the playbook permits autonomous execution — dispatches it server-side. Tier-3 actions and playbook-gated actions are recorded as proposed (awaiting approval) without executing; playbook-denied actions are recorded as rejected and return ErrActionNotPermitted.

func (*Executor) ExecuteApproved 

func (e *Executor) ExecuteApproved(ctx context.Context, actionID uuid.UUID) (*models.AgentAction, error)

ExecuteApproved dispatches a tier-3 action a human approved.

It is invoked by the incident REST service AFTER the approval decision transaction commits (api/rest/service/incident/approvals.go), which is what makes the audit spine honest: the row moves proposed → approved → executed, each step durable before the next begins, and a crash between them leaves an approved-but-unexecuted action a human can see rather than a silent nothing.

Guards:

  • the action is CLAIMED with a conditional approved → executing UPDATE, so dispatch happens at most once no matter how many callers arrive. A rejected action, an already-executed one, a raw proposal, and a loser of the claim race are all refused with ErrActionNotApproved, and the loser does nothing rather than re-running the mutation;
  • the incident boundary is re-verified inside dispatch, so an approval cannot be used to reach a run or job outside the incident's own;
  • the decider is read from the ApprovalRequest, never from the caller, so the audit actor is the recorded human decision.

The claim is what makes this method safe to call from BOTH the synchronous post-decision path and ApprovalRedriver's sweep — which is what gives an approved action a way to recover from a process death between the decision commit and the dispatch. Without it, execution was fire-once-and-hope: a crash stranded the action `approved` forever, and re-approval is refused by the approvals service's pending-only guard.

func (*Executor) ExecutePolicy 

func (e *Executor) ExecutePolicy(ctx context.Context, incidentID uuid.UUID, actionType string, params ActionParams) (*models.AgentAction, error)

ExecutePolicy runs a deterministic server-side action as actor=policy: no playbook allowlist gate (a deterministic rule is pre-approved by being deterministic) and no agent session, but the same audit recording, metric, and dispatch path. Used by the Phase-0 deterministic rules (rules.go).

func (*Executor) RegisterTimerHandlers 

func (e *Executor) RegisterTimerHandlers(sup *TimerSupervisor)

RegisterTimerHandlers wires the durable-timer handlers this executor owns onto a TimerSupervisor. snooze_retry timers fire an admit-aware retry when due.

func (*Executor) SetEventSink 

func (e *Executor) SetEventSink(bus event.Bus, store *event.Store)

SetEventSink wires the process event bus and the durable event store onto the executor. Wired once at startup (cmd/start/start.go) behind the remediation master gate; nil-safe, so a test executor simply emits nothing.

type FailureClass 

type FailureClass string

FailureClass buckets a failure for remediation routing. The classifier is purely deterministic — the same Signal always yields the same class. 

const (
	// ClassTransientInfra covers startup/resource engine failures that a plain
	// retry often clears.
	ClassTransientInfra FailureClass = "transient_infra"
	// ClassSchemaViolation covers a task whose output violated its declared
	// schema (in warn mode the task did not fail).
	ClassSchemaViolation FailureClass = "schema_violation"
	// ClassSLARisk covers run timeouts and SLA misses.
	ClassSLARisk FailureClass = "sla_risk"
	// ClassDataUnavailable covers missing/unreachable inputs.
	ClassDataUnavailable FailureClass = "data_unavailable"
	// ClassAuthFailure covers credential/permission failures.
	ClassAuthFailure FailureClass = "auth_failure"
	// ClassOOM covers out-of-memory kills. Best-effort until the OOM-flag
	// detection from design-resource-right-sizing lands.
	ClassOOM FailureClass = "oom"
	// ClassQuota covers rate limits / quota exhaustion.
	ClassQuota FailureClass = "quota"
	// ClassUnknown is the fallback — always agent-eligible.
	ClassUnknown FailureClass = "unknown"
)

type LeaderCheck 

type LeaderCheck func(context.Context) (bool, error)

LeaderCheck reports whether this node currently hosts the cluster leader. The incident subscriber is leader-gated (mirroring the run-queue dequeuer, NOT the per-node notification subscriber) so an N-node cluster opens exactly one incident per failure. A nil LeaderCheck means "always act" (single-node).

type OpenOutcome 

type OpenOutcome string

OpenOutcome describes what OpenOrAppend did. 

const (
	// OutcomeOpened: a new incident row was created.
	OutcomeOpened OpenOutcome = "opened"
	// OutcomeAppended: an existing open incident absorbed this as an occurrence.
	OutcomeAppended OpenOutcome = "appended"
	// OutcomeSuppressed: skipped because a same-key incident closed within the
	// cooldown window.
	OutcomeSuppressed OpenOutcome = "suppressed"
)

type OpenParams 

type OpenParams struct {
	Namespace  *string
	JobID      uuid.UUID
	RunID      *uuid.UUID
	TaskID     *uuid.UUID
	TaskName   string
	Class      FailureClass
	LastError  string
	Evidence   datatypes.JSON
	BackfillID *uuid.UUID
	// RemediationTargetRunID is the run whose later success would close this
	// incident as remediated.
	RemediationTargetRunID *uuid.UUID
	// Cooldown suppresses re-opening within this window after the last incident
	// for the same key closed. Zero disables cooldown suppression.
	Cooldown time.Duration
}

OpenParams describes a failure the subscriber wants to record as an incident.

type Playbook 

type Playbook struct {
	// Allow is the configured set of action types the agent may take
	// autonomously; nil means unconfigured (see above), not empty.
	Allow map[string]bool
	// RequireApproval forces listed action types through the approval gate even
	// if their tier would otherwise be autonomous.
	RequireApproval map[string]bool
	// ParamOverrides whitelists rerun_with_params keys → allowed values; nil
	// means unconfigured, and an unconfigured whitelist denies every key.
	ParamOverrides map[string][]string
}

Playbook is the effective, resolved remediation policy the executor enforces for one incident. It is produced by resolving `metadata.remediation` over the AgentProfile defaults (Playbook.Override); here it is purely the enforcement input.

NIL AND EMPTY MEAN DIFFERENT THINGS, and the distinction is the whole policy model — read it before touching decide():

  • Allow == nil → NOT CONFIGURED. Tier defaults apply: tier 0/1 is autonomous, tier 2 needs an explicit allow (so it is denied), tier 3 always goes to a human.
  • Allow != nil → CONFIGURED ALLOWLIST, and it governs at EVERY tier below 3. An action runs autonomously if and only if it is listed. An empty non-nil map therefore allows NOTHING — which is what `allow: []` in a playbook document plainly says, and what the shipped `triage-only` profile relies on.

Conflating the two is how a "zero risk" profile ends up granting every tier-1 action, and how combining two playbooks can silently widen tier 2.

func DecodePlaybook 

func DecodePlaybook(raw []byte) Playbook

DecodePlaybook parses a stored playbook document into the enforcement input. An empty or malformed document yields the zero Playbook (unconfigured; tier 3 still → approval), never a permissive one.

A PRESENT BUT EMPTY `allow: []` decodes to a configured, empty allowlist — "allow nothing" — not to nil. That is the difference between a profile that declined to configure autonomy and one that deliberately grants none; the shipped `triage-only` profile depends on it.

func DenyAllPlaybook 

func DenyAllPlaybook() Playbook

DenyAllPlaybook is the fail-closed policy: no action type is autonomously permitted at any tier, so every proposal is either denied or routed to a human. It is what a caller uses when a job's DECLARED policy cannot be resolved — substituting any other policy there would enforce something the job did not ask for, in the widening direction.

It is a CONFIGURED, empty allowlist, which is why it is not the zero Playbook: the zero value means unconfigured and still lets tier 0/1 run autonomously.

func ResolvePlaybook 

func ResolvePlaybook(ctx context.Context, db *gorm.DB, incidentID uuid.UUID, defaultProfile string) Playbook

ResolvePlaybook returns the effective policy for an incident.

Resolution, in order:

  1. incident → job. The job's persisted `metadata.remediation` block (models.Job.Remediation) is the job's policy.
  2. That block names the AgentProfile whose playbook is the base; a job that declares a block but no profile uses the deployment default. The block's `autonomy` sub-block then resolves over that base (Playbook.Override).
  3. A job with no remediation block at all falls back to the deployment default profile: that is what "default profile" means, and the job has expressed no policy to override it.

Every failure fails CLOSED, with the severity matched to what was lost:

  • the incident or job cannot be read → the zero Playbook (unconfigured: tier 3 to a human, tier 2 denied, tier 0/1 autonomous);
  • the job DECLARED a policy whose profile cannot be loaded → DenyAllPlaybook, which permits no autonomous action at all. Falling back to the deployment default there would substitute a policy the job explicitly replaced, in the widening direction.

func (Playbook) Document 

func (pb Playbook) Document() json.RawMessage

Document re-encodes a resolved Playbook into the stored document shape, so the triage bundle can show the agent EXACTLY the policy that will be enforced on its proposals rather than the raw profile document it was derived from.

A nil Allow (unconfigured) is omitted; a configured-but-empty one is rendered as `[]`, because "grants nothing" and "not configured" are different policies and the agent must be able to tell them apart.

func (Playbook) Override 

func (pb Playbook) Override(job Playbook) Playbook

Override resolves a job's authored `metadata.remediation.autonomy` block over the AgentProfile playbook it names, returning the effective policy. The receiver is the PROFILE (the default); the argument is the JOB block.

The job block is an AUTHORED POLICY, so it may GRANT as well as narrow — the design's `metadata.remediation` overrides profile defaults, and its canonical example declares a wider allowlist than the profile it names. That is safe because the block is human-authored and an agent may not edit it: ErrPatchAltersRemediation refuses any `apply_jobdef_patch` that would change it. Without that refusal this method would be a privilege-escalation path, which is exactly why the two ship together.

Per field, honouring the nil-vs-empty rule on Playbook:

  • Allow: a configured job list REPLACES the profile's (grant or narrow, as authored); a nil job list inherits the profile's. `allow: []` is a configured list that grants nothing.
  • ParamOverrides: same replace-or-inherit rule.
  • RequireApproval: a UNION, and the one field where the stricter side always wins. Removing an approval gate is the single edit that can only reduce safety, and nothing in the design asks a job to do it, so a profile's gate survives a job block that omits it.

type Rules 

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

Rules is a deterministic rule table keyed by failure class. It holds no state beyond its table and is safe for concurrent reads.

func DefaultRuleSet 

func DefaultRuleSet() *Rules

DefaultRuleSet returns the shipped default deterministic rule table.

func NewRules 

func NewRules(rules ...DeterministicRule) *Rules

NewRules builds a rule table from the supplied rules (last rule for a class wins).

func (*Rules) Match 

func (r *Rules) Match(class FailureClass) (DeterministicRule, bool)

Match returns the deterministic rule for a class, if one exists.

type Scrubber 

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

Scrubber removes secret material from free-text log output before it enters a triage bundle, an agent-readable endpoint, or an escalation message. It does exact removal of every resolved secret:// value (subject to the over-redaction guard) plus a conservative high-entropy token heuristic.

func NewScrubber 

func NewScrubber(secretValues []string) *Scrubber

NewScrubber builds a Scrubber from a set of resolved secret values. Values that fail the over-redaction guard (too short, a denylisted literal, or a bare small number) are dropped from exact-match scrubbing.

func (*Scrubber) Scrub 

func (s *Scrubber) Scrub(text string) string

Scrub returns text with every guard-passing secret value removed by exact replacement and every high-entropy token redacted. The exact-match pass runs first so a known secret is always removed even if the entropy heuristic would have missed it.

type Signal 

type Signal struct {
	// EventType is the bus event.Type string that triggered classification.
	EventType string
	// Result is the atom.Result string persisted on TaskRun.Result.
	Result string
	// HasSchemaViolations reports whether the task run recorded schema violations.
	HasSchemaViolations bool
	// ExitCode is the raw process exit code, or nil when none was captured.
	ExitCode *int
	// LogTail is the (already-scrubbed) tail of the task log.
	LogTail string
	// Error is the TaskRun.Error text.
	Error string
}

Signal is the deterministic input the classifier consumes. It is derived from a bus event plus the persisted TaskRun (Result/ExitCode/SchemaViolations/ LogText/Error), never from an LLM.

type Store 

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

Store persists incidents. It is the single writer the leader-gated subscriber uses to open, correlate, and advance incidents.

func NewStore 

func NewStore(db *gorm.DB) *Store

NewStore returns an incident store over db.

func (*Store) AllowedJobsForIncident 

func (s *Store) AllowedJobsForIncident(ctx context.Context, incidentID uuid.UUID) ([]string, error)

AllowedJobsForIncident returns the frozen agent read-scope allowlist for an incident (the job aliases the incident manager snapshotted at open). The incident's own job alias is guaranteed present when it was resolvable at open.

func (*Store) CancelTimersForIncident 

func (s *Store) CancelTimersForIncident(ctx context.Context, incidentID uuid.UUID) (int64, error)

CancelTimersForIncident cancels all pending timers for an incident (used on human take-over). Terminal transitions already cancel via Transition.

func (*Store) ClaimTimer 

func (s *Store) ClaimTimer(ctx context.Context, id uuid.UUID) (bool, error)

ClaimTimer atomically flips a pending timer to fired, returning true if this caller won the claim. The conditional update (status = pending) makes the claim safe even if two sweeps race.

func (*Store) CountActiveAgentSessions 

func (s *Store) CountActiveAgentSessions(ctx context.Context) (int64, error)

CountActiveAgentSessions returns the number of non-terminal agent sessions (pending or running) across all incidents. The leader-gated dispatcher uses this to enforce the global concurrent-session cap.

func (*Store) CountActiveAgentSessionsForJob 

func (s *Store) CountActiveAgentSessionsForJob(ctx context.Context, jobID uuid.UUID) (int64, error)

CountActiveAgentSessionsForJob returns the number of non-terminal agent sessions bound to incidents of the given job. The dispatcher uses this to enforce the per-job concurrent-session cap (default 1).

func (*Store) DB 

func (s *Store) DB() *gorm.DB

DB exposes the underlying handle for read APIs built on top of the store.

func (*Store) DueTimers 

func (s *Store) DueTimers(ctx context.Context, now time.Time, limit int) ([]models.RemediationTimer, error)

DueTimers returns pending timers whose fire time has passed.

func (*Store) Get 

func (s *Store) Get(ctx context.Context, id uuid.UUID) (*models.Incident, error)

Get returns a single incident by id.

func (*Store) OpenForJobTask 

func (s *Store) OpenForJobTask(ctx context.Context, jobID uuid.UUID, taskName string) ([]models.Incident, error)

OpenForJobTask returns the open (non-terminal) incidents for a job whose task name matches. Used by the terminal-verification success path.

func (*Store) OpenOrAppend 

func (s *Store) OpenOrAppend(ctx context.Context, p OpenParams) (*models.Incident, OpenOutcome, error)

OpenOrAppend opens exactly one incident per dedupe key or folds the failure into the existing open incident as an occurrence. The open is an ATOMIC conditional insert on active_dedupe_key (unique index, ON CONFLICT DO NOTHING), so failover races and duplicate per-node sla_missed events cannot open twins. A recently-closed same-key incident within Cooldown suppresses a fresh open.

func (*Store) Remediate 

func (s *Store) Remediate(ctx context.Context, id uuid.UUID, summary string) (*models.Incident, error)

Remediate marks an incident remediated then closed in one step. It is the terminal-verified success path: the caller invokes it only when a subsequent run for the incident's job/task actually succeeded.

func (*Store) ReserveAgentSession 

func (s *Store) ReserveAgentSession(ctx context.Context, session *models.AgentSession, jobID uuid.UUID, globalCap, perJobCap int) (CapExceeded, error)

ReserveAgentSession atomically reserves a slot for a new agent session AT THE DATABASE, so the concurrent-session caps hold across processes and nodes — an in-process mutex cannot serialize two supervisors on different nodes (or a leader-failover split-brain window). It conditionally inserts the pending session row in a SINGLE statement, guarded by both the global and per-job active-session counts; the count and the insert therefore evaluate atomically under SQLite/dqlite's serialized-writer semantics, so two concurrent reservations cannot both observe "under cap" and both insert.

The active-session predicate (state IN pending|running) matches CountActiveAgentSessions / CountActiveAgentSessionsForJob EXACTLY, so the reservation and the counters always agree on what "active" means.

It returns which cap (if any) blocked the reservation. On CapNone the pending row identified by session.ID exists and counts as active.

func (*Store) ScheduleTimer 

func (s *Store) ScheduleTimer(ctx context.Context, incidentID uuid.UUID, kind string, fireAt time.Time, payload datatypes.JSON, actionID *uuid.UUID, namespace *string) (*models.RemediationTimer, error)

ScheduleTimer persists a durable timer owned by an incident. The timer survives restart/failover so a pending snooze/retry is not lost.

func (*Store) Transition 

func (s *Store) Transition(ctx context.Context, id uuid.UUID, to models.IncidentStatus, summary string) (*models.Incident, error)

Transition advances an incident to a new status, enforcing the status machine. On any terminal transition it clears active_dedupe_key (so a future same-key failure may open a fresh incident) and stamps closed_at. remediated/escalated set closed_at only when they are the final state reached (they still permit a later → closed transition, which is a no-op on closed_at).

type Subscriber 

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

Subscriber is the leader-gated incident manager. It consumes failure events, classifies each, and opens/correlates an incident; it consumes success events to close incidents as remediated when a later run succeeds. When a remediator is wired (SetRemediator, behind the master gate) it also runs the Phase-0 deterministic rules on incident open; it never invokes an LLM.

func NewSubscriber 

func NewSubscriber(bus event.Bus, db *gorm.DB, leaderCheck LeaderCheck, cooldown time.Duration) *Subscriber

NewSubscriber constructs an incident subscriber.

func (*Subscriber) SetRemediator 

func (s *Subscriber) SetRemediator(executor *Executor, rules *Rules)

SetRemediator wires the deterministic-rule executor and rule table so that opening an incident whose class maps to a deterministic rule (auto_retry_backoff / snooze_until_cron) runs that rule as an actor=policy action — the live Phase-0 autonomous path. Only invoked behind the master gate (CAESIUM_AGENT_REMEDIATION_ENABLED) from cmd/start. A subscriber without a remediator opens and classifies incidents but takes no autonomous action.

func (*Subscriber) Start 

func (s *Subscriber) Start(ctx context.Context) error

Start subscribes to the failure and success event types and processes them until ctx is cancelled.

func (*Subscriber) StartWithReady 

func (s *Subscriber) StartWithReady(ctx context.Context, ready chan<- struct{}) error

StartWithReady subscribes and signals readiness once the subscription is live (used by tests to avoid a publish race).

type Supervisor 

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

Supervisor drives a single agent container through the existing atom.Engine (create → wait → logs → stop) with wall-clock enforcement and persisted session logs, materializing an AgentSession record — deliberately NOT a JobRun/TaskRun (a session as a run would pollute the quarantine-filtered run stats and feed its own exhaust into the incident bus). It runs on the leader node in v1 and enforces the concurrent-session caps against the shared store, not per-process, so an N-node cluster does not multiply them.

func NewSupervisor 

func NewSupervisor(db *gorm.DB, creds AgentCredentialManager, factory EngineFactory, cfg SupervisorConfig) *Supervisor

NewSupervisor constructs a session supervisor.

func SessionSupervisor 

func SessionSupervisor() *Supervisor

SessionSupervisor returns the registered session supervisor, or nil.

func (*Supervisor) Dispatch 

func (s *Supervisor) Dispatch(ctx context.Context, inc *models.Incident, profile *models.AgentProfile) (*models.AgentSession, error)

Dispatch enforces the concurrent-session caps and launches a session for the incident. The cap-check and the slot reservation (creating the pending session row) are performed atomically under dispatchMu, so concurrent dispatches cannot both pass the cap and overshoot MaxConcurrentSessions. It returns ErrGlobalSessionCap / ErrJobSessionCap without launching when a cap is already reached — the caller leaves the incident open to queue for later triage rather than dropping it.

func (*Supervisor) EndSession 

func (s *Supervisor) EndSession(ctx context.Context, sessionID uuid.UUID) error

EndSession terminates a still-active agent session and revokes its scoped credential. It is the "no idle container burning tokens while a human decides" half of the tier-3 approval flow (design-agent-in-the-loop.md, Approval gates): once a proposal parks the incident in awaiting_approval, the session that made it has nothing left to do.

The state write is CONDITIONAL on the session still being pending/running, so it can never rewrite a session the supervisor's own execute() already finalized — the two race by construction (the agent is mid-HTTP-call to the API when this runs). Token revocation is unconditional and idempotent: a credential that outlives its session is the thing worth being paranoid about.

Winning that conditional write also makes this call responsible for the CONTAINER. Marking the row succeeded and revoking the credential without stopping the container leaves the agent running — burning model tokens against a revoked key for the rest of the session timeout, which is precisely the cost this method exists to avoid. The container is stopped through the same atom.Engine handle execute() uses, resolved from the row's own engine and container id.

func (*Supervisor) Run 

func (s *Supervisor) Run(ctx context.Context, inc *models.Incident, profile *models.AgentProfile) (*models.AgentSession, error)

Run launches a session WITHOUT cap enforcement (used by callers that gate concurrency elsewhere, and by tests). It reserves a slot (mint + create the pending session row) then drives the container to a terminal state.

type SupervisorConfig 

type SupervisorConfig struct {
	// APIBaseURL is the Caesium API base URL injected into the agent container so
	// it can reach its scoped tool surface. Falls back to a sane localhost value.
	APIBaseURL string
	// SessionTimeout is the wall-clock budget a session may run before being
	// forcibly stopped and marked timed_out.
	SessionTimeout time.Duration
	// MaxConcurrentSessions caps globally-active sessions (<=0 means 1).
	MaxConcurrentSessions int
	// PerJobConcurrentSessions caps active sessions per job (<=0 means 1).
	PerJobConcurrentSessions int
}

SupervisorConfig carries the supervisor's operational limits (from env).

type TimerFunc 

type TimerFunc func(ctx context.Context, timer models.RemediationTimer) error

TimerFunc handles a fired durable timer. Stream B registers the concrete snooze_retry handler; in Phase 0 with no handler registered the sweeper simply records the timer as fired.

type TimerSupervisor 

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

TimerSupervisor is the leader-gated durable-timer sweeper. It periodically fires due RemediationTimer rows so a pending snooze/retry survives restart/failover (no in-process time.NewTimer is the sole record). Timers whose owning incident has reached a terminal state are never fired — they are cancelled at the transition, and the sweeper double-checks before firing.

func NewTimerSupervisor 

func NewTimerSupervisor(db *gorm.DB, leaderCheck LeaderCheck, interval time.Duration) *TimerSupervisor

NewTimerSupervisor constructs the sweeper. A zero interval defaults to 5s.

func (*TimerSupervisor) RegisterHandler 

func (t *TimerSupervisor) RegisterHandler(kind string, fn TimerFunc)

RegisterHandler registers the handler invoked when a timer of the given kind fires.

func (*TimerSupervisor) Run 

func (t *TimerSupervisor) Run(ctx context.Context)

Run drives the sweep loop until ctx is cancelled.

func (*TimerSupervisor) SweepOnce 

func (t *TimerSupervisor) SweepOnce(ctx context.Context) error

SweepOnce fires all due timers once. It is leader-gated so an N-node cluster fires each timer exactly once.

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.