fix(ae): remaining CodeRabbit Majors (#53 followup)

- controller: don't fan out Pixie rows when attribution Sink.Write fails
  (avoids orphaned rows; release in-flight slot, non-fatal). (controller.go:347)
- controller.Rehydrate: re-arm rev-1 pushPixieRows for restored windows so a
  restart doesn't silently miss post-restart Pixie data. (controller.go:255)
- passthrough.pull: per-table timeout context so a hung dependency can't stall
  the sweep / delay shutdown (covers serial+concurrent ticks). (passthrough.go:147)
- schema: TTL (30d) on ae_reconcile to cap append-only growth. (schema.sql:485)

Verified no-change-needed: adaptive_attribution ORDER BY (hostname, anomaly_hash)
is safe — anomaly_hash already encodes namespace+pod (anomaly/hash.go), so rows
never collapse across ns/pod. (schema.sql:430)
Already on ae-prod (no-op): filter timer leak, stats.py EXACT guard, sink async,
watermark paging, pixieapi WithDirectTLSSkipVerify, http.Server timeouts.

Author: pixie-agent

src/vizier/services/adaptive_export/internal/clickhouse/schema.sql

@@ -482,7 +482,10 @@ CREATE TABLE IF NOT EXISTS forensic_db.ae_reconcile (
     hostname    String
 ) ENGINE = MergeTree
   PARTITION BY toYYYYMMDD(ts)
-  ORDER BY (table_name, ts);
+  ORDER BY (table_name, ts)
+  -- append-only debug log; cap growth so long reconcile runs don't accumulate
+  -- unbounded storage (CodeRabbit). 30d matches the pixie observation tables.
+  TTL toDateTime(ts) + INTERVAL 30 DAY DELETE;
 
 -- dx_attack_graph — dx evidence-graph edge list: one row per directed hop of an
 -- investigation (delivery/egress/execution/exfil/pivot), read by the Pixie

src/vizier/services/adaptive_export/internal/controller/controller.go

@@ -254,12 +254,33 @@ func (c *Controller) Rehydrate(ctx context.Context) error {
 		return err
 	}
 	c.mu.Lock()
-	defer c.mu.Unlock()
+	var resume []sink.AttributionRow
 	for i := range rows {
 		row := rows[i]
 		c.active[row.AnomalyHash] = &row
+		// Rev-1: a restart restored the window but no pushPixieRows goroutine —
+		// without this, post-restart Pixie data is silently missed until another
+		// event for the same hash arrives (CodeRabbit). Re-arm the fan-out for
+		// each restored window, mirroring handle()'s spawn (in-flight guarded).
+		if c.querier != nil && len(c.cfg.PushPixieTables) > 0 && !c.inFlight[row.AnomalyHash] {
+			c.inFlight[row.AnomalyHash] = true
+			resume = append(resume, row)
+		}
+	}
+	c.mu.Unlock()
+	for i := range resume {
+		r := resume[i]
+		go func() {
+			defer func() {
+				c.mu.Lock()
+				delete(c.inFlight, r.AnomalyHash)
+				c.mu.Unlock()
+			}()
+			c.pushPixieRows(ctx, r)
+		}()
 	}
-	log.WithField("rehydrated", len(rows)).Info("controller: active set restored")
+	log.WithFields(log.Fields{"rehydrated": len(rows), "resumed": len(resume)}).
+		Info("controller: active set restored")
 	return nil
 }
 
@@ -331,7 +352,17 @@ func (c *Controller) handle(ctx context.Context, ev kubescape.Event) {
 	c.mu.Unlock()
 
 	if err := c.sink.Write(ctx, []sink.AttributionRow{snapshot}); err != nil {
-		log.WithError(err).Warn("controller: sink write failed")
+		// Attribution persistence failed → do NOT fan out, or we'd write Pixie
+		// rows with no persisted attribution anchor (orphaned rows, CodeRabbit).
+		// Non-fatal (system-stability rule): release the reserved in-flight slot
+		// and return; a later event for the same hash retries.
+		log.WithError(err).Warn("controller: sink write failed — skipping fan-out")
+		if spawn {
+			c.mu.Lock()
+			delete(c.inFlight, hash)
+			c.mu.Unlock()
+		}
+		return
 	}
 	if c.cfg.OnAttribution != nil {
 		c.cfg.OnAttribution(snapshot.Namespace, snapshot.Pod, snapshot.TEnd)

src/vizier/services/adaptive_export/internal/passthrough/passthrough.go

@@ -252,6 +252,11 @@ func (l *Loop) tickConcurrent(ctx context.Context, sliceStart, sliceEnd time.Tim
 // sink, and recorder are all pool/HTTP-backed and concurrency-safe, and
 // each call touches a different forensic_db.<table>.
 func (l *Loop) pull(ctx context.Context, table, src string, sliceStart, sliceEnd time.Time) {
+	// Bound this table's external query+write+record so a hung dependency can't
+	// stall the whole sweep or delay shutdown (CodeRabbit). Derived per-table
+	// from the parent ctx; covers both the serial and concurrent tick paths.
+	ctx, cancel := context.WithTimeout(ctx, l.cfg.Refresh)
+	defer cancel()
 	rows, err := l.q.Query(ctx, src)
 	if err != nil {
 		log.WithError(err).WithField("table", table).Warn("ADAPTIVE_PASSTHROUGH: pixie query failed")