Implement local decision delivery and reconcile blocked workplan requirements

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a1046e-fcad-7ea3-bf36-361fc8d679e8

cmd/flex-auth/main.go

@@ -12,6 +12,7 @@ import (
 	"net/http"
 	"os"
 	"strings"
+	"time"
 
 	"gopkg.in/yaml.v3"
 
@@ -329,6 +330,8 @@ func runServe(args []string, stdout, stderr io.Writer) int {
 	logPath := fs.String("log", "", "optional JSONL decision log path")
 	outboxDir := fs.String("outbox-dir", "", "durable emission outbox directory (FLEX-WP-0031); exclusive with --log")
 	emissionSource := fs.String("emission-source", "", "audit-core source name for emitted decision events, e.g. flex-auth.tenant-engine")
+	auditURL := fs.String("audit-url", "", "admitted audit-core /v1/events endpoint; empty disables drain and heartbeats")
+	auditTokenFile := fs.String("audit-token-file", "", "mounted sender token file, never a token value")
 	callerAuthMode := fs.String("caller-auth-mode", "disabled", "disabled, warn, or enforce")
 	callerAudience := fs.String("caller-audience", "flex-auth", "required caller token audience")
 	callerKubernetesURL := fs.String("caller-kubernetes-url", "https://kubernetes.default.svc", "Kubernetes API base URL for TokenReview")
@@ -348,6 +351,17 @@ func runServe(args []string, stdout, stderr io.Writer) int {
 		fmt.Fprintln(stderr, "serve --outbox-dir requires --emission-source and excludes --log")
 		return 64
 	}
+	delivery := emission.DeliveryConfig{URL: *auditURL, TokenFile: *auditTokenFile}
+	if *auditURL != "" || *auditTokenFile != "" {
+		if *outboxDir == "" {
+			fmt.Fprintln(stderr, "audit delivery requires --outbox-dir")
+			return 64
+		}
+		if err := delivery.Validate(); err != nil {
+			fmt.Fprintln(stderr, err)
+			return 64
+		}
+	}
 
 	engine, err := buildEngine(context.Background(), *registryPath, *policyPath, *logPath)
 	if err != nil {
@@ -380,6 +394,15 @@ func runServe(args []string, stdout, stderr io.Writer) int {
 	if outbox != nil {
 		mountEmission(mux, outbox)
 	}
+	if *auditURL != "" {
+		ctx, cancel := context.WithCancel(context.Background())
+		done := make(chan struct{})
+		go func() {
+			defer close(done)
+			runEmissionDelivery(ctx, outbox, delivery, stderr)
+		}()
+		defer func() { cancel(); <-done }()
+	}
 
 	fmt.Fprintf(stderr, "flex-auth serving on http://%s\n", *addr)
 	if err := http.ListenAndServe(*addr, mux); err != nil {
@@ -388,6 +411,31 @@ func runServe(args []string, stdout, stderr io.Writer) int {
 	return 0
 }
 
+func runEmissionDelivery(ctx context.Context, outbox *emission.Outbox, config emission.DeliveryConfig, stderr io.Writer) {
+	ticker := time.NewTicker(time.Minute)
+	defer ticker.Stop()
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+		if err := outbox.AppendDueHeartbeats(); err != nil {
+			fmt.Fprintf(stderr, "emission heartbeat commit: %v\n", err)
+		}
+		n, err := outbox.Drain(ctx, config, 100)
+		if err != nil && ctx.Err() == nil {
+			fmt.Fprintf(stderr, "emission drain: %v\n", err)
+		}
+		if err == nil && n == 100 {
+			continue // Drain the next bounded batch without imposing a rate cap.
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+		}
+	}
+}
+
 func newServeMux(engine *decisioncore.Engine) *http.ServeMux {
 	return newServeMuxWithCallerAuth(engine, callerauth.Disabled())
 }