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
- type AmbiguousError
- type DeliveredHook
- type Deliverer
- type Duplicate
- type Entry
- 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) bool
- 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) bool
- 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) SetOnDelivered(h DeliveredHook)
- func (s *Service) SetOnStaged(h StagedHook)
- func (s *Service) SetVerifier(v Verifier)
- 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 Verdict
- type Verifier
Constants ¶
const ( StatusPending = "pending" StatusDelivering = "delivering" StatusVerifying = "verifying" StatusError = "error" )
Status values.
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 )
Tunables (vars for tests).
var ErrCapped = errors.New("session outbox is full")
ErrCapped is returned when the session's outbox is at Cap.
Functions ¶
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 — both the synchronous 2xx path and the verified path. 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 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"`
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 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) 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.
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) 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) 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 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