askuser

Documentation

Overview

Package askuser is the per-domain Store over aura.paused_states (PRD 1.5). It copies the canonical Store pattern proved in internal/identity (D-A4-01): Store{pool,q} over the generated sqlc surface, SQLSTATE-based error classification via errors.As + pgErr.Code (never message matching), sentinel errors, pgtype conversion at the boundary, and db.WithTx for atomic multi-row writes.

Scope is the pause-PERSISTENCE half (SPEC Req#3): FIFO ListPending with a total order + crash recovery + Resume/Batch + Loop.Stop auto-resolve. The Runner (04-05) owns resume ORCHESTRATION and is the sole writer beyond this Store's Insert. This package NEVER imports internal/agent/tools (D-A1-04) — it takes plain fields; the Event carries the pause payload, not a tools type. There is no internal timeout / expiry state (SPEC Req#4): the only resolution paths are Resume, ResumeBatch, and AutoResolveForConversation.

Index

• Constants
• Variables
• type InsertParams
• type Pending
• type Record
• type ResumeAnswer
• type Store
  • func New(pool *pgxpool.Pool) *Store
  • func (s *Store) AutoResolveForConversation(ctx context.Context, conversationID string) error
  • func (s *Store) CleanupResumedOlderThan(ctx context.Context, cutoff pgtype.Timestamptz) error
  • func (s *Store) GetByToken(ctx context.Context, token string) (Pending, error)
  • func (s *Store) Insert(ctx context.Context, p InsertParams) error
  • func (s *Store) ListPending(ctx context.Context, conversationID string) ([]Pending, error)
  • func (s *Store) ListPendingAll(ctx context.Context, limit int) ([]Pending, error)
  • func (s *Store) ListRecent(ctx context.Context, limit int) ([]Record, error)
  • func (s *Store) MarkResumed(ctx context.Context, token string, ans ResumeAnswer) error
  • func (s *Store) MarkResumedBatch(ctx context.Context, answers map[string]ResumeAnswer) error

Constants

const (
	ActionAccept  = "accept"
	ActionDecline = "decline"
	ActionCancel  = "cancel"
)

Action is the MCP three-action resolution model (AM-02 / D-A3-01). resumed_answer is stored as the JSON {action, content}: accept injects content as a RoleTool, decline tells the model the user declined, cancel aborts the turn.

Variables

var (
	ErrPauseNotFound = errors.New("paused state not found or already resumed")
	ErrInvalidAnswer = errors.New("invalid resume answer")
)

Sentinel errors so callers classify failures without string matching. ErrPauseNotFound is an unknown/already-resolved token; ErrInvalidAnswer is a malformed resolution action.

Types

type InsertParams

type InsertParams struct {
	Token              string
	ConversationID     string
	Kind               string
	Question           string
	Options            json.RawMessage
	Priority           int
	ToolCallID         string
	ResumeContext      json.RawMessage
	ProxiedFromChildID *string
	ProxiedToolCallID  string
}

InsertParams carries the plain fields for one new pause. NEVER a tools type (D-A1-04): the caller (the Runner, observing the pause Event) supplies these. ProxiedFromChildID/ProxiedToolCallID are the optional D-05 swarm-relay ids: a nil child id (and an empty tool_call id) persist as SQL NULL for direct calls.

type Pending

type Pending struct {
	Token              string
	ConversationID     string
	Kind               string
	Question           string
	Options            json.RawMessage
	Priority           int
	ToolCallID         string
	ResumeContext      json.RawMessage
	ProxiedFromChildID string
}

Pending is the domain projection of a pending aura.paused_states row — plain Go types at the package boundary instead of the sqlc pgtype wrappers. ProxiedFromChildID is the flat swarm-worker id ("w1".."wN", D-05/D-15) the pause proxies for, or "" for a direct (non-proxied) call — it is the APRV-01 "source thread" the cross-thread approval list shows so the operator knows which (possibly background) thread raised the interrupt.

type Record

type Record struct {
	Token          string
	ConversationID string
	Kind           string
	Question       string
	Priority       int
	Resumed        bool
	ResumedAnswer  string // the {action,content} content, or "" when still pending
}

Record is the richer projection of a paused_states row for the operator-facing `aura paused-states list` CLI: it carries the resolution state + the persisted resumed_answer (the auto-terminated marker when Loop.Stop closed it), which the pending-only Pending projection omits.

type ResumeAnswer

type ResumeAnswer struct {
	Action  string `json:"action"`
	Content string `json:"content"`
}

ResumeAnswer is the AM-02 resolution payload persisted as resumed_answer jsonb.

type Store

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

Store wraps a pgx pool and the generated Queries — the canonical shape from internal/identity. Non-tx reads/writes use s.q; multi-row atomic writes (MarkResumedBatch, AutoResolveForConversation) wrap db.WithTx.

func New

func New(pool *pgxpool.Pool) *Store

New builds a Store over an open pool.

func (*Store) AutoResolveForConversation

func (s *Store) AutoResolveForConversation(ctx context.Context, conversationID string) error

AutoResolveForConversation is the SPEC Req#11 Loop.Stop helper: resolve every open pending for a conversation with the auto-terminated marker, atomically. It leaves zero resumed_at IS NULL rows for that conversation.

func (*Store) CleanupResumedOlderThan

func (s *Store) CleanupResumedOlderThan(ctx context.Context, cutoff pgtype.Timestamptz) error

CleanupResumedOlderThan deletes resolved rows older than the cutoff (a GC the CLI `aura paused-states purge` drives). It never touches pending rows.

func (*Store) GetByToken

func (s *Store) GetByToken(ctx context.Context, token string) (Pending, error)

GetByToken fetches one pause by token, mapping a missing row to ErrPauseNotFound.

func (*Store) Insert

func (s *Store) Insert(ctx context.Context, p InsertParams) error

Insert persists one pending pause. Options/ResumeContext are jsonb (nil → SQL NULL). The proxied_* columns carry the D-05 relay ids when the pause proxies a child's needs_user_input report; they stay NULL for direct calls (a nil child id leaves the pgtype.Text Valid:false, an empty tool_call id leaves the pgtype.Text Valid:false). proxied_from_child_id is the flat worker id ("w1".."wN", D-15/D-16) stored verbatim as text — NOT a uuid (CR-01): the swarm report carries no uuid for a model to relay, so parsing one here would fail the documented happy path.

func (*Store) ListPending

func (s *Store) ListPending(ctx context.Context, conversationID string) ([]Pending, error)

ListPending returns the conversation's still-pending pauses (resumed_at IS NULL) in the total FIFO order priority DESC, created_at ASC, token ASC. The token tiebreaker is mandatory: rows inserted in one tx share created_at = now() (RESEARCH Pitfall 4), so without it the order would be non-deterministic.

func (*Store) ListPendingAll

func (s *Store) ListPendingAll(ctx context.Context, limit int) ([]Pending, error)

ListPendingAll returns the still-pending pauses ACROSS ALL conversations (resumed_at IS NULL) in the same total FIFO order as ListPending — priority DESC, created_at ASC, token ASC — capped at limit (APRV-01 / D-04, the approval center's cross-thread read). The token tiebreaker is mandatory for the same reason as ListPending: tx-batched rows share created_at = now() (RESEARCH Pitfall 4), so without it the cross-thread order would be non-deterministic. limit<=0 falls back to 100 (mirroring ListRecent's <=0 guard). It reuses the SAME fromRow projector, so each Pending carries Question/Options/Priority/Kind/ConversationID and the source thread (ProxiedFromChildID) the operator needs to jump to the originating thread.

func (*Store) ListRecent

func (s *Store) ListRecent(ctx context.Context, limit int) ([]Record, error)

ListRecent returns the most-recent paused_states rows (pending + resolved) across all conversations, newest first, for the CLI. limit<=0 falls back to 50.

func (*Store) MarkResumed

func (s *Store) MarkResumed(ctx context.Context, token string, ans ResumeAnswer) error

MarkResumed resolves one pause with the AM-02 {action, content} answer. An unknown / already-resumed token (zero rows affected) returns ErrPauseNotFound so the caller gets a clear error rather than a silent no-op.

func (*Store) MarkResumedBatch

func (s *Store) MarkResumedBatch(ctx context.Context, answers map[string]ResumeAnswer) error

MarkResumedBatch resolves many pauses atomically (one tx via db.WithTx). Every token must resolve a still-pending row; if any token is unknown/already-resumed the whole batch rolls back with ErrPauseNotFound (no partial resolution).