From 056e4e5c4b732593300833fd1d414487559cf505 Mon Sep 17 00:00:00 2001 From: WeiCHEN Date: Tue, 22 Sep 2026 21:51:20 +0800 Subject: [PATCH 1/3] feat(chat): productionize semantic summary shadow path without REPLACE --- ...6-09-22-chat-summary-shadow-provenance.sql | 40 ++ internal/ctxmgr/manager.go | 13 +- internal/ctxmgr/semantic_summary.go | 130 +++++- internal/ctxmgr/shadow_task.go | 407 ++++++++++++++++++ internal/ctxmgr/summarizer.go | 80 +++- 5 files changed, 648 insertions(+), 22 deletions(-) create mode 100644 deploy/postgres/goose/00062_2026-09-22-chat-summary-shadow-provenance.sql create mode 100644 internal/ctxmgr/shadow_task.go diff --git a/deploy/postgres/goose/00062_2026-09-22-chat-summary-shadow-provenance.sql b/deploy/postgres/goose/00062_2026-09-22-chat-summary-shadow-provenance.sql new file mode 100644 index 00000000..d0971ed5 --- /dev/null +++ b/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; diff --git a/internal/ctxmgr/manager.go b/internal/ctxmgr/manager.go index 864ec7a3..dfaf010b 100644 --- a/internal/ctxmgr/manager.go +++ b/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 创建上下文管理器 diff --git a/internal/ctxmgr/semantic_summary.go b/internal/ctxmgr/semantic_summary.go index de8d72ab..5540eafa 100644 --- a/internal/ctxmgr/semantic_summary.go +++ b/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 } diff --git a/internal/ctxmgr/shadow_task.go b/internal/ctxmgr/shadow_task.go new file mode 100644 index 00000000..193b39a3 --- /dev/null +++ b/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 +} diff --git a/internal/ctxmgr/summarizer.go b/internal/ctxmgr/summarizer.go index 4fdc7e7f..777976ac 100644 --- a/internal/ctxmgr/summarizer.go +++ b/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) From 0c572ddaca117c1033251ce8e33da90f342334f6 Mon Sep 17 00:00:00 2001 From: WeiCHEN Date: Tue, 22 Sep 2026 21:59:12 +0800 Subject: [PATCH 2/3] test(chat): align semantic shadow tests with async pending path Drain/wait after GenerateRollingSummary; register summary table writers; document consolidation discarding duplicate S1 dirty worktrees. --- ...26-09-22-track-s1-summary-consolidation.md | 42 +++++++++++++++ .doc/system-design/module-boundaries.yaml | 6 +++ internal/ctxmgr/semantic_summary_test.go | 52 +++++++++++++++---- 3 files changed, 89 insertions(+), 11 deletions(-) create mode 100644 .doc/ops/short-term/2026-09-22-track-s1-summary-consolidation.md diff --git a/.doc/ops/short-term/2026-09-22-track-s1-summary-consolidation.md b/.doc/ops/short-term/2026-09-22-track-s1-summary-consolidation.md new file mode 100644 index 00000000..160bde6a --- /dev/null +++ b/.doc/ops/short-term/2026-09-22-track-s1-summary-consolidation.md @@ -0,0 +1,42 @@ +--- +title: "W3-S1 semantic summary：三 worktree 收敛" +task_id: "track-s1-summary-shadow-2026-09-22" +status: done +created: 2026-09-22 +updated: 2026-09-22 +owner: Cursor +task_type: implementation +task_tier: small +branch: feature/track-s1-summary-shadow-2026-09-22 +worktree: .worktrees/track-s1-summary-shadow-2026-09-22 +write_scope: [".doc/ops/short-term/**", ".doc/system-design/module-boundaries.yaml", "internal/ctxmgr/**", "deploy/postgres/goose/**"] +remote_sync: not_requested +integration_state: pending +--- + +# W3-S1 收敛结论 + +## Keeper + +- **唯一实现分支**：`feature/track-s1-summary-shadow-2026-09-22` +- **feat tip**：`056e4e5c` `feat(chat): productionize semantic summary shadow path without REPLACE` +- 覆盖：durable `chat_summary_shadow_jobs`、`shadow_task.go` 后台租约、watermark 校验、served Narrative 保持确定性、REPLACE 不接线。 + +## 丢弃（duplicate dirty WIP） + +| worktree | 原因 | +|---|---| +| `track-s1-semantic-summary-2026-09-22` | ahead=0；未完成的另一套 `semantic_shadow_jobs.go` / 内存队列设计；与 shadow feat 重叠且不可编译进同一契约 | +| `track-s1-summary-prod-2026-09-22` | ahead=0；并行草稿（`semantic_shadow_worker.go` / `summary_watermark.go` / 双 00062 迁移）；核心能力已在 shadow feat | + +丢弃动作：`git reset --hard` + `git clean -fd`，工作树回到与本地 `dev` 同 tip、无脏文件。 + +## 从 duplicate 回收的独有价值 + +- `module-boundaries.yaml`：补登记 `chat_summaries` / `chat_summary_shadow_jobs` / `chat_daily_digests` 写者（原仅在 dirty WIP）。 +- 测试：将同步期望改为 `pending → DrainSemanticShadowTasks / WaitShadowIdle → ok|fallback`，对齐异步生产路径。 + +## 非目标 + +- 不 push；不合入 `dev`（由既有 `integrate-track-s1-summary-shadow-2026-09-22` 串行认领）。 +- 不采纳 PROD 的 `input_digest` 列独立迁移（shadow 已把 digest 放在 job 行与 report JSON）。 diff --git a/.doc/system-design/module-boundaries.yaml b/.doc/system-design/module-boundaries.yaml index ab8e9d2b..cc51b053 100644 --- a/.doc/system-design/module-boundaries.yaml +++ b/.doc/system-design/module-boundaries.yaml @@ -298,6 +298,12 @@ table_owners: - ctxmgr chat_turns: - ctxmgr + chat_summaries: + - ctxmgr + chat_summary_shadow_jobs: + - ctxmgr + chat_daily_digests: + - ctxmgr freeform_documents: - freedoc task_runs: diff --git a/internal/ctxmgr/semantic_summary_test.go b/internal/ctxmgr/semantic_summary_test.go index f7e7fdfd..6816b166 100644 --- a/internal/ctxmgr/semantic_summary_test.go +++ b/internal/ctxmgr/semantic_summary_test.go @@ -56,8 +56,28 @@ func TestSemanticSummary_ShadowOKKeepsDeterministic(t *testing.T) { if err != nil { t.Fatal(err) } + if sum.Shadow == nil { + t.Fatal("expected advisory shadow attachment") + } + if strings.Contains(sum.Narrative, "语义影子叙事") { + t.Fatalf("REPLACE must stay unwired; narrative replaced: %s", sum.Narrative) + } + if sum.Facts["device_type"] != "超声手术设备" { + t.Fatalf("deterministic facts lost: %+v", sum.Facts) + } + // Production path is async: instant response is pending; drain / wait for OK. + if sum.Shadow.Status == ShadowStatusPending { + _ = mgr.DrainSemanticShadowTasks(ctx) + if !mgr.WaitShadowIdle(2 * time.Second) { + t.Fatal("shadow worker did not become idle") + } + sum, err = mgr.GetChatSummary(ctx, "sess_shadow_ok") + if err != nil || sum == nil { + t.Fatalf("get after drain: %v %+v", err, sum) + } + } if sum.Shadow == nil || sum.Shadow.Status != ShadowStatusOK { - t.Fatalf("expected shadow ok, got %+v", sum.Shadow) + t.Fatalf("expected shadow ok after drain, got %+v", sum.Shadow) } if sum.Shadow.Narrative != "语义影子叙事：超声刀立项" { t.Fatalf("shadow narrative mismatch: %q", sum.Shadow.Narrative) @@ -69,13 +89,10 @@ func TestSemanticSummary_ShadowOKKeepsDeterministic(t *testing.T) { t.Fatal("missing input digest") } if strings.Contains(sum.Narrative, "语义影子叙事") { - t.Fatalf("REPLACE must stay unwired; narrative replaced: %s", sum.Narrative) - } - if sum.Facts["device_type"] != "超声手术设备" { - t.Fatalf("deterministic facts lost: %+v", sum.Facts) + t.Fatalf("REPLACE must stay unwired after async attach: %s", sum.Narrative) } if sum.FreshnessNote != "" { - t.Fatalf("ok path should not set freshness fallback note: %q", sum.FreshnessNote) + t.Fatalf("ok path should clear freshness note: %q", sum.FreshnessNote) } } @@ -94,11 +111,8 @@ func TestSemanticSummary_ShadowFailureKeepsDeterministicWithFreshness(t *testing if err != nil { t.Fatal(err) } - if sum.Shadow == nil || sum.Shadow.Status != ShadowStatusFallback { - t.Fatalf("expected fallback shadow, got %+v", sum.Shadow) - } - if sum.FreshnessNote == "" || !strings.Contains(sum.FreshnessNote, "deterministic retained") { - t.Fatalf("expected freshness note, got %q", sum.FreshnessNote) + if sum.Shadow == nil { + t.Fatal("expected advisory shadow attachment") } if sum.Narrative == "" { t.Fatal("deterministic narrative must be retained") @@ -106,6 +120,22 @@ func TestSemanticSummary_ShadowFailureKeepsDeterministicWithFreshness(t *testing if strings.Contains(sum.Narrative, "model timeout") { t.Fatal("error text must not pollute narrative") } + if sum.Shadow.Status == ShadowStatusPending { + _ = mgr.DrainSemanticShadowTasks(ctx) + if !mgr.WaitShadowIdle(2 * time.Second) { + t.Fatal("shadow worker did not become idle") + } + sum, err = mgr.GetChatSummary(ctx, "sess_shadow_fail") + if err != nil || sum == nil { + t.Fatalf("get after drain: %v %+v", err, sum) + } + } + if sum.Shadow == nil || sum.Shadow.Status != ShadowStatusFallback { + t.Fatalf("expected fallback shadow after drain, got %+v", sum.Shadow) + } + if sum.FreshnessNote == "" || !strings.Contains(sum.FreshnessNote, "deterministic retained") { + t.Fatalf("expected freshness note, got %q", sum.FreshnessNote) + } } func TestSemanticSummary_MinimizeDropsMetadataAndCaps(t *testing.T) { From 3dc7b85b7ea16dd3f6e29b7037ea2204496f77c0 Mon Sep 17 00:00:00 2001 From: WeiCHEN Date: Tue, 22 Sep 2026 22:08:39 +0800 Subject: [PATCH 3/3] docs(ops): pin track-s1 consolidation head_sha to tip --- .../short-term/2026-09-22-track-s1-summary-consolidation.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.doc/ops/short-term/2026-09-22-track-s1-summary-consolidation.md b/.doc/ops/short-term/2026-09-22-track-s1-summary-consolidation.md index 160bde6a..cf20966b 100644 --- a/.doc/ops/short-term/2026-09-22-track-s1-summary-consolidation.md +++ b/.doc/ops/short-term/2026-09-22-track-s1-summary-consolidation.md @@ -11,7 +11,10 @@ branch: feature/track-s1-summary-shadow-2026-09-22 worktree: .worktrees/track-s1-summary-shadow-2026-09-22 write_scope: [".doc/ops/short-term/**", ".doc/system-design/module-boundaries.yaml", "internal/ctxmgr/**", "deploy/postgres/goose/**"] remote_sync: not_requested +head_sha: "0c572ddaca117c1033251ce8e33da90f342334f6" integration_state: pending +base_dev_sha: "364ed7f1a3f0393ef77ec4623a9518116ed2cb87" +policy_revision: "agent-governance-2026-09-21-r1 @364ed7f1" --- # W3-S1 收敛结论