FLEX-WP-0031-T03: durable decision outbox and the FLEX-DEC-2026-018 release rule

internal/emission commits one audit-core-shaped event per decision with
fsync before release (one sync per batch), random event ids with the
decision id as correlation_id, torn-tail and failed-commit truncation, and
per-class committed/released_uncommitted counts at GET /v1/emission.
The engine releases restrictions whose record failed to commit and
withholds allow/audit_only (503). api.DecisionEffects() is pinned by a
source-parsing test and cadence.yaml must classify exactly it.

T03 split under the task budget: heartbeat+drain is T05, reconciliation,
profile check and PVC are T06. Nothing deployed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 307130@bnt-lap001
Assistant-Session: 270c79f7-0823-4b0d-990d-aad5af9935ce

internal/emission/outbox.go

@@ -0,0 +1,268 @@
+// Package emission delivers the decision-record emission guarantee declared in
+// cadence.yaml (FLEX-WP-0031). A decision is released only after its event is
+// durably committed to a local outbox (FLEX-DEC-2026-018); draining to
+// audit-core is asynchronous.
+package emission
+
+import (
+	"bytes"
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+
+	"github.com/netkingdom/flex-auth/pkg/api"
+)
+
+// ClassPrefix prefixes every decision event class; the suffix is the effect.
+const ClassPrefix = "flex-auth.decision."
+
+// PlatformTenant is sent when a decision names no tenant, because audit-core
+// requires one on every event.
+const PlatformTenant = "tenant:platform"
+
+const outboxFile = "outbox.jsonl"
+
+// Class returns the event class for a decision effect.
+func Class(effect api.DecisionEffect) string {
+	return ClassPrefix + string(effect)
+}
+
+// Event is one audit-core.event envelope: exactly the fields audit-core's
+// normalize() requires. Data carries the decision record itself.
+type Event struct {
+	ID            string `json:"id"`
+	Type          string `json:"type"`
+	Source        string `json:"source"`
+	Subject       string `json:"subject"`
+	Tenant        string `json:"tenant"`
+	CorrelationID string `json:"correlation_id"`
+	OccurredAt    string `json:"occurred_at"`
+	Data          any    `json:"data"`
+}
+
+// entry is one committed outbox line.
+type entry struct {
+	Seq         uint64 `json:"seq"`
+	CommittedAt string `json:"committed_at"`
+	Event       Event  `json:"event"`
+}
+
+// Status is the local half of reconciliation: what this outbox committed, per
+// class, and what it released without committing.
+type Status struct {
+	Source              string            `json:"source"`
+	Committed           map[string]uint64 `json:"committed"`
+	ReleasedUncommitted map[string]uint64 `json:"released_uncommitted"`
+	// ReleasedUncommitted is held in memory: a restart during an outbox outage
+	// loses it (FLEX-DEC-2026-018, consequences).
+	Means string `json:"means"`
+}
+
+// Outbox is a durable, append-only, fsync-per-commit event log.
+type Outbox struct {
+	source string
+	path   string
+	clock  func() time.Time
+
+	mu          sync.Mutex
+	file        *os.File
+	seq         uint64
+	size        int64 // bytes committed; a failed commit is cut back to it
+	committed   map[string]uint64
+	uncommitted map[string]uint64
+}
+
+// Open opens or creates the outbox in dir and rebuilds committed counts from it.
+func Open(dir, source string) (*Outbox, error) {
+	if source == "" {
+		return nil, errors.New("emission source is required")
+	}
+	if err := os.MkdirAll(dir, 0o750); err != nil {
+		return nil, fmt.Errorf("create outbox directory: %w", err)
+	}
+	o := &Outbox{
+		source:      source,
+		path:        filepath.Join(dir, outboxFile),
+		clock:       time.Now,
+		committed:   map[string]uint64{},
+		uncommitted: map[string]uint64{},
+	}
+	if err := o.load(); err != nil {
+		return nil, err
+	}
+	file, err := os.OpenFile(o.path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o640)
+	if err != nil {
+		return nil, fmt.Errorf("open outbox: %w", err)
+	}
+	o.file = file
+	return o, nil
+}
+
+func (o *Outbox) load() error {
+	data, err := os.ReadFile(o.path)
+	if errors.Is(err, os.ErrNotExist) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("read outbox: %w", err)
+	}
+	good := 0
+	for line := 1; good < len(data); line++ {
+		end := bytes.IndexByte(data[good:], '\n')
+		if end < 0 {
+			// A torn final line is a write whose sync never returned, so its
+			// decision was never released as committed. Drop it so the next
+			// append does not bury it mid-file.
+			o.size = int64(good)
+			return os.Truncate(o.path, o.size)
+		}
+		var e entry
+		if err := json.Unmarshal(data[good:good+end], &e); err != nil {
+			return fmt.Errorf("outbox line %d is corrupt: %w", line, err)
+		}
+		o.seq = e.Seq
+		o.committed[e.Event.Type]++
+		good += end + 1
+	}
+	o.size = int64(good)
+	return nil
+}
+
+// SetClock overrides the commit clock. Tests use it.
+func (o *Outbox) SetClock(clock func() time.Time) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	o.clock = clock
+}
+
+// Close closes the outbox file.
+func (o *Outbox) Close() error {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	return o.file.Close()
+}
+
+// Append commits one decision; it satisfies decision.DecisionRecorder.
+func (o *Outbox) Append(decision api.DecisionEnvelope) error {
+	return o.AppendBatch([]api.DecisionEnvelope{decision})
+}
+
+// AppendBatch commits decisions with one sync, so a batch check pays the
+// durable-write cost once rather than per resource.
+func (o *Outbox) AppendBatch(decisions []api.DecisionEnvelope) error {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	now := o.clock().UTC().Format(time.RFC3339Nano)
+	var buf []byte
+	seq := o.seq
+	for _, decision := range decisions {
+		id, err := eventID()
+		if err != nil {
+			return err
+		}
+		seq++
+		line, err := json.Marshal(entry{Seq: seq, CommittedAt: now, Event: o.event(id, now, decision)})
+		if err != nil {
+			return fmt.Errorf("marshal outbox entry: %w", err)
+		}
+		buf = append(append(buf, line...), '\n')
+	}
+	if _, err := o.file.Write(buf); err != nil {
+		return o.rollback(fmt.Errorf("write outbox: %w", err))
+	}
+	if err := o.file.Sync(); err != nil {
+		return o.rollback(fmt.Errorf("sync outbox: %w", err))
+	}
+	o.seq = seq
+	o.size += int64(len(buf))
+	for _, decision := range decisions {
+		o.committed[Class(decision.Effect)]++
+	}
+	return nil
+}
+
+// rollback cuts a failed commit back to the last committed size, so a partial
+// line never sits in front of the next commit. The commit error is returned
+// either way; a failed cut is reported alongside it.
+func (o *Outbox) rollback(cause error) error {
+	if err := o.file.Truncate(o.size); err != nil {
+		return errors.Join(cause, fmt.Errorf("truncate outbox after failed commit: %w", err))
+	}
+	return cause
+}
+
+// NoteReleasedUncommitted counts a restriction released after its commit
+// failed. Every non-zero count is a finding.
+func (o *Outbox) NoteReleasedUncommitted(effect api.DecisionEffect) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	o.uncommitted[Class(effect)]++
+}
+
+// Status reports committed and released-uncommitted counts per class.
+func (o *Outbox) Status() Status {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	status := Status{
+		Source:              o.source,
+		Committed:           map[string]uint64{},
+		ReleasedUncommitted: map[string]uint64{},
+		Means: "committed counts this outbox holds since it was created; " +
+			"released_uncommitted is decisions released without a record " +
+			"since this process started, and any non-zero value is a finding",
+	}
+	for _, effect := range api.DecisionEffects() {
+		class := Class(effect)
+		status.Committed[class] = o.committed[class]
+		status.ReleasedUncommitted[class] = o.uncommitted[class]
+	}
+	return status
+}
+
+func (o *Outbox) event(id, at string, decision api.DecisionEnvelope) Event {
+	tenant := decision.Resource.Tenant
+	if decision.Binding != nil && decision.Binding.Tenant != "" {
+		tenant = decision.Binding.Tenant
+	}
+	if tenant == "" {
+		tenant = PlatformTenant
+	}
+	return Event{
+		ID:            id,
+		Type:          Class(decision.Effect),
+		Source:        o.source,
+		Subject:       resourceSubject(decision.Resource),
+		Tenant:        tenant,
+		CorrelationID: decision.ID,
+		OccurredAt:    at,
+		Data:          decision,
+	}
+}
+
+func resourceSubject(resource api.ResourceRef) string {
+	subject := resource.ID
+	if resource.Type != "" {
+		subject = resource.Type + "/" + subject
+	}
+	if resource.System != "" {
+		subject = resource.System + "/" + subject
+	}
+	return subject
+}
+
+// eventID is random, not derived from the decision: decision ids are content
+// digests, so two identical checks share one, and audit-core's idempotency key
+// would merge two real decisions into one.
+func eventID() (string, error) {
+	var b [16]byte
+	if _, err := rand.Read(b[:]); err != nil {
+		return "", fmt.Errorf("event id: %w", err)
+	}
+	return "flex-auth-evt:" + hex.EncodeToString(b[:]), nil
+}