feat(chat): productionize semantic summary shadow path without REPLACE

deploy/postgres/goose/00062_2026-09-22-chat-summary-shadow-provenance.sql

@@ -0,0 +1,40 @@
+-- +goose Up
+-- S1: persist shadow provenance on chat_summaries and durable shadow generation jobs.
+-- SEMANTIC_SUMMARY_REPLACE remains unwired; jobs only fill advisory shadow fields.
+
+ALTER TABLE chat_summaries
+    ADD COLUMN IF NOT EXISTS freshness_note TEXT NOT NULL DEFAULT '',
+    ADD COLUMN IF NOT EXISTS shadow_json JSONB;
+
+CREATE TABLE IF NOT EXISTS chat_summary_shadow_jobs (
+    id                      BIGSERIAL PRIMARY KEY,
+    session_id              TEXT NOT NULL,
+    summary_revision        INT NOT NULL,
+    input_digest            TEXT NOT NULL,
+    visibility_digest       TEXT NOT NULL,
+    watermark_to_message_id BIGINT NOT NULL DEFAULT 0,
+    prior_revision          INT NOT NULL DEFAULT 0,
+    payload                 JSONB NOT NULL DEFAULT '{}'::jsonb,
+    status                  TEXT NOT NULL DEFAULT 'pending',
+    attempt                 INT NOT NULL DEFAULT 0,
+    lease_owner             TEXT NOT NULL DEFAULT '',
+    lease_until             TIMESTAMPTZ,
+    last_error              TEXT NOT NULL DEFAULT '',
+    created_at              TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at              TIMESTAMPTZ NOT NULL DEFAULT now(),
+    CONSTRAINT chk_chat_summary_shadow_job_status
+        CHECK (status IN ('pending', 'leased', 'done', 'failed')),
+    CONSTRAINT uq_chat_summary_shadow_job
+        UNIQUE (session_id, summary_revision, input_digest)
+);
+
+CREATE INDEX IF NOT EXISTS idx_chat_summary_shadow_jobs_claim
+    ON chat_summary_shadow_jobs (status, updated_at, id)
+    WHERE status IN ('pending', 'leased');
+
+-- +goose Down
+DROP INDEX IF EXISTS idx_chat_summary_shadow_jobs_claim;
+DROP TABLE IF EXISTS chat_summary_shadow_jobs;
+ALTER TABLE chat_summaries
+    DROP COLUMN IF EXISTS shadow_json,
+    DROP COLUMN IF EXISTS freshness_note;

internal/ctxmgr/manager.go

@@ -34,11 +34,14 @@ type ContextManager struct {
 	digests            map[string]map[string]*DailyDigest // session_id -> day_key -> DailyDigest
 	readPositions      map[string]*ChatReadPosition       // session_id:user_id -> read position
 	turns              map[string]*ChatTurnRecord         // sessionID|clientTurnID -> 幂等回合（仅内存降级）
-	cache              *MultiLevelCache                   // 多级统一缓存系统 (C1-C5)
-	semanticSummarizer SemanticSummarizer                 // no-tool summarizer; nil → heuristic default
-	nextActivityID     int64
-	nextMessageID      int64
-	nextChunkID        int64
+	cache                *MultiLevelCache                   // 多级统一缓存系统 (C1-C5)
+	semanticSummarizer   SemanticSummarizer                 // no-tool summarizer; nil → heuristic default
+	shadowJobs           []SemanticShadowJob                // in-process durable fallback queue
+	shadowWorkerRunning  bool
+	shadowReclaimStarted bool
+	nextActivityID       int64
+	nextMessageID        int64
+	nextChunkID          int64
 }
 
 // NewContextManager 创建上下文管理器

internal/ctxmgr/semantic_summary.go

@@ -5,6 +5,7 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"fmt"
+	"sort"
 	"strings"
 	"time"
 	"unicode/utf8"
@@ -37,16 +38,24 @@ type MinimizedMessage struct {
 
 // SemanticSummaryRequest is the sole input to a no-tool summarizer.
 type SemanticSummaryRequest struct {
-	SessionID   string
-	Messages    []MinimizedMessage
-	InputDigest string
+	SessionID            string
+	Messages             []MinimizedMessage
+	InputDigest          string
+	VisibilityDigest     string
+	PriorNarrative       string
+	PriorFacts           map[string]string
+	WatermarkToMessageID int64
+	ExpectedRevision     int
 }
 
 // SemanticSummaryResult is the summarizer output (advisory under shadow).
 type SemanticSummaryResult struct {
-	Narrative string
-	Model     string
-	Protocol  string
+	Narrative            string
+	Model                string
+	Protocol             string
+	InputDigest          string // optional echo for server validation
+	VisibilityDigest     string // optional echo
+	WatermarkToMessageID int64  // optional echo
 }
 
 // SemanticShadowReport captures shadow evaluation without replacing deterministic narrative.
@@ -184,6 +193,99 @@ func DigestAuthorizedInput(msgs []MinimizedMessage) string {
 	return hex.EncodeToString(h.Sum(nil))
 }
 
+// DigestAuthorizedVisibility digests the authorized-visible message identity set
+// (id+role only). Used as chat_summaries.visibility_digest binding.
+func DigestAuthorizedVisibility(msgs []MinimizedMessage) string {
+	h := sha256.New()
+	for _, m := range msgs {
+		_, _ = fmt.Fprintf(h, "%d|%s\n", m.ID, m.Role)
+	}
+	return hex.EncodeToString(h.Sum(nil))
+}
+
+// MergePriorSummaryIntoInput folds the previous deterministic summary into the
+// semantic request so shadow generation does not drop earlier covered facts.
+func MergePriorSummaryIntoInput(msgs []MinimizedMessage, prior *ChatSummary) (merged []MinimizedMessage, priorNarrative string, priorFacts map[string]string) {
+	merged = msgs
+	if prior == nil {
+		return merged, "", nil
+	}
+	priorNarrative = strings.TrimSpace(prior.Narrative)
+	if len(prior.Facts) > 0 {
+		priorFacts = make(map[string]string, len(prior.Facts))
+		for k, v := range prior.Facts {
+			priorFacts[k] = v
+		}
+	}
+	return merged, priorNarrative, priorFacts
+}
+
+// DigestSemanticInput digests minimized messages plus merged prior summary material.
+func DigestSemanticInput(msgs []MinimizedMessage, priorNarrative string, priorFacts map[string]string) string {
+	h := sha256.New()
+	for _, m := range msgs {
+		_, _ = fmt.Fprintf(h, "%d|%s|%s\n", m.ID, m.Role, m.Content)
+	}
+	_, _ = fmt.Fprintf(h, "prior|%s\n", priorNarrative)
+	if len(priorFacts) > 0 {
+		keys := make([]string, 0, len(priorFacts))
+		for k := range priorFacts {
+			keys = append(keys, k)
+		}
+		sort.Strings(keys)
+		for _, k := range keys {
+			_, _ = fmt.Fprintf(h, "fact|%s=%s\n", k, priorFacts[k])
+		}
+	}
+	return hex.EncodeToString(h.Sum(nil))
+}
+
+// scheduleSemanticShadow attaches a pending advisory report and enqueues background work.
+// Served Narrative/Facts remain deterministic; REPLACE stays unwired.
+func scheduleSemanticShadow(ctx context.Context, m *ContextManager, sum *ChatSummary, msgs []ChatMessageRecord, prior *ChatSummary) {
+	if sum == nil || m == nil {
+		return
+	}
+	sum.Source = "deterministic"
+	if !SemanticSummaryShadowEnabled() {
+		return
+	}
+	now := time.Now()
+	minimized := MinimizeAuthorizedMessages(msgs)
+	merged, priorNarrative, priorFacts := MergePriorSummaryIntoInput(minimized, prior)
+	visDigest := DigestAuthorizedVisibility(merged)
+	if sum.VisibilityDigest == "" {
+		sum.VisibilityDigest = visDigest
+	}
+	inputDigest := DigestSemanticInput(merged, priorNarrative, priorFacts)
+	report := &SemanticShadowReport{
+		Status:          ShadowStatusPending,
+		InputDigest:     inputDigest,
+		ToolPermissions: SemanticToolPermissions,
+		GeneratedAt:     now,
+		FreshnessNote:   fmt.Sprintf("semantic_shadow_pending; deterministic served immediately at %s", now.UTC().Format(time.RFC3339)),
+	}
+	sum.Shadow = report
+	sum.FreshnessNote = report.FreshnessNote
+	job := SemanticShadowJob{
+		SessionID:            sum.SessionID,
+		SummaryRevision:      sum.Revision,
+		InputDigest:          inputDigest,
+		VisibilityDigest:     sum.VisibilityDigest,
+		WatermarkToMessageID: sum.ToMessageID,
+		PriorRevision:        sum.PreviousRevision,
+		Messages:             merged,
+		PriorNarrative:       priorNarrative,
+		PriorFacts:           priorFacts,
+	}
+	if err := m.EnqueueSemanticShadow(ctx, job); err != nil {
+		report.Status = ShadowStatusFallback
+		report.FreshnessNote = fmt.Sprintf("semantic_shadow_fallback: enqueue_failed; deterministic retained at %s", now.UTC().Format(time.RFC3339))
+		sum.Shadow = report
+		sum.FreshnessNote = report.FreshnessNote
+	}
+}
+
 func (m *ContextManager) SetSemanticSummarizer(s SemanticSummarizer) {
 	if m == nil {
 		return
@@ -206,7 +308,8 @@ func (m *ContextManager) semanticSummarizerOrDefault() SemanticSummarizer {
 	return &HeuristicSemanticSummarizer{}
 }
 
-// applySemanticShadow evaluates shadow semantic summary without replacing deterministic fields.
+// applySemanticShadow is retained for daily digest sync advisory evaluation.
+// Rolling chat summaries use scheduleSemanticShadow (background reliable task).
 // REPLACE is intentionally unwired: served narrative/facts stay deterministic.
 func applySemanticShadow(ctx context.Context, sum *ChatSummary, msgs []ChatMessageRecord, summarizer SemanticSummarizer) {
 	if sum == nil {
@@ -218,6 +321,9 @@ func applySemanticShadow(ctx context.Context, sum *ChatSummary, msgs []ChatMessa
 	}
 	now := time.Now()
 	minimized := MinimizeAuthorizedMessages(msgs)
+	if sum.VisibilityDigest == "" {
+		sum.VisibilityDigest = DigestAuthorizedVisibility(minimized)
+	}
 	digest := DigestAuthorizedInput(minimized)
 	report := &SemanticShadowReport{
 		Status:          ShadowStatusFallback,
@@ -232,9 +338,12 @@ func applySemanticShadow(ctx context.Context, sum *ChatSummary, msgs []ChatMessa
 		return
 	}
 	res, err := summarizer.Summarize(ctx, SemanticSummaryRequest{
-		SessionID:   sum.SessionID,
-		Messages:    minimized,
-		InputDigest: digest,
+		SessionID:            sum.SessionID,
+		Messages:             minimized,
+		InputDigest:          digest,
+		VisibilityDigest:     sum.VisibilityDigest,
+		WatermarkToMessageID: sum.ToMessageID,
+		ExpectedRevision:     sum.Revision,
 	})
 	if err != nil || res == nil || strings.TrimSpace(res.Narrative) == "" {
 		reason := "empty_or_error"
@@ -255,7 +364,6 @@ func applySemanticShadow(ctx context.Context, sum *ChatSummary, msgs []ChatMessa
 	}
 	// Explicitly do NOT swap sum.Narrative even if REPLACE env is true
 	// (Wave 5 X-2 sealed; see wave5-deadcode-decisions.md#x-2).
-	// Keep the call so envBool parsing stays covered; do not "fix" by wiring REPLACE.
 	_ = SemanticSummaryReplaceEnabled()
 	sum.Shadow = report
 }

internal/ctxmgr/shadow_task.go

@@ -0,0 +1,407 @@
+package ctxmgr
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"strings"
+	"time"
+)
+
+const (
+	ShadowStatusPending = "pending"
+	shadowJobMaxAttempt = 3
+	shadowLeaseTTL      = 2 * time.Minute
+)
+
+// SemanticShadowJob is a durable unit of work for advisory semantic generation.
+// It never replaces the served deterministic Narrative (REPLACE stays unwired).
+type SemanticShadowJob struct {
+	ID                   int64              `json:"id,omitempty"`
+	SessionID            string             `json:"session_id"`
+	SummaryRevision      int                `json:"summary_revision"`
+	InputDigest          string             `json:"input_digest"`
+	VisibilityDigest     string             `json:"visibility_digest"`
+	WatermarkToMessageID int64              `json:"watermark_to_message_id"`
+	PriorRevision        int                `json:"prior_revision"`
+	Messages             []MinimizedMessage `json:"messages"`
+	PriorNarrative       string             `json:"prior_narrative,omitempty"`
+	PriorFacts           map[string]string  `json:"prior_facts,omitempty"`
+	Status               string             `json:"status,omitempty"`
+	Attempt              int                `json:"attempt,omitempty"`
+}
+
+type shadowJobPayload struct {
+	Messages       []MinimizedMessage `json:"messages"`
+	PriorNarrative string             `json:"prior_narrative,omitempty"`
+	PriorFacts     map[string]string  `json:"prior_facts,omitempty"`
+}
+
+// EnqueueSemanticShadow schedules background semantic evaluation after the
+// deterministic summary is already persisted and returned to the caller.
+func (m *ContextManager) EnqueueSemanticShadow(ctx context.Context, job SemanticShadowJob) error {
+	if m == nil || job.SessionID == "" || job.InputDigest == "" {
+		return fmt.Errorf("invalid semantic shadow job")
+	}
+	if job.Status == "" {
+		job.Status = "pending"
+	}
+	if m.pool == nil {
+		m.rememberShadowJob(job)
+		m.kickShadowWorker()
+		return nil
+	}
+	payload, err := json.Marshal(shadowJobPayload{
+		Messages:       job.Messages,
+		PriorNarrative: job.PriorNarrative,
+		PriorFacts:     job.PriorFacts,
+	})
+	if err != nil {
+		return err
+	}
+	_, err = m.pool.Exec(ctx, `
+		INSERT INTO chat_summary_shadow_jobs (
+			session_id, summary_revision, input_digest, visibility_digest,
+			watermark_to_message_id, prior_revision, payload, status, attempt, updated_at
+		) VALUES ($1,$2,$3,$4,$5,$6,$7,'pending',0,now())
+		ON CONFLICT (session_id, summary_revision, input_digest) DO NOTHING
+	`, job.SessionID, job.SummaryRevision, job.InputDigest, job.VisibilityDigest,
+		job.WatermarkToMessageID, job.PriorRevision, payload)
+	if err != nil {
+		slog.Warn("semantic shadow job persist failed; using in-process queue", "session_id", job.SessionID, "err", err)
+		m.rememberShadowJob(job)
+	}
+	m.kickShadowWorker()
+	return nil
+}
+
+func (m *ContextManager) rememberShadowJob(job SemanticShadowJob) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.shadowJobs == nil {
+		m.shadowJobs = make([]SemanticShadowJob, 0, 4)
+	}
+	for _, existing := range m.shadowJobs {
+		if existing.SessionID == job.SessionID && existing.SummaryRevision == job.SummaryRevision && existing.InputDigest == job.InputDigest {
+			return
+		}
+	}
+	m.shadowJobs = append(m.shadowJobs, job)
+}
+
+func (m *ContextManager) kickShadowWorker() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.shadowWorkerRunning {
+		return
+	}
+	m.shadowWorkerRunning = true
+	go m.shadowWorkerLoop()
+}
+
+// StartSemanticShadowWorker starts a long-lived reclaim loop for durable jobs.
+func (m *ContextManager) StartSemanticShadowWorker(stop <-chan struct{}) {
+	if m == nil || m.pool == nil {
+		return
+	}
+	m.mu.Lock()
+	if m.shadowReclaimStarted {
+		m.mu.Unlock()
+		return
+	}
+	m.shadowReclaimStarted = true
+	m.mu.Unlock()
+	go func() {
+		ticker := time.NewTicker(5 * time.Second)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-stop:
+				return
+			case <-ticker.C:
+				m.kickShadowWorker()
+			}
+		}
+	}()
+	m.kickShadowWorker()
+}
+
+// DrainSemanticShadowTasks processes queued shadow jobs synchronously (tests / recovery).
+func (m *ContextManager) DrainSemanticShadowTasks(ctx context.Context) int {
+	if m == nil {
+		return 0
+	}
+	n := 0
+	for {
+		job, ok := m.claimNextShadowJob(ctx)
+		if !ok {
+			return n
+		}
+		m.runShadowJob(ctx, job)
+		n++
+	}
+}
+
+func (m *ContextManager) shadowWorkerLoop() {
+	defer func() {
+		m.mu.Lock()
+		m.shadowWorkerRunning = false
+		m.mu.Unlock()
+	}()
+	ctx := context.Background()
+	for {
+		job, ok := m.claimNextShadowJob(ctx)
+		if !ok {
+			return
+		}
+		m.runShadowJob(ctx, job)
+	}
+}
+
+func (m *ContextManager) claimNextShadowJob(ctx context.Context) (SemanticShadowJob, bool) {
+	if m.pool != nil {
+		if job, ok := m.claimDurableShadowJob(ctx); ok {
+			return job, true
+		}
+	}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if len(m.shadowJobs) == 0 {
+		return SemanticShadowJob{}, false
+	}
+	job := m.shadowJobs[0]
+	m.shadowJobs = m.shadowJobs[1:]
+	job.Status = "leased"
+	job.Attempt++
+	return job, true
+}
+
+func (m *ContextManager) claimDurableShadowJob(ctx context.Context) (SemanticShadowJob, bool) {
+	owner := fmt.Sprintf("shadow-%d", time.Now().UnixNano())
+	tx, err := m.pool.Begin(ctx)
+	if err != nil {
+		return SemanticShadowJob{}, false
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+
+	var (
+		job     SemanticShadowJob
+		payload []byte
+	)
+	err = tx.QueryRow(ctx, `
+		SELECT id, session_id, summary_revision, input_digest, visibility_digest,
+		       watermark_to_message_id, prior_revision, payload, attempt
+		FROM chat_summary_shadow_jobs
+		WHERE status IN ('pending','leased')
+		  AND (lease_until IS NULL OR lease_until < now())
+		ORDER BY updated_at, id
+		FOR UPDATE SKIP LOCKED
+		LIMIT 1
+	`).Scan(
+		&job.ID, &job.SessionID, &job.SummaryRevision, &job.InputDigest, &job.VisibilityDigest,
+		&job.WatermarkToMessageID, &job.PriorRevision, &payload, &job.Attempt,
+	)
+	if err != nil {
+		return SemanticShadowJob{}, false
+	}
+	job.Attempt++
+	var body shadowJobPayload
+	_ = json.Unmarshal(payload, &body)
+	job.Messages = body.Messages
+	job.PriorNarrative = body.PriorNarrative
+	job.PriorFacts = body.PriorFacts
+	job.Status = "leased"
+	_, err = tx.Exec(ctx, `
+		UPDATE chat_summary_shadow_jobs
+		SET status='leased', attempt=$2, lease_owner=$3, lease_until=now()+$4::interval, updated_at=now()
+		WHERE id=$1
+	`, job.ID, job.Attempt, owner, shadowLeaseTTL.String())
+	if err != nil {
+		return SemanticShadowJob{}, false
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return SemanticShadowJob{}, false
+	}
+	return job, true
+}
+
+func (m *ContextManager) runShadowJob(ctx context.Context, job SemanticShadowJob) {
+	report := &SemanticShadowReport{
+		Status:          ShadowStatusFallback,
+		InputDigest:     job.InputDigest,
+		ToolPermissions: SemanticToolPermissions,
+		GeneratedAt:     time.Now(),
+	}
+	summarizer := m.semanticSummarizerOrDefault()
+	res, err := summarizer.Summarize(ctx, SemanticSummaryRequest{
+		SessionID:            job.SessionID,
+		Messages:             job.Messages,
+		InputDigest:          job.InputDigest,
+		VisibilityDigest:     job.VisibilityDigest,
+		PriorNarrative:       job.PriorNarrative,
+		PriorFacts:           job.PriorFacts,
+		WatermarkToMessageID: job.WatermarkToMessageID,
+		ExpectedRevision:     job.SummaryRevision,
+	})
+	if err != nil || res == nil || strings.TrimSpace(res.Narrative) == "" {
+		reason := "empty_or_error"
+		if err != nil {
+			reason = "model_error"
+		}
+		report.FreshnessNote = fmt.Sprintf("semantic_shadow_fallback: %s; deterministic retained at %s", reason, time.Now().UTC().Format(time.RFC3339))
+		_ = m.persistShadowProvenance(ctx, job, report, err)
+		return
+	}
+	if verr := ValidateShadowResult(job, res); verr != nil {
+		report.FreshnessNote = fmt.Sprintf("semantic_shadow_fallback: validation_failed (%v); deterministic retained at %s", verr, time.Now().UTC().Format(time.RFC3339))
+		_ = m.persistShadowProvenance(ctx, job, report, verr)
+		return
+	}
+	if err := m.validateShadowAgainstCurrent(ctx, job); err != nil {
+		report.FreshnessNote = fmt.Sprintf("semantic_shadow_fallback: stale (%v); deterministic retained at %s", err, time.Now().UTC().Format(time.RFC3339))
+		_ = m.persistShadowProvenance(ctx, job, report, err)
+		return
+	}
+	report.Status = ShadowStatusOK
+	report.Narrative = res.Narrative
+	report.Model = res.Model
+	report.Protocol = res.Protocol
+	if report.Protocol == "" {
+		report.Protocol = semanticProtocolV1
+	}
+	// Explicitly do NOT swap Narrative even if REPLACE env is true.
+	_ = SemanticSummaryReplaceEnabled()
+	_ = m.persistShadowProvenance(ctx, job, report, nil)
+}
+
+func (m *ContextManager) validateShadowAgainstCurrent(ctx context.Context, job SemanticShadowJob) error {
+	current, err := m.GetChatSummary(ctx, job.SessionID)
+	if err != nil {
+		return err
+	}
+	if current == nil {
+		return fmt.Errorf("summary missing")
+	}
+	return ValidateShadowWatermark(job, current)
+}
+
+// ValidateShadowWatermark rejects shadow results that no longer match the served summary.
+func ValidateShadowWatermark(job SemanticShadowJob, current *ChatSummary) error {
+	if current == nil {
+		return fmt.Errorf("summary missing")
+	}
+	if current.Revision != job.SummaryRevision {
+		return fmt.Errorf("revision mismatch want %d got %d", job.SummaryRevision, current.Revision)
+	}
+	if current.VisibilityDigest != "" && job.VisibilityDigest != "" && current.VisibilityDigest != job.VisibilityDigest {
+		return fmt.Errorf("visibility digest mismatch")
+	}
+	if job.WatermarkToMessageID > 0 && current.ToMessageID > 0 && current.ToMessageID != job.WatermarkToMessageID {
+		return fmt.Errorf("watermark mismatch want %d got %d", job.WatermarkToMessageID, current.ToMessageID)
+	}
+	return nil
+}
+
+// ValidateShadowResult checks optional echoed digests/watermarks from the summarizer.
+func ValidateShadowResult(job SemanticShadowJob, res *SemanticSummaryResult) error {
+	if res == nil {
+		return fmt.Errorf("nil result")
+	}
+	if strings.TrimSpace(res.Narrative) == "" {
+		return fmt.Errorf("empty narrative")
+	}
+	if res.InputDigest != "" && res.InputDigest != job.InputDigest {
+		return fmt.Errorf("input digest mismatch")
+	}
+	if res.WatermarkToMessageID > 0 && job.WatermarkToMessageID > 0 && res.WatermarkToMessageID != job.WatermarkToMessageID {
+		return fmt.Errorf("result watermark mismatch")
+	}
+	if res.VisibilityDigest != "" && job.VisibilityDigest != "" && res.VisibilityDigest != job.VisibilityDigest {
+		return fmt.Errorf("result visibility digest mismatch")
+	}
+	return nil
+}
+
+func (m *ContextManager) persistShadowProvenance(ctx context.Context, job SemanticShadowJob, report *SemanticShadowReport, runErr error) error {
+	current, err := m.GetChatSummary(ctx, job.SessionID)
+	if err != nil {
+		m.finishShadowJob(ctx, job, runErr)
+		return err
+	}
+	if current == nil {
+		m.finishShadowJob(ctx, job, fmt.Errorf("summary missing"))
+		return fmt.Errorf("summary missing")
+	}
+	if err := ValidateShadowWatermark(job, current); err != nil {
+		m.finishShadowJob(ctx, job, err)
+		return err
+	}
+	current.Shadow = report
+	if report != nil && report.FreshnessNote != "" {
+		current.FreshnessNote = report.FreshnessNote
+	} else if report != nil && report.Status == ShadowStatusOK {
+		current.FreshnessNote = ""
+	}
+	if report != nil && report.Model != "" {
+		current.ModelVersion = report.Model
+	}
+	if report != nil && report.Protocol != "" {
+		current.ProtocolVersion = report.Protocol
+	}
+	current.Source = "deterministic"
+	saveErr := m.SaveChatSummary(ctx, current)
+	finishErr := runErr
+	if saveErr != nil && finishErr == nil {
+		finishErr = saveErr
+	}
+	m.finishShadowJob(ctx, job, finishErr)
+	return saveErr
+}
+
+func (m *ContextManager) finishShadowJob(ctx context.Context, job SemanticShadowJob, runErr error) {
+	if m.pool == nil || job.ID == 0 {
+		return
+	}
+	status := "done"
+	errText := ""
+	if runErr != nil {
+		errText = runErr.Error()
+		if job.Attempt >= shadowJobMaxAttempt {
+			status = "failed"
+		} else {
+			status = "pending"
+		}
+	}
+	_, _ = m.pool.Exec(ctx, `
+		UPDATE chat_summary_shadow_jobs
+		SET status=$2, last_error=$3, lease_owner='', lease_until=NULL, updated_at=now()
+		WHERE id=$1
+	`, job.ID, status, errText)
+}
+
+// PendingShadowJobCount returns in-process queued jobs (tests).
+func (m *ContextManager) PendingShadowJobCount() int {
+	if m == nil {
+		return 0
+	}
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return len(m.shadowJobs)
+}
+
+// WaitShadowIdle waits until the in-process worker is idle (tests).
+func (m *ContextManager) WaitShadowIdle(timeout time.Duration) bool {
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		m.mu.RLock()
+		running := m.shadowWorkerRunning
+		pending := len(m.shadowJobs)
+		m.mu.RUnlock()
+		if !running && pending == 0 {
+			return true
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	return false
+}

internal/ctxmgr/summarizer.go

@@ -124,12 +124,19 @@ func (m *ContextManager) SaveChatSummary(ctx context.Context, sum *ChatSummary)
 	if err != nil {
 		return fmt.Errorf("chat summary questions: %w", err)
 	}
+	var shadowJSON []byte
+	if sum.Shadow != nil {
+		shadowJSON, err = json.Marshal(sum.Shadow)
+		if err != nil {
+			return fmt.Errorf("chat summary shadow: %w", err)
+		}
+	}
 	_, err = m.pool.Exec(ctx, `
 		INSERT INTO chat_summaries (
 			session_id, narrative, facts, open_questions, covered_count, covered_until, updated_at,
 			source, visibility_digest, from_message_id, to_message_id, previous_revision, revision,
-			protocol_version, model_version
-		) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)
+			protocol_version, model_version, freshness_note, shadow_json
+		) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)
 		ON CONFLICT (session_id) DO UPDATE SET
 			narrative = EXCLUDED.narrative,
 			facts = EXCLUDED.facts,
@@ -144,10 +151,12 @@ func (m *ContextManager) SaveChatSummary(ctx context.Context, sum *ChatSummary)
 			previous_revision = EXCLUDED.previous_revision,
 			revision = EXCLUDED.revision,
 			protocol_version = EXCLUDED.protocol_version,
-			model_version = EXCLUDED.model_version
+			model_version = EXCLUDED.model_version,
+			freshness_note = EXCLUDED.freshness_note,
+			shadow_json = EXCLUDED.shadow_json
 	`, sum.SessionID, sum.Narrative, factsJSON, openQuestionsJSON, sum.CoveredCount, sum.CoveredUntil, sum.UpdatedAt,
 		sum.Source, sum.VisibilityDigest, nullSummaryID(sum.FromMessageID), nullSummaryID(sum.ToMessageID),
-		sum.PreviousRevision, sum.Revision, sum.ProtocolVersion, sum.ModelVersion)
+		sum.PreviousRevision, sum.Revision, sum.ProtocolVersion, sum.ModelVersion, sum.FreshnessNote, nullShadowJSON(shadowJSON))
 	if err != nil {
 		sum.Persisted = false
 		noteSummaryNotDurable(sum)
@@ -160,7 +169,54 @@ func (m *ContextManager) SaveChatSummary(ctx context.Context, sum *ChatSummary)
 	return nil
 }
 
+func nullShadowJSON(b []byte) any {
+	if len(b) == 0 {
+		return nil
+	}
+	return b
+}
+
 func (m *ContextManager) loadChatSummary(ctx context.Context, sessionID string) (*ChatSummary, error) {
+	var sum ChatSummary
+	var factsJSON, openQuestionsJSON, shadowJSON []byte
+	var fromID, toID *int64
+	err := m.pool.QueryRow(ctx, `
+		SELECT session_id, narrative, facts, open_questions, covered_count, covered_until, updated_at,
+		       source, visibility_digest, from_message_id, to_message_id, previous_revision, revision,
+		       protocol_version, model_version, freshness_note, shadow_json
+		FROM chat_summaries
+		WHERE session_id = $1
+	`, sessionID).Scan(
+		&sum.SessionID, &sum.Narrative, &factsJSON, &openQuestionsJSON, &sum.CoveredCount, &sum.CoveredUntil, &sum.UpdatedAt,
+		&sum.Source, &sum.VisibilityDigest, &fromID, &toID, &sum.PreviousRevision, &sum.Revision,
+		&sum.ProtocolVersion, &sum.ModelVersion, &sum.FreshnessNote, &shadowJSON,
+	)
+	if err != nil {
+		// Compatibility: older schemas without provenance columns.
+		if isUndefinedColumn(err) {
+			return m.loadChatSummaryLegacy(ctx, sessionID)
+		}
+		return nil, err
+	}
+	_ = json.Unmarshal(factsJSON, &sum.Facts)
+	_ = json.Unmarshal(openQuestionsJSON, &sum.OpenQuestions)
+	if len(shadowJSON) > 0 {
+		var report SemanticShadowReport
+		if json.Unmarshal(shadowJSON, &report) == nil {
+			sum.Shadow = &report
+		}
+	}
+	if fromID != nil {
+		sum.FromMessageID = *fromID
+	}
+	if toID != nil {
+		sum.ToMessageID = *toID
+	}
+	sum.Persisted = true
+	return &sum, nil
+}
+
+func (m *ContextManager) loadChatSummaryLegacy(ctx context.Context, sessionID string) (*ChatSummary, error) {
 	var sum ChatSummary
 	var factsJSON, openQuestionsJSON []byte
 	var fromID, toID *int64
@@ -190,6 +246,14 @@ func (m *ContextManager) loadChatSummary(ctx context.Context, sessionID string)
 	return &sum, nil
 }
 
+func isUndefinedColumn(err error) bool {
+	if err == nil {
+		return false
+	}
+	msg := strings.ToLower(err.Error())
+	return strings.Contains(msg, "freshness_note") || strings.Contains(msg, "shadow_json") || strings.Contains(msg, "undefined_column")
+}
+
 func (m *ContextManager) rememberSummary(sum *ChatSummary) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -296,9 +360,13 @@ func (m *ContextManager) GenerateRollingSummary(ctx context.Context, sessionID s
 		newSum.Revision = 1
 	}
 
-	// SEMANTIC_SUMMARY_SHADOW：仅在授权可见集合上评估语义摘要；失败保留 deterministic + freshness。
+	// Always bind the authorized-visible set digest (even when shadow is off).
+	minimized := MinimizeAuthorizedMessages(earlyMsgs)
+	newSum.VisibilityDigest = DigestAuthorizedVisibility(minimized)
+
+	// SEMANTIC_SUMMARY_SHADOW：后台可靠任务生成 advisory 语义摘要；请求路径立即返回确定性结果。
 	// SEMANTIC_SUMMARY_REPLACE 故意不接线，served Narrative 永不被语义结果替换。
-	applySemanticShadow(ctx, newSum, earlyMsgs, m.semanticSummarizerOrDefault())
+	scheduleSemanticShadow(ctx, m, newSum, earlyMsgs, oldSum)
 
 	// 写库失败仍返回确定性摘要，供本轮使用；错误留在 Persisted/FreshnessNote，不伪装成已持久化。
 	_ = m.SaveChatSummary(ctx, newSum)