transcript

package
v0.2.0 Latest Latest
Warning

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

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

Documentation

Overview

Pause records and resume tokens — the storage half of the v0.2 programmatic pause/abort surface (docs/durable-execution-design.md, "The v0.2 pause/abort mechanics").

Two pause planes share one record shape:

  • Gate pause (plane B): a single per-session record under pauseGateKey on the companion ops row. While active, the daemon's turn chokepoint refuses every turn kind on the session.
  • Interrupt pause (plane A): one record per parked interrupt, under pauseIntrKeyPrefix + interruptID. The park itself is the ADK LongRunningToolIDs event; the record adds the resume token, scope, and timer metadata.

Per-interrupt keys are deliberate (adversarial-gate finding M7): a shared JSON map would make mint/consume a cross-process read-modify-write; per-key records never collide. Records are consumed, not deleted — ConsumedAt distinguishes "resumed already" (a structured no-op) from "no such token" (an operator error). Tokens are minted here, never caller-chosen, so a weak caller token cannot undermine the capability.

Package session is the operator-facing read/inspect surface over ADK's session store (docs/durable-execution-design.md, "Operator-facing surface"). It answers the questions durability raises for an operator: which sessions exist, which are paused waiting for input, on what interrupt, and with what response schema.

The pause model it inspects is the one verified in spike 2 (docs/spike-findings.md) and extended by the v0.2 pause/abort design (docs/durable-execution-design.md, "The v0.2 pause/abort mechanics"): a paused session carries a pending interrupt — an event with a non-nil RequestedInput, OR an unanswered LongRunningToolIDs entry (a long-running tool park: request_operator_input, pause_session; both spellings of ADK's one pause primitive) — or an active gate-pause record on its ops row. The matching resume for an interrupt is a later user turn whose FunctionResponse.ID equals the interrupt ID. The LongRunningToolIDs source was added by the v0.2 design's adversarial gate (finding H1): without it, v0.1 planner parks projected idle/interrupted — invisible to operators and, worse, auto-resume candidates.

State labels are derived strictly from what the store can prove:

  • StatePaused: at least one pending (unresolved) interrupt, or an active gate pause (Store.PauseGate).
  • StateAborted: an operator abort marker is present (written by Store.Abort to the session's companion ops row — see opsSuffix; v0.1.0 markers in the primary row's state are still honored).
  • StateInterrupted: a daemon shutdown cut a turn short — the daemon wrote an interruption marker before draining (Store.MarkInterrupted) and no clean completion cleared it (Store.ClearInterrupted). This is still strictly log-proven: the process that WAS running the turn recorded the fact durably before it stopped.
  • StateIdle: everything else. The store cannot distinguish "a turn is in flight right now" from "the last turn completed" — that is in-process runner state, not event-log state — so this package deliberately does not claim "running" or "completed".

Precedence: aborted > paused > interrupted > idle.

Index

Constants

View Source 
const (
	PlaneGate      = "gate"
	PlaneInterrupt = "interrupt"
)

Pause planes, as recorded in PauseRecord.Plane.

View Source 
const (
	StatePaused      = "paused"
	StateAborted     = "aborted"
	StateInterrupted = "interrupted"
	StateIdle        = "idle"
)

Session states derived from the event log. See the package doc for why there is no "running" or "completed".

View Source 
const DefaultTokenTTL = 7 * 24 * time.Hour

DefaultTokenTTL is the default resume-token lifetime (open question #2, resolved via #9's direction). PauseSpec.TokenTTL may only shorten it; lengthening is exclusively the audited ExtendToken path.

Variables

View Source 
var (
	ErrTokenNotFound  = fmt.Errorf("resume token not found: %w", ErrNotFound)
	ErrTokenExpired   = errors.New("resume token expired (the pause remains; extend-token is the recovery)")
	ErrAlreadyResumed = errors.New("pause already resumed")
)

Pause/token errors. ErrTokenNotFound wraps ErrNotFound so surfaces that map not-found to 400/exit-1 handle both uniformly.

View Source 
var ErrAlreadyAborted = errors.New("session already aborted")

ErrAlreadyAborted reports that an abort marker is already present.

View Source 
var ErrNotFound = errors.New("session not found")

ErrNotFound reports that no session with the requested ID exists in the store (under the store's app name).

Functions

func IsReservedSessionID added in v0.1.2 

func IsReservedSessionID(sessionID string) bool

IsReservedSessionID reports whether sessionID names a companion ops row rather than a real session. Every surface that accepts a session ID must refuse reserved IDs (#56) — a runner turn driven into an ops row would hold its write lease and corrupt subsequent marker writes; a Get would present marker storage as a phantom session.

func ValidReasons added in v0.2.0 

func ValidReasons() []string

ValidReasons lists the accepted Reason values, for error messages and CLI help.

Types

type Detail 

type Detail struct {
	Summary
	EventCount int            `json:"event_count"`
	Pending    []PendingInput `json:"pending,omitempty"`
	GatePause  *PauseRecord   `json:"gate_pause,omitempty"`
}

Detail is the show-view projection: Summary plus event count, the full pending-interrupt records, and the active gate pause if any.

type InterruptedCandidate added in v0.2.0 

type InterruptedCandidate struct {
	SessionID       string
	UserID          string
	InterruptReason string
	InterruptedAt   time.Time
	LastEventTime   time.Time
	EventCount      int
	Events          adksession.Events
}

InterruptedCandidate is one session projected as StateInterrupted, with the material the daemon's boot-time auto-resume pass needs (cmd/mast, #41): the resume identity (SessionID/UserID), the freshness inputs (InterruptedAt), the supersession-recheck inputs (LastEventTime/ EventCount), and the loaded Events for the effects dangling scan.

type PauseHandle added in v0.2.0 

type PauseHandle struct {
	Token     string    `json:"token"`
	SessionID string    `json:"session_id"`
	ExpiresAt time.Time `json:"expires_at"`
}

PauseHandle is what a successful pause returns to its caller.

type PauseRecord added in v0.2.0 

type PauseRecord struct {
	Token       string         `json:"token"`
	Plane       string         `json:"plane"`
	InterruptID string         `json:"interrupt_id,omitempty"`
	App         string         `json:"app"`
	User        string         `json:"user"`
	SessionID   string         `json:"session_id"`
	Reason      Reason         `json:"reason"`
	Message     string         `json:"message,omitempty"`
	Metadata    map[string]any `json:"metadata,omitempty"`
	MintedAt    time.Time      `json:"minted_at"`
	ExpiresAt   time.Time      `json:"expires_at"`
	ResumeAt    time.Time      `json:"resume_at,omitzero"`
	ConsumedAt  time.Time      `json:"consumed_at,omitzero"`
	ConsumedBy  string         `json:"consumed_by,omitempty"`
}

PauseRecord is the durable pause record + resume token, stored as JSON on the session's companion ops row. Scope (App, User) is checked at resume before any execution (open question #9, ratified); v0.2's single-tenant reality makes scope the (app, user) pair, and a tenant field slots in here when multi-tenancy lands.

func (*PauseRecord) Active added in v0.2.0 

func (r *PauseRecord) Active() bool

Active reports whether the record still gates/awaits a resume.

func (*PauseRecord) Expired added in v0.2.0 

func (r *PauseRecord) Expired(now time.Time) bool

Expired reports whether the record's token has passed its TTL. An expired token refuses resume but the pause itself remains.

type PauseSpec added in v0.2.0 

type PauseSpec struct {
	Reason    Reason
	Message   string
	Metadata  map[string]any
	ResumeAt  time.Time
	Interrupt bool
	TokenTTL  time.Duration
}

PauseSpec is the caller-facing pause request (docs/durable-execution-design.md, final PauseSpec). Interrupt is a daemon-level instruction (cancel the in-flight turn) and is not persisted in the record.

type PendingInput 

type PendingInput struct {
	// InterruptID is the resume correlation key: the resume turn's
	// FunctionResponse.ID must equal it (spike-2 verified contract).
	InterruptID string `json:"interrupt_id"`
	// Message is the human-readable prompt from the pausing node (for
	// long-running parks: the tool call's "message" argument, if any).
	Message string `json:"message,omitempty"`
	// LongRunning marks a long-running tool park (pause_session,
	// request_operator_input) rather than a RequestedInput. The resume
	// wire shape is identical either way.
	LongRunning bool `json:"long_running,omitempty"`
	// ToolName is the parked tool's name (long-running parks only).
	ToolName string `json:"tool_name,omitempty"`
	// Author is the agent that raised the interrupt.
	Author string `json:"author,omitempty"`
	// RaisedAt is the timestamp of the pausing event.
	RaisedAt time.Time `json:"raised_at"`
	// ResponseSchema, when non-nil, is the JSON schema the resume
	// response payload must conform to.
	ResponseSchema *jsonschema.Schema `json:"response_schema,omitempty"`
	// Payload is optional context the pausing node attached.
	Payload any `json:"payload,omitempty"`
}

PendingInput is a pending interrupt — a RequestedInput, or a long-running tool park — that has not been resolved by a later matching FunctionResponse. It carries everything an operator needs to script a resume.

type Reason added in v0.2.0 

type Reason string

Reason is the pause-reason taxonomy (open question #1, resolved: enum with an `other` escape hatch). 

const (
	ReasonBudgetExhaustion  Reason = "budget_exhaustion"
	ReasonWatchdogAnomaly   Reason = "watchdog_anomaly"
	ReasonCostCoolDown      Reason = "cost_cool_down"
	ReasonMaintenanceWindow Reason = "maintenance_window"
	ReasonRateLimitBackoff  Reason = "rate_limit_backoff"
	ReasonAmbiguity         Reason = "ambiguity"
	ReasonOperator          Reason = "operator"
	ReasonA2ATaskPending    Reason = "a2a_task_pending"
	ReasonOther             Reason = "other"
)

type Store 

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

Store wraps an ADK session.Service with the operator-facing projections. It works over any Service implementation — the SQLite / Postgres database service and the in-memory service alike.

func NewStore 

func NewStore(svc adksession.Service, appName string) *Store

NewStore wraps an already-open session service (the path the daemon uses: same service instance the runner writes through).

func Open 

func Open(path, appName string) (*Store, error)

Open opens the SQLite session DB at path read-through (the path the CLI uses: `mast sessions list/show --session-db=...` against a DB a daemon owns or owned). The file must already exist — opening a missing path would silently create an empty store and report zero sessions for what is actually an operator typo.

func (*Store) Abort 

func (s *Store) Abort(ctx context.Context, userID, sessionID, reason string) error

Abort appends a durable operator-abort marker for the session.

Semantics contract (minimal and honest — read before relying on it):

  • Abort is a marker, not preemption. It does NOT cancel a turn that is in flight in some daemon process; it appends an event whose StateDelta records the abort reason and time in the session's companion ops row (see opsSuffix — writing to the primary row would invalidate a live runner handle and kill the turn, the opposite of this contract; issue #46). The abort event is therefore NOT part of the primary transcript the model sees — previously incidental, since the daemon refuses resumes on aborted sessions anyway.
  • ADK's workflow reconstruction does not read the marker: as far as the engine is concerned, a pending RequestedInput is still resumable. It is mast's surface that treats the marker as terminal — List/Get report StateAborted with pending interrupts cleared, and the daemon's /resume handler refuses aborted sessions (cmd/mast). A real engine-level terminal state is the v0.2 programmatic-pause/abort work (docs/durable-execution-design.md, Phasing).
  • Idempotency: a second Abort returns ErrAlreadyAborted rather than stacking markers. Legacy v0.1.0 abort markers (written to the primary row's state) count.

func (*Store) AckEffects added in v0.2.0 

func (s *Store) AckEffects(ctx context.Context, userID, sessionID, reason string) error

AckEffects records the operator's acknowledgement of ambiguous prior effects: dangling mutating tool calls persisted at or before now stop tripping the recorded-effect outbox's ambiguous-effect mode (pkg/effects). The marker is a watermark, not a state — List/Get derivation ignores it. Re-acking overwrites (last write wins); the new watermark also covers intents the first ack already covered.

func (*Store) AutoResumeAttempts added in v0.2.0 

func (s *Store) AutoResumeAttempts(ctx context.Context, userID, sessionID string) (int, time.Time)

AutoResumeAttempts reports how many boot-time auto-resume attempts have been recorded for the session and when the last one was stamped (#41 restart-loop breaker). A missing or unparseable counter reads as zero — the fail-open direction for the breaker is to allow the attempt.

func (*Store) ClearAutoResumeAttempts added in v0.2.0 

func (s *Store) ClearAutoResumeAttempts(ctx context.Context, userID, sessionID string) error

ClearAutoResumeAttempts blanks the attempt counter after a successful auto-resume, so a session that later interrupts again starts fresh.

func (*Store) ClearInterrupted added in v0.1.1 

func (s *Store) ClearInterrupted(ctx context.Context, userID, sessionID string) error

ClearInterrupted resolves a MarkInterrupted marker after the turn completed inside the drain window. Clearing an unmarked session is a harmless no-op event (shutdown-path callers cannot atomically check).

func (*Store) ConsumeScheduled added in v0.2.0 

func (s *Store) ConsumeScheduled(ctx context.Context, token, by string) (*PauseRecord, error)

ConsumeScheduled consumes for the timed-pause scheduler. Unlike the operator path it does NOT enforce the token's TTL: a resume_at is the daemon's own scheduled commitment, and the operator-facing token expiry (a guard against stale possession) must not veto it. A resume_at legitimately set beyond the token's life — the only way to schedule a pause longer than the TTL cap, which mint can only shorten — would otherwise livelock the scheduler, firing forever against an expired token. The already-consumed no-op and scope check still hold.

func (*Store) ConsumeToken added in v0.2.0 

func (s *Store) ConsumeToken(ctx context.Context, token, by string) (*PauseRecord, error)

ConsumeToken marks the token's pause record consumed — for a gate pause this IS the resume (the chokepoint stops refusing); for an interrupt pause the caller drives the resume turn and consumes on the durable append of the resume FunctionResponse (adversarial-gate finding M5: consumption keys on the append, not on turn completion — a resume turn that fails later has still legitimately ended the pause). Scope is checked before anything else (OQ #9): under v0.2's single-tenant reality the store's app is the scope, and FindToken already lists within s.appName so a cross-app token reads as not-found; the belt-and-suspenders App check below is where a per-tenant/per-user check slots in when multi-tenancy lands. Expired tokens refuse with ErrTokenExpired and leave the pause intact — this is the operator-facing resume path.

func (*Store) EffectsAckedAt added in v0.2.0 

func (s *Store) EffectsAckedAt(ctx context.Context, userID, sessionID string) (time.Time, bool)

EffectsAckedAt returns the session's effects-acknowledgement watermark, if one was recorded. Read failures and missing markers both report false — the outbox then treats every dangling intent as unacknowledged, which is the fail-closed direction.

func (*Store) ExtendToken added in v0.2.0 

func (s *Store) ExtendToken(ctx context.Context, token string, ttl time.Duration) (*PauseRecord, error)

ExtendToken moves the token's expiry to now+ttl — the audited operator path for lengthening a token's life (mint can only shorten). Consumed tokens cannot be extended.

func (*Store) FindToken added in v0.2.0 

func (s *Store) FindToken(ctx context.Context, token string) (*PauseRecord, error)

FindToken resolves a resume token to its pause record (active or consumed — the caller distinguishes via ConsumedAt to give already_resumed its structured no-op).

func (*Store) GatePause added in v0.2.0 

func (s *Store) GatePause(ctx context.Context, userID, sessionID string) *PauseRecord

GatePause returns the session's active gate-pause record, if any. Read failures report nil — consistent with every other marker read: the ops row is an overlay, and an unreadable overlay must not wedge the session (the fail-open direction is deliberate here; abort and gate refusal are availability guards, not safety guards — the safety guard is the effects outbox, which fails closed).

func (*Store) Get 

func (s *Store) Get(ctx context.Context, userID, sessionID string) (*Detail, error)

Get returns the detail view for one session. An empty userID is resolved by scanning List for the session ID (the CLI knows session IDs, not the daemon-internal user ID). Reserved ops-row IDs are refused as not-found (#56).

func (*Store) List 

func (s *Store) List(ctx context.Context, userID string) ([]Summary, error)

List returns summaries for all sessions under the store's app name, most recent last-event first. userID narrows to one user; empty lists all users.

Note: ADK's Service.List returns sessions without events, and paused state is an event-log property, so List issues one Get per session. Fine at operator-CLI scale; a paged/indexed path is a v0.2+ concern alongside the eventlog query surface (docs/fork-design.md P1.3).

func (*Store) MarkInterrupted added in v0.1.1 

func (s *Store) MarkInterrupted(ctx context.Context, userID, sessionID, reason string) error

MarkInterrupted appends a durable interrupted-by-shutdown marker for the session (docs/durable-execution-design.md, "Shutdown contract").

The daemon writes it for every session with a turn in flight when a shutdown begins, BEFORE draining — so a SIGKILL mid-drain leaves the marker on disk — and clears it via ClearInterrupted when the turn completes inside the drain window. The marker lives in the companion ops row (see opsSuffix): writing it to the primary row would invalidate the live runner handle and kill the very turn being marked (issue #45). Like the abort marker it is state, not preemption: the engine ignores it, and a later turn on the session proceeds normally (reconstruct-and-re-execute); it exists so operators can see which sessions a restart cut short.

The primary session need not exist yet (a turn interrupted before the runner's auto-create): the marker parks in the ops row and surfaces if/when the primary appears. userID must then be explicit — with userID == "" resolution scans primaries and returns ErrNotFound. Re-marking overwrites (last write wins) — a second shutdown racing the first is not worth an error.

func (*Store) PauseGate added in v0.2.0 

func (s *Store) PauseGate(ctx context.Context, userID, sessionID string, spec PauseSpec) (PauseHandle, bool, error)

PauseGate writes (or updates) the session's plane-B gate pause. The session must exist (pausing a typo is an operator error). A second gate pause on an already-gated session updates reason, message, metadata, and resume_at in place — the token and its expiry are kept (open question #5: single-writer per session, no stacking). Pausing an aborted session is refused: aborted is terminal. The returned bool reports whether a NEW gate pause was opened (a fresh token minted) as opposed to an in-place refresh of an already-active one — so a caller counting distinct pauses does not double-count a refresh (#50).

func (*Store) PauseInterrupt added in v0.2.0 

func (s *Store) PauseInterrupt(ctx context.Context, userID, sessionID, interruptID string, spec PauseSpec) (PauseHandle, error)

PauseInterrupt mints the plane-A pause record for a parked interrupt (the pause_session tool body and the graph RequestInput helper call this). The park itself is the caller's ADK event; this only records the token. Re-minting for the same interrupt ID overwrites (last write wins — interrupt IDs are model-minted unique; a collision means a re-fire of the same call).

func (*Store) PauseRecords added in v0.2.0 

func (s *Store) PauseRecords(ctx context.Context, userID, sessionID string) (map[string]*PauseRecord, error)

PauseRecords returns all pause records for one session (active and consumed), keyed as stored. Used by show, the boot scan, and tests.

func (*Store) RecordAutoResumeAttempt added in v0.2.0 

func (s *Store) RecordAutoResumeAttempt(ctx context.Context, userID, sessionID string) (int, error)

RecordAutoResumeAttempt durably increments the session's auto-resume attempt counter and stamps the attempt time, returning the new count. The daemon calls it BEFORE driving the continuation turn so an attempt that crashes the process (the exact restart-loop threat) is still counted (#41 M2). The boot pass is a single sequential goroutine, so the read-then-write needs no cross-call locking beyond appendOpsDelta's own ops-row serialization.

func (*Store) ScanInterrupted added in v0.2.0 

func (s *Store) ScanInterrupted(ctx context.Context) ([]InterruptedCandidate, error)

ScanInterrupted returns every session under the store's app name that currently projects as StateInterrupted, oldest interruption first. It mirrors List (iterate primary rows, skip companion ops rows, Get+ project each) rather than the ops-row scans (ScanPauses): interrupted state is a projection over the primary transcript plus the ops-row marker, so a candidate must be found the way List finds sessions.

Like List it issues one Get per session; fine at single-instance operator scale (docs/fork-design.md P1.3 tracks an indexed path).

func (*Store) ScanPauses added in v0.2.0 

func (s *Store) ScanPauses(ctx context.Context) ([]*PauseRecord, error)

ScanPauses walks every session's ops row and returns all ACTIVE pause records. The daemon runs it once at boot to seed the token index and the timed-pause scheduler's heap; it is also FindToken's substrate. O(sessions) — the P1.3 eventlog/query surface owns fleet-scale indexing.

type Summary 

type Summary struct {
	ID            string    `json:"id"`
	AppName       string    `json:"app_name"`
	UserID        string    `json:"user_id"`
	LastEventTime time.Time `json:"last_event_time"`
	State         string    `json:"state"`
	// PendingInterruptIDs are the unresolved interrupt IDs (empty
	// unless State is StatePaused).
	PendingInterruptIDs []string `json:"pending_interrupt_ids,omitempty"`
	// AbortReason is set when State is StateAborted.
	AbortReason string `json:"abort_reason,omitempty"`
	// InterruptReason is set when State is StateInterrupted: the reason
	// recorded by the daemon whose shutdown cut the session's turn short.
	InterruptReason string `json:"interrupt_reason,omitempty"`
	// InterruptedAt is set when State is StateInterrupted: when the daemon
	// recorded the interruption (from interruptTimeKey). Drives the
	// auto-resume freshness window (cmd/mast, #41).
	InterruptedAt time.Time `json:"interrupted_at,omitempty"`
	// PauseReason / PauseMessage are set when an active gate pause
	// contributes to StatePaused (Store.PauseGate).
	PauseReason  string `json:"pause_reason,omitempty"`
	PauseMessage string `json:"pause_message,omitempty"`
}

Summary is the list-view projection of one session.

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.