Documentation
¶
Overview ¶
Package outbox implements the D3 durable-prompt outbox (design 0050, #907): accept-then-deliver decoupling so a client disconnect can never lose an accepted message.
The incident (2026-08-15/16) lost six messages to iOS killing in-flight POSTs — the synchronous adapter.Send bound delivery to the client's request context, so a disconnect canceled it mid-flight. The outbox accepts into Valkey and a detached worker delivers.
Storage layout (per session):
outboxq:{ws}:{ses} LIST — FIFO of JSON entries (the durable queue)
outboxd:{ws}:{ses} LIST — staging: entries currently being delivered
(crash recovery requeues these on start)
outboxdedupe:{ws}:{ses}:{cmid} STRING — dedupe marker; VALUE is the
accepted entry ID (set only AFTER a successful
push — a marker with no entry is a false
"duplicate" that silently drops retries)
outboxlock:{ws}:{ses} STRING — per-session delivery lock; value is a
random owner token, released only by the owner
(compare-and-del), TTL > DeliveryTimeout so a
long turn cannot expire its own lock
Delivery semantics: at-least-once for PROVEN-not-delivered failures only. An attempt whose outcome is UNKNOWN (send timeout mid-turn, transport cut mid-flight) is never blind-retried: opencode persists the user message before the turn starts, so the message is most likely in the transcript. Unknown outcomes move the entry to verifying and a Verifier (transcript check) decides: delivered → remove; absent → retry; inconclusive → recheck with backoff (#987 — the sent-once/ delivered-3x incident class).
Crash-safety: every multi-step transition is ordered so the worst case is DUPLICATE delivery (at-least-once, collapsed at render via clientMessageID), never LOSS:
- stage-out: RPUSH staging BEFORE LREM main (crash between → entry in both → delivered possibly twice, never zero; Recover dedupes by ID)
- restore: reinsert main BEFORE LREM staging
- recover: LRANGE (no pop), dedupe against main by ID, then DEL
- recover status: staged entries requeue as VERIFYING — the outcome of the interrupted send is unknown, same as a timeout
Index ¶
- Constants
- Variables
- func Ambiguous(err error) error
- func PriorAttemptPending(err error) error
- func Transient(err error) error
- type AmbiguousError
- type DeliveredHook
- type Deliverer
- type DismissResult
- type Duplicate
- type Entry
- type LedgerProbe
- type PriorAttemptPendingError
- type RetryResult
- type Service
- func (s *Service) Accept(ctx context.Context, ...) (*Entry, error)
- func (s *Service) CleanupWorkspace(ctx context.Context, workspaceID string) (int, error)
- func (s *Service) DeliverOnce(ctx context.Context, ws, ses string, d Deliverer) bool
- func (s *Service) Dismiss(ctx context.Context, workspaceID, sessionID, id string) DismissResult
- func (s *Service) List(ctx context.Context, workspaceID, sessionID string) ([]Entry, error)
- func (s *Service) ParkWorkspace(ctx context.Context, workspaceID, reason string) (int, error)
- func (s *Service) Recover(ctx context.Context) int
- func (s *Service) Retry(ctx context.Context, workspaceID, sessionID, id string) RetryResult
- func (s *Service) Run(ctx context.Context, d Deliverer, tick time.Duration)
- func (s *Service) SeedEntryForTest(ctx context.Context, ws, ses, rawEntry string) error
- func (s *Service) SetLedgerProbe(p LedgerProbe)
- func (s *Service) SetOnDelivered(h DeliveredHook)
- func (s *Service) SetOnStaged(h StagedHook)
- func (s *Service) SetVerifier(v Verifier)
- func (s *Service) SweepWorkspaceParkedErrors(ctx context.Context, workspaceID string) (int, error)
- func (s *Service) SweepWorkspaceUnverifiable(ctx context.Context, workspaceID string) (int, error)
- func (s *Service) UnparkWorkspace(ctx context.Context, workspaceID string) (int, error)
- type StagedHook
- type TransientDeliveryError
- type Verdict
- type Verifier
Constants ¶
const ( StatusPending = "pending" StatusDelivering = "delivering" StatusVerifying = "verifying" StatusError = "error" )
Status values.
const ( LedgerStateLedgered = "LEDGER_STATE_LEDGERED" LedgerStateAdmitted = "LEDGER_STATE_ADMITTED" LedgerStatePromoted = "LEDGER_STATE_PROMOTED" LedgerStateTurnEnded = "LEDGER_STATE_TURN_ENDED" LedgerStateStalled = "LEDGER_STATE_STALLED" LedgerStateFailed = "LEDGER_STATE_FAILED" )
Ledger state constants — the frozen ABI enum names (pkg/abi/v1 LedgerState_*). Kept local so the outbox keeps zero generated-code coupling (same principle as the terminus's hand-rolled transport); a handler-side test pins them against the generated enum.
const StatusParked = "parked"
StatusParked marks an entry parked by an operator action (the flip gate's mode_transition). Parked entries are inert to the delivery loop — no auto-retry, no backoff — and re-arm only through UnparkWorkspace (the rollback drain) or explicit dismissal.
Variables ¶
var ( // MaxAttempts bounds retries before an entry parks as error. With the // exponential backoff below, 5 attempts span ~3.5 minutes — covering // an opencode restart cycle (the #903 reconciler pattern: retries // must span agent generations, not exhaust in seconds). MaxAttempts = 5 // Cap is the per-session outbox length limit (accept returns capped). Cap = 25 // DedupeTTL bounds how long a clientMessageID is remembered. DedupeTTL = 24 * time.Hour // DeliveryTimeout bounds one delivery attempt (a long turn). DeliveryTimeout = 10 * time.Minute // RetryBackoff is the BASE delay before attempt N+1; it doubles per // attempt and caps at MaxBackoff. Var for tests. RetryBackoff = 10 * time.Second // MaxBackoff caps the exponential retry delay. MaxBackoff = 2 * time.Minute // LockTTL must exceed DeliveryTimeout: a delivery longer than the // lock's TTL would expire mid-turn and let a second worker deliver // concurrently (one-in-flight violation). LockTTL = DeliveryTimeout + 2*time.Minute // VerifyDelay is the initial delay before the first verification // pass after an ambiguous send (lets a racing persist land). VerifyDelay = 3 * time.Second // VerifyBackoff is the BASE delay between inconclusive verification // passes; it doubles per pass and caps at MaxVerifyBackoff. VerifyBackoff = 3 * time.Second // MaxVerifyBackoff caps the verification recheck delay. MaxVerifyBackoff = 2 * time.Minute // MaxVerifyAttempts bounds inconclusive passes before the entry // parks as error (agent unreachable for the full backoff span). MaxVerifyAttempts = 40 // MaxTransientFailures bounds no-attempt-driven failures (resolve // errors, canceled polls) before the entry parks — transient // failures never mint attempt numbers (#1316 review round 2), so // they need their own convergence bound. MaxTransientFailures = 20 )
Tunables (vars for tests).
var ErrCapped = errors.New("session outbox is full")
ErrCapped is returned when the session's outbox is at Cap.
var ( // ParkedSweepInterval bounds parked-entry indeterminacy: L9 // (#1312/#1316 proposed budget) caps parked → {completed, re-armed, // confirmed-terminal} at ≤5min; a 60s sweep leaves four re-tries of // headroom for unreachable pods. This is an API-side recovery bound, // distinct from the agentd lease clock (#1319: LeaseConvergenceBound // 30s / ReconcileCadence 15s governs lease expiry, not parked-pill // recovery); coherence between the two families is pinned in the // #1312 budget table (proposal on #1314). ParkedSweepInterval = 60 * time.Second )
Sweep cadence and guard tunables (vars for tests).
Functions ¶
func PriorAttemptPending ¶ added in v0.29.0
PriorAttemptPending wraps err as an agentd-owned prior-attempt timeout.
Types ¶
type AmbiguousError ¶ added in v0.19.0
type AmbiguousError struct{ Err error }
AmbiguousError marks a deliverer outcome as UNKNOWN: the send may or may not have persisted (timeout mid-turn, transport cut mid-flight). The deliverer wraps such errors via Ambiguous; the outbox responds by verifying against the agent transcript instead of blind-retrying (#987: blind retries re-POST the same text as a NEW message and duplicate the turn).
func (*AmbiguousError) Error ¶ added in v0.19.0
func (a *AmbiguousError) Error() string
func (*AmbiguousError) Unwrap ¶ added in v0.19.0
func (a *AmbiguousError) Unwrap() error
type DeliveredHook ¶ added in v0.19.0
DeliveredHook fires exactly once per entry on confirmed delivery — the synchronous 2xx path and the verified path. Concurrent completers (this replica's loops or a peer's) are serialized by the atomic cross-list claim: only the completer whose claim removed at least one copy fires. Failure direction is at-least-once (a failed claim retains the entry for the next completer). SSE queue.update/sent, metering, and session-index recording ride it.
type Deliverer ¶
Deliverer delivers one entry. The worker calls it with a detached, timeout-bounded context — it must not reference the accepting request.
type DismissResult ¶ added in v0.29.0
type DismissResult int
DismissResult discriminates the queue UI's dismiss outcome — the same contention-is-not-absence principle as RetryResult (r4 review).
const ( DismissRemoved DismissResult = iota DismissNotFound DismissBusy )
type Duplicate ¶
type Duplicate struct{ AcceptedID string }
Duplicate is returned when the clientMessageID was already accepted. AcceptedID carries the original entry's ID.
type Entry ¶
type Entry struct {
ID string `json:"id"` // outbox entry ID (unique per accept)
ClientMessageID string `json:"clientMessageID"` // caller-supplied dedupe key
UserID string `json:"userID"` // accepting user (metering at delivery)
Text string `json:"text"` // prompt text (already validated ≤100KB)
Model json.RawMessage `json:"model,omitempty"` // per-prompt model selector (raw JSON)
AcceptedAt time.Time `json:"acceptedAt"`
Attempts int `json:"attempts"` // definitive failures only
LastAttemptAt time.Time `json:"lastAttemptAt,omitempty"` // send-window start (verifier anchor)
VerifyAttempts int `json:"verifyAttempts,omitempty"`
TransientFails int `json:"transientFails,omitempty"` // no-attempt-driven failures (never mint numbering)
NextAttemptAt time.Time `json:"nextAttemptAt,omitempty"` // backoff gate (zero = now)
LastError string `json:"lastError,omitempty"`
Status string `json:"status"` // pending | delivering | verifying | error
// #1019 D: frozen-queue surfacing. Set by List (never persisted):
// pending entries whose session lock is held — by a delivering worker
// (healthy queueing) or a stale lock from an ungraceful kill (the
// residual 12-min freeze). InFlightFor approximates the hold age
// (LockTTL − remaining TTL), letting clients distinguish "sending…"
// from "frozen behind a stale lock".
BlockedByInFlight bool `json:"blockedByInFlight,omitempty"`
InFlightFor time.Duration `json:"inFlightFor,omitempty"`
}
Entry is one accepted prompt in the outbox.
type LedgerProbe ¶ added in v0.29.0
type LedgerProbe func(ctx context.Context, workspaceID, sessionID, entryID string, attempt uint32) (state string, err error)
LedgerProbe reports the agentd delivery ledger's state for (entry, attempt). Contract:
- ("", nil): no row exists for that (entry, attempt)
- (state, nil): the row's current state
- (_, err): the probe failed (pod unreachable) — indeterminate
Read-only and idempotent; wired iff the agentd terminus deliverer is the delivery regime (adapter mode leaves it nil and the sweeper is a no-op, preserving legacy behavior).
type PriorAttemptPendingError ¶ added in v0.29.0
type PriorAttemptPendingError struct{ Err error }
PriorAttemptPendingError marks a terminus outcome where the PRIOR attempt's ledger row is still LEDGERED and agentd owns admission: no new attempt was driven, so the outbox must not mint an attempt number (a phantom attempt parks the entry against a row the sweeper can never find — the unrecoverable shape from the #1316 review).
func (*PriorAttemptPendingError) Error ¶ added in v0.29.0
func (p *PriorAttemptPendingError) Error() string
func (*PriorAttemptPendingError) Unwrap ¶ added in v0.29.0
func (p *PriorAttemptPendingError) Unwrap() error
type RetryResult ¶ added in v0.29.0
type RetryResult int
RetryResult discriminates the queue UI's retry outcome: contention is NOT absence — surfacing it as 404 told users their entry was gone while a delivery merely held the session lock (r3 review finding 4).
const ( RetryUpdated RetryResult = iota RetryNotFound RetryBusy )
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service is the outbox. All methods are safe for concurrent use; the Valkey client is borrowed (lifecycle owned by the caller).
func New ¶
New returns a Service backed by client, or nil if client is nil (outbox disabled — callers must nil-check).
func (*Service) Accept ¶
func (s *Service) Accept(ctx context.Context, workspaceID, sessionID, userID, clientMessageID, text string, model json.RawMessage) (*Entry, error)
Accept validates nothing (callers validate); it caps, persists, and only THEN records the dedupe marker. Marker ordering matters: a marker set before persistence survives a failed push for the dedupe TTL and turns every retry into a false "duplicate" — silent loss (the exact class D3 exists to kill; round-1 finding 3). Returns the accepted entry (status pending). A Duplicate return carries the original ID — callers should respond 200 with it, not an error.
func (*Service) CleanupWorkspace ¶ added in v0.27.0
CleanupWorkspace deletes every outbox key of the workspace — queues, staging, dedupe markers, locks (#1119 follow-up from the #1211 F2 triage: deleting a Workspace CR left orphaned Valkey residue with no agent to verify against, no queue UI to dismiss from, and no expiry; the rows inflated the outbox gauges forever and were re-scanned every worker tick). Returns the number of keys removed. Idempotent: a missed sweep leaves the keys for the next transition.
func (*Service) DeliverOnce ¶
DeliverOnce delivers the first due pending entry for one session (same semantics as the Run loop's per-session step). Exported for the handler-level tests; the Run loop is the production driver.
func (*Service) Dismiss ¶
func (s *Service) Dismiss(ctx context.Context, workspaceID, sessionID, id string) DismissResult
Dismiss removes an entry by ID (the queue UI's dismiss action). Holds the session delivery lock: a lock-free value-LRem mid-sweep shifts the sweeper's snapshot indices — its subsequent LSet lands on an innocent neighbor (the r3 reproduction of the sibling-mutator class).
func (*Service) List ¶
List returns the session's outbox entries (FIFO order): pending and error entries from the main list plus any stuck in delivering. Pending entries carry the frozen-queue signal (#1019 D): when the session's delivery lock is held — by a live worker or a stale lock — blockedByInFlight is set with the approximate hold age.
func (*Service) ParkWorkspace ¶ added in v0.27.0
ParkWorkspace moves every in-flight (pending/delivering/verifying) outbox entry of the workspace to StatusParked carrying the explicit reason — design 0055 M4's drain-before-flip: entries that would race a mode change are held visible with why, never auto-re-sent. Returns the number parked.
func (*Service) Recover ¶
Recover requeues staged (delivering) entries left by a crash — run at worker start. LRANGE (no pop) + ID-dedupe against main, then DEL: the crash windows above can leave an entry in both lists; the ID check prevents Recover itself from duplicating. Requeued entries enter VERIFYING: the interrupted send's outcome is unknown — blind re-send is the #987 duplicate class.
Holds the per-session lock BEFORE reading staging: the requeue decision (entry absent from main) is only valid against state no concurrent delivery can change. A pre-lock staging snapshot races a completing delivery — restore → verify → claim between the snapshot and the lock acquisition leaves the entry absent from main, and the stale snapshot requeues it as a verifying zombie: a later pass verifies and claims it again, firing OnDelivered twice for one send (the 2026-09-16 multi-replica storm failure). This is also why the head-LPush shifts every snapshot index and the boot overlap with the seed transition's parked sweep matters: a sweep LRange before the LPush with its LSet after would overwrite an innocent neighbor (the r4 review's boot-overlap window — seconds of probe I/O, not µs).
func (*Service) Retry ¶
func (s *Service) Retry(ctx context.Context, workspaceID, sessionID, id string) RetryResult
Retry clears an error entry back to pending (the queue UI's retry). Holds the session delivery lock: a lock-free snapshot LSet racing a concurrent sweep's LRem shifts indices and overwrites an innocent neighbor (#1316 review round 2 finding 1).
func (*Service) Run ¶
Run drives delivery until ctx is done. Sessions are scanned each tick and delivered CONCURRENTLY (bounded semaphore): deliverOne blocks for up to DeliveryTimeout on a long turn, and a single sequential loop would let one slow session starve every other session's outbox — the starvation shape design 0050 exists to eliminate. The per-session lock serializes same-session delivery; cross-session delivery is parallel. Run drives delivery until ctx is canceled, THEN joins every worker it spawned before returning. The join is load-bearing: abandoning in-flight deliverOne goroutines on exit (a) leaks them past the caller's lifetime — observed as goroutine-stack races against subsequent package-var mutations in tests, and as unbounded teardown stalls behind detached deliveries in production shutdown — and (b) makes "Run returned" mean nothing. Workers quiesce promptly on a canceled ctx: pre-delivery Redis ops fail fast, in-flight deliveries complete under their detached contexts (D3) with detached-bounded bookkeeping, bounded by DeliveryTimeout + bookkeepingTimeout.
func (*Service) SeedEntryForTest ¶ added in v0.27.0
SeedEntryForTest pushes a raw queue entry (test harnesses only).
func (*Service) SetLedgerProbe ¶ added in v0.29.0
func (s *Service) SetLedgerProbe(p LedgerProbe)
SetLedgerProbe wires the ledger truth source. Call before Run.
func (*Service) SetOnDelivered ¶ added in v0.19.0
func (s *Service) SetOnDelivered(h DeliveredHook)
SetOnDelivered wires the confirmed-delivery hook. Call before Run.
func (*Service) SetOnStaged ¶ added in v0.25.1
func (s *Service) SetOnStaged(h StagedHook)
SetOnStaged wires the picked-up-for-delivery hook. Call before Run.
func (*Service) SetVerifier ¶ added in v0.19.0
SetVerifier wires the ambiguity resolver. Call before Run.
func (*Service) SweepWorkspaceParkedErrors ¶ added in v0.29.0
SweepWorkspaceParkedErrors re-verifies one workspace's parked error entries against the ledger. Returns the number recovered (completed or re-armed). Best-effort per session — a missed entry retries on the next sweep; idempotent on already-swept state.
func (*Service) SweepWorkspaceUnverifiable ¶ added in v0.25.1
SweepWorkspaceUnverifiable re-arms parked unverifiable error entries for one workspace's sessions: StatusError → StatusVerifying with a fresh verify window (VerifyAttempts reset, due immediately). The verify-first path (#987) then either confirms delivery (entry removed, no duplicate send) or resumes bounded verification. Entries parked for any other reason stay parked. Returns the number of re-armed entries.
Called on the Active phase transition (proxy onPhaseChange): a resume is precisely when an agent that was unreachable becomes reachable, so the stranded-entry class self-heals instead of waiting for manual Retry. Safe to call repeatedly (idempotent on already-swept state).
func (*Service) UnparkWorkspace ¶ added in v0.27.0
UnparkWorkspace re-arms exactly the mode_transition parks of the workspace back to pending (the rollback's ledger back-drain via the 0052 path). Returns the number re-armed.
type StagedHook ¶ added in v0.25.1
StagedHook fires when a pending entry is staged out for delivery — the POST is starting and the entry has left the visible queue. SSE queue.update/delivering rides it: the frontend's pill-clearing signal for in-flight delivery. Without it, a multi-minute turn (the V1 sync send blocks until the assistant completes) keeps the entry rendering as "queued" long after the agent owns the message — the TUI-parity gap: once admitted, a prompt is in the conversation, not the queue. Confirmed-delivery bookkeeping still rides DeliveredHook at turn end.
type TransientDeliveryError ¶ added in v0.29.0
type TransientDeliveryError struct{ Err error }
TransientDeliveryError marks a failure in which NO attempt was driven (endpoint resolve failure, a canceled prior-row poll): minting an attempt number would drift the counter past the real ledger row — parking an entry whose admitted row the sweeper can then never find (#1316 review round 2). Transient failures retry on their own bounded budget (MaxTransientFailures) without touching attempt numbering.
func (*TransientDeliveryError) Error ¶ added in v0.29.0
func (t *TransientDeliveryError) Error() string
func (*TransientDeliveryError) Unwrap ¶ added in v0.29.0
func (t *TransientDeliveryError) Unwrap() error
type Verdict ¶ added in v0.19.0
type Verdict int
Verdict is a verifier's decision about an ambiguous delivery attempt.
const ( // VerdictInconclusive: the transcript could not be read (agent // unreachable, page coverage incomplete) — recheck with backoff. VerdictInconclusive Verdict = iota // VerdictDelivered: the entry's text is present in the agent // transcript within the send window — the entry is complete. VerdictDelivered // VerdictAbsent: the transcript proves the message never persisted // — a definitive failure, safe to re-send. VerdictAbsent )
Source Files