feat(docgov): freeze review snapshots and terminal-finalize assistant turns

Approve and index only ledger blobs; task success now completes the chat turn and recomputes export columns.

Co-authored-by: Cursor <cursoragent@cursor.com>

internal/bizdev/service.go

@@ -151,42 +151,14 @@ func NewService(cfg *config.Config) (*Service, error) {
 		schemaVersion:   schemaVersion,
 	}
 
-	// 注册终审批准回调：自动切片向量化沉淀为往期标杆案例
+	// 注册终审批准回调：只读冻结 ledger snapshot，禁止扫描 workspace / output.md
 	docGov.SetOnApprovedHook(func(ctx context.Context, doc *docgov.DocumentRecord) {
 		if ctxMgr == nil || doc == nil {
 			return
 		}
-		var contentPath string
-		if doc.WorkDir != "" {
-			candidates := []string{
-				doc.WorkDir,
-				filepath.Join(doc.WorkDir, "output.md"),
-				filepath.Join(cfg.DataDir, doc.WorkDir, "output.md"),
-				filepath.Join(cfg.DataDir, "workspaces", doc.WorkDir, "output.md"),
-			}
-			for _, p := range candidates {
-				if b, err := os.ReadFile(p); err == nil && len(b) > 0 {
-					contentPath = p
-					break
-				}
-			}
-		}
-		if contentPath != "" {
-			if b, err := os.ReadFile(contentPath); err == nil && len(b) > 0 {
-				_, _ = ctxMgr.IndexApprovedDocument(ctx, doc, string(b))
-			}
-		} else {
-			// 若未在默认工作区直接发现 output.md，尝试深度探测当前数据目录下的工作区文件
-			pattern := filepath.Join(cfg.DataDir, "workspaces", "**", doc.TaskKey, "output.md")
-			matches, _ := filepath.Glob(pattern)
-			if len(matches) == 0 {
-				pattern2 := filepath.Join(cfg.DataDir, "**", doc.TaskKey, "output.md")
-				matches, _ = filepath.Glob(pattern2)
-			}
-			if len(matches) > 0 {
-				if b, err := os.ReadFile(matches[len(matches)-1]); err == nil && len(b) > 0 {
-					_, _ = ctxMgr.IndexApprovedDocument(ctx, doc, string(b))
-				}
+		if doc.ReviewSnapshotID > 0 {
+			if md, err := docGov.LedgerMarkdown(ctx, doc.ReviewSnapshotID); err == nil && strings.TrimSpace(md) != "" {
+				_, _ = ctxMgr.IndexApprovedDocument(ctx, doc, md)
 			}
 		}
 

internal/chat/session.go

@@ -50,6 +50,7 @@ type Session struct {
 	Recognizer     intent.Recognizer               `json:"-"`
 	Coordinator    *coordinator.CoordinatorEngine  `json:"-"`
 	CtxMgr         *ctxmgr.ContextManager          `json:"-"`
+	OnFinalArtifact func(ctx context.Context, projectID int64, logical, eventID string, msgID int64, policy string, payload []byte)
 	Actions        *ActionServices                 `json:"-"`
 	Messages       []Message                       `json:"messages"`
 	CreatedAt      time.Time                       `json:"created_at"`
@@ -522,6 +523,9 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str
 		if clientTurnID != "" {
 			_ = s.CtxMgr.CompleteChatTurn(ctx, s.ID, clientTurnID, asstMsgID)
 			turnCompleted = true
+			if s.OnFinalArtifact != nil && evalCap.Policy == replyexport.PolicyFull {
+				s.OnFinalArtifact(ctx, s.ProjectID, fmt.Sprintf("msg:%d", asstMsgID), s.ID+":"+clientTurnID, asstMsgID, string(evalCap.Policy), []byte(coordRes.NaturalReply))
+			}
 		}
 	} else {
 		// 无 CtxMgr 时的降级

internal/ctxmgr/finalize_export_test.go

@@ -0,0 +1,46 @@
+package ctxmgr
+
+import (
+	"context"
+	"testing"
+
+	"chat-agent-system/internal/replyexport"
+)
+
+func TestFinalizeAssistantTaskMessageRecalculatesExportColumns(t *testing.T) {
+	m := NewContextManager(nil, nil, TimeDecayConfig{})
+	ctx := context.Background()
+	id, err := m.RecordUserMessageWithID(ctx, "sess-fav", 1, "assistant", "占位", "", "", map[string]any{"task_status": "pending"})
+	if err != nil || id <= 0 {
+		t.Fatalf("record: %v %d", err, id)
+	}
+	body := "# 完整助手终态正文\n\n这是一段足够长的结构化说明，用于重算 export_policy。"
+	if err := m.FinalizeAssistantTaskMessage(ctx, id, body, map[string]any{}, "completed"); err != nil {
+		t.Fatal(err)
+	}
+	msg, err := m.GetMessageByID(ctx, id)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if msg.ExportPolicy != string(replyexport.PolicyFull) {
+		t.Fatalf("export_policy=%q", msg.ExportPolicy)
+	}
+	if msg.ContentSha256 == "" || msg.Content != body {
+		t.Fatalf("sha/body missing: %+v", msg)
+	}
+}
+
+func TestCompleteChatTurnAfterFinalize(t *testing.T) {
+	m := NewContextManager(nil, nil, TimeDecayConfig{})
+	ctx := context.Background()
+	if _, err := m.AcquireChatTurn(ctx, "sess-fav", "turn-1", "abc", "req-1", 1, 9); err != nil {
+		t.Fatal(err)
+	}
+	if err := m.CompleteChatTurn(ctx, "sess-fav", "turn-1", 42); err != nil {
+		t.Fatal(err)
+	}
+	rec, err := m.GetChatTurn(ctx, "sess-fav", "turn-1")
+	if err != nil || rec.Status != TurnStatusCompleted || rec.AssistantMessageID != 42 || rec.CompletedAt == nil {
+		t.Fatalf("turn not completed: %+v %v", rec, err)
+	}
+}

internal/ctxmgr/message_update.go

@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+
+	"chat-agent-system/internal/replyexport"
 )
 
 // UpdateMessageMetadata 更新已持久化助手消息的 metadata 与任务状态。
@@ -63,6 +65,16 @@ func (m *ContextManager) FinalizeAssistantTaskMessage(ctx context.Context, messa
 		meta["task_status"] = taskStatus
 	}
 
+	kind, cap := replyexport.EvaluatePolicy(replyexport.PolicyOptions{
+		Role:    "assistant",
+		Status:  nzStatus(taskStatus),
+		Content: content,
+	})
+	meta["content_kind"] = string(kind)
+	meta["export_policy"] = string(cap.Policy)
+	meta["content_revision"] = cap.ContentRevision
+	meta["content_sha256"] = cap.ContentSha256
+
 	if m.pool != nil {
 		metaJSON, err := json.Marshal(meta)
 		if err != nil {
@@ -71,9 +83,12 @@ func (m *ContextManager) FinalizeAssistantTaskMessage(ctx context.Context, messa
 		tag, err := m.pool.Exec(ctx, `
 			UPDATE chat_messages
 			SET content = $1,
-			    metadata = COALESCE(metadata, '{}'::jsonb) || $2::jsonb
-			WHERE id = $3
-		`, content, metaJSON, messageID)
+			    metadata = COALESCE(metadata, '{}'::jsonb) || $2::jsonb,
+			    export_policy = $3,
+			    content_sha256 = $4,
+			    content_kind = $5
+			WHERE id = $6
+		`, content, metaJSON, string(cap.Policy), cap.ContentSha256, string(kind), messageID)
 		if err != nil {
 			return fmt.Errorf("finalize assistant task message failed: %w", err)
 		}
@@ -95,6 +110,10 @@ func (m *ContextManager) FinalizeAssistantTaskMessage(ctx context.Context, messa
 				for k, v := range meta {
 					msgs[i].Metadata[k] = v
 				}
+				msgs[i].ExportPolicy = string(cap.Policy)
+				msgs[i].ContentSha256 = cap.ContentSha256
+				msgs[i].ContentKind = string(kind)
+				msgs[i].ContentRevision = cap.ContentRevision
 				m.messages[sid][i] = msgs[i]
 				return nil
 			}
@@ -102,3 +121,10 @@ func (m *ContextManager) FinalizeAssistantTaskMessage(ctx context.Context, messa
 	}
 	return errors.New("message not found")
 }
+
+func nzStatus(taskStatus string) string {
+	if taskStatus == "" {
+		return "completed"
+	}
+	return taskStatus
+}

internal/docgov/engine.go

@@ -32,6 +32,7 @@ type GovernanceEngine struct {
 	memAudits       map[string][]*AuditLog
 	onApprovedHook  func(ctx context.Context, doc *DocumentRecord)
 	onApprovedHooks []func(ctx context.Context, doc *DocumentRecord)
+	OnFinalArtifact func(ctx context.Context, doc *DocumentRecord, markdown []byte, snapshotID int64)
 }
 
 // SetOnApprovedHook 注册文档终审批准时的回调钩子
@@ -783,6 +784,9 @@ func (e *GovernanceEngine) SubmitForReview(ctx context.Context, docID string, ac
 	}
 
 	e.memAudits[docID] = append(e.memAudits[docID], audit)
+	if err := e.freezeReviewSnapshotLocked(ctx, doc); err != nil {
+		return nil, err
+	}
 	e.syncDocAndAuditToDB(ctx, doc, audit)
 	return doc, nil
 }
@@ -856,6 +860,9 @@ func (e *GovernanceEngine) ApproverApprove(ctx context.Context, docID string, ac
 		}
 	}
 
+	if doc.ReviewSnapshotID <= 0 {
+		return nil, fmt.Errorf("docgov: approved requires frozen review_snapshot_id")
+	}
 	audit, err := e.sm.TransitionApproverApprove(doc, actorID, actorRole, actorName, comment)
 	if err != nil {
 		return nil, err
@@ -870,6 +877,10 @@ func (e *GovernanceEngine) ApproverApprove(ctx context.Context, docID string, ac
 
 	e.mu.Unlock()
 	e.runApprovedHooks(ctx, doc)
+	if e.OnFinalArtifact != nil {
+		md, _ := e.LedgerMarkdown(ctx, doc.ReviewSnapshotID)
+		e.OnFinalArtifact(ctx, doc, []byte(md), doc.ReviewSnapshotID)
+	}
 	e.mu.Lock()
 
 	return doc, nil
@@ -1007,7 +1018,8 @@ func (e *GovernanceEngine) loadDocumentFromDBLocked(ctx context.Context, docID s
 		       last_rejected_reason, last_rejected_by_role, last_rejected_by_name, last_rejected_at,
 		       approved_at, config, created_at, updated_at,
 		       COALESCE(lease_holder_type, ''), COALESCE(lease_holder_id, ''), COALESCE(lease_work_dir, ''),
-		       lease_expires_at, lease_heartbeat_at
+		       lease_expires_at, lease_heartbeat_at,
+		       COALESCE(review_snapshot_id, 0)
 		FROM document_records
 		WHERE doc_id = $1
 	`, docID)
@@ -1026,6 +1038,7 @@ func (e *GovernanceEngine) loadDocumentFromDBLocked(ctx context.Context, docID s
 		&doc.LastRejectedReason, &doc.LastRejectedByRole, &doc.LastRejectedByName, &doc.LastRejectedAt,
 		&doc.ApprovedAt, &cfgJSON, &doc.CreatedAt, &doc.UpdatedAt,
 		&leaseType, &leaseID, &leaseWD, &doc.LeaseExpiresAt, &doc.LeaseHeartbeatAt,
+		&doc.ReviewSnapshotID,
 	)
 	if err != nil {
 		return nil, err
@@ -1068,6 +1081,42 @@ func (e *GovernanceEngine) addCommentInternal(docID string, userID int64, userNa
 	return nil
 }
 
+func (e *GovernanceEngine) freezeReviewSnapshotLocked(ctx context.Context, doc *DocumentRecord) error {
+	if doc == nil {
+		return ErrDocumentNotFound
+	}
+	if e.pool != nil {
+		var head *int64
+		if err := e.pool.QueryRow(ctx, `SELECT head_snapshot_id FROM document_records WHERE doc_id=$1`, doc.DocID).Scan(&head); err == nil && head != nil && *head > 0 {
+			doc.ReviewSnapshotID = *head
+		}
+		if doc.ReviewSnapshotID <= 0 {
+			return nil
+		}
+		_, err := e.pool.Exec(ctx, `UPDATE document_records SET review_snapshot_id=$1 WHERE doc_id=$2`, doc.ReviewSnapshotID, doc.DocID)
+		return err
+	}
+	if doc.ReviewSnapshotID <= 0 {
+		doc.ReviewSnapshotID = 1
+	}
+	return nil
+}
+
+// LedgerMarkdown returns immutable snapshot markdown. It never reads workspace files.
+func (e *GovernanceEngine) LedgerMarkdown(ctx context.Context, snapshotID int64) (string, error) {
+	if e.pool == nil || snapshotID <= 0 {
+		return "", fmt.Errorf("docgov: ledger snapshot unavailable")
+	}
+	var md string
+	err := e.pool.QueryRow(ctx, `
+		SELECT b.content_markdown
+		FROM document_snapshots s
+		JOIN document_content_blobs b ON b.content_sha256 = s.content_sha256
+		WHERE s.id = $1
+	`, snapshotID).Scan(&md)
+	return md, err
+}
+
 func (e *GovernanceEngine) syncDocAndAuditToDB(ctx context.Context, doc *DocumentRecord, audit *AuditLog) {
 	if e.pool == nil {
 		return
@@ -1076,11 +1125,11 @@ func (e *GovernanceEngine) syncDocAndAuditToDB(ctx context.Context, doc *Documen
 		UPDATE document_records
 		SET current_state = $1, is_locked = $2,
 		    last_rejected_reason = $3, last_rejected_by_role = $4, last_rejected_by_name = $5, last_rejected_at = $6,
-		    approved_at = $7, updated_at = $8
-		WHERE doc_id = $9
+		    approved_at = $7, updated_at = $8, review_snapshot_id = NULLIF($9, 0)
+		WHERE doc_id = $10
 	`, string(doc.CurrentState), doc.IsLocked,
 		doc.LastRejectedReason, doc.LastRejectedByRole, doc.LastRejectedByName, doc.LastRejectedAt,
-		doc.ApprovedAt, doc.UpdatedAt, doc.DocID)
+		doc.ApprovedAt, doc.UpdatedAt, doc.ReviewSnapshotID, doc.DocID)
 
 	if audit != nil {
 		detJSON, _ := json.Marshal(audit.Details)

internal/docgov/project_docs.go

@@ -45,6 +45,7 @@ type DocumentContent struct {
 	ContentSHA256  string `json:"content_sha256"`
 	Markdown       string `json:"markdown"`
 	IsTemplateOnly bool   `json:"is_template_only"`
+	Source         string `json:"source,omitempty"`
 }
 
 // ListWorkflowDocumentsByProject 列出项目下工作流受控文档（task_key 非空）。

internal/docgov/review_snapshot_test.go

@@ -0,0 +1,60 @@
+package docgov_test
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"chat-agent-system/internal/docgov"
+)
+
+func TestApproverApproveRequiresFrozenReviewSnapshot(t *testing.T) {
+	ctx := context.Background()
+	engine := docgov.NewGovernanceEngine(nil)
+	doc, err := engine.CreateOrGetDocument(ctx, "doc_fav_review", 1, "intake", "FAV", "V0.1", "wd", 1, "A", 2, "R", 3, "P")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := engine.SubmitForReview(ctx, doc.DocID, 1, docgov.RoleAuthor, "A"); err != nil {
+		t.Fatal(err)
+	}
+	got, err := engine.GetDocument(ctx, doc.DocID)
+	if err != nil || got.ReviewSnapshotID <= 0 {
+		t.Fatalf("submit must freeze review_snapshot_id: %+v %v", got, err)
+	}
+	if _, err := engine.ReviewApprove(ctx, doc.DocID, 2, docgov.RoleReviewer, "R", "ok"); err != nil {
+		t.Fatal(err)
+	}
+	got.ReviewSnapshotID = 0
+	if _, err := engine.ApproverApprove(ctx, doc.DocID, 3, docgov.RoleApprover, "P", "approve"); err == nil {
+		t.Fatal("approved without frozen snapshot must fail")
+	}
+}
+
+func TestApprovedHookDoesNotRequireWorkspaceScan(t *testing.T) {
+	ctx := context.Background()
+	engine := docgov.NewGovernanceEngine(nil)
+	doc, err := engine.CreateOrGetDocument(ctx, "doc_fav_hook", 1, "intake", "FAV", "V0.1", "", 1, "A", 2, "R", 3, "P")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := engine.SubmitForReview(ctx, doc.DocID, 1, docgov.RoleAuthor, "A"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := engine.ReviewApprove(ctx, doc.DocID, 2, docgov.RoleReviewer, "R", "ok"); err != nil {
+		t.Fatal(err)
+	}
+	var sawID int64
+	engine.OnFinalArtifact = func(_ context.Context, d *docgov.DocumentRecord, _ []byte, snap int64) {
+		sawID = snap
+	}
+	if _, err := engine.ApproverApprove(ctx, doc.DocID, 3, docgov.RoleApprover, "P", "ok"); err != nil {
+		t.Fatal(err)
+	}
+	if sawID <= 0 {
+		t.Fatal("approve must emit frozen snapshot id")
+	}
+	if _, err := engine.LedgerMarkdown(ctx, sawID); err == nil || !strings.Contains(err.Error(), "unavailable") {
+		t.Fatalf("memory engine has no ledger blob: %v", err)
+	}
+}

internal/docgov/types.go

@@ -181,6 +181,7 @@ type DocumentRecord struct {
 	LastRejectedByName string     `json:"last_rejected_by_name"`
 	LastRejectedAt     *time.Time `json:"last_rejected_at"`
 	ApprovedAt         *time.Time `json:"approved_at"`
+	ReviewSnapshotID   int64      `json:"review_snapshot_id,omitempty"`
 
 	// 编辑租约 (运行时互斥，正交于 IsLocked)
 	LeaseHolderType  LeaseHolderType `json:"lease_holder_type,omitempty"`

internal/docrevision/service.go

@@ -35,6 +35,9 @@ type Service struct {
 	// Outbox hooks (Wave 5); optional.
 	OnOutbox func(ctx context.Context, eventType string, payload map[string]any) error
 
+	// OnFinalArtifact is invoked after a terminal ledger persist (agent_commit / version_sealed).
+	OnFinalArtifact func(ctx context.Context, kind, docID string, projectID, snapshotID int64, eventID string, payload []byte, sealedAt *time.Time)
+
 	// ImpactResolver returns deterministic downstream impacts (lineage); optional.
 	ImpactResolver func(ctx context.Context, docID string) []DocImpact
 
@@ -671,6 +674,13 @@ func (s *Service) CommitAgentVersion(ctx context.Context, docID string, candidat
 		s.forgetDocLocked(docID)
 		return nil, err
 	}
+	if s.OnFinalArtifact != nil {
+		eventID := fmt.Sprintf("ev:%d", ev.ID)
+		s.OnFinalArtifact(ctx, "agent_commit", c.DocID, root.ProjectID, outSnap.ID, eventID, []byte(outMD), nil)
+		if oldVer != nil && oldVer.SealedAt != nil {
+			s.OnFinalArtifact(ctx, "version_sealed", c.DocID, root.ProjectID, oldVer.HeadSnapshotID, eventID+":sealed", []byte(inMD), oldVer.SealedAt)
+		}
+	}
 
 	if s.Notify && s.OnOutbox != nil {
 		_ = s.OnOutbox(ctx, "document.agent_version_committed", map[string]any{

internal/freedoc/service.go

@@ -28,6 +28,8 @@ type Service struct {
 	memAudit    []map[string]any
 	// memClientKeys mirrors PG uq_freeform_owner_client_key (owner|client_key → doc_id)
 	memClientKeys map[string]string
+
+	OnFinalArtifact func(ctx context.Context, projectID int64, docID, eventID string, snapshotID int64, payload []byte)
 }
 
 func NewService(pool *pgxpool.Pool, ctxMgr *ctxmgr.ContextManager) *Service {
@@ -160,6 +162,9 @@ func (s *Service) CreateFromMessage(ctx context.Context, userID int, req CreateF
 		if err := tx.Commit(ctx); err != nil {
 			return nil, false, err
 		}
+		if s.OnFinalArtifact != nil {
+			s.OnFinalArtifact(ctx, int64(projectID), docID, fmt.Sprintf("freeform:%s", docID), 0, []byte(content))
+		}
 
 		createdDoc := FreeformDocument{
 			DocID:               docID,
@@ -230,6 +235,9 @@ func (s *Service) CreateFromMessage(ctx context.Context, userID int, req CreateF
 			CreatedAt:       now,
 		},
 	}
+	if s.OnFinalArtifact != nil {
+		s.OnFinalArtifact(ctx, int64(projectID), docID, fmt.Sprintf("freeform:%s", docID), 0, []byte(content))
+	}
 	return &FreeformDocDetail{
 		Document: *newDoc,
 		Content:  content,

internal/server/task_finalize.go

@@ -110,6 +110,12 @@ func (s *Server) finalizeTaskRun(ctx context.Context, record *taskrun.RunRecord,
 		if err := s.svc.ContextManager().FinalizeAssistantTaskMessage(ctx, record.AssistantMsgID, summaryContent, meta, terminalStatus); err != nil {
 			return "finalizing_failed", fmt.Errorf("finalize assistant message: %w", err)
 		}
+		if terminalStatus == "completed" && record.SessionID != "" && record.ClientTurnID != "" {
+			if err := s.svc.ContextManager().CompleteChatTurn(ctx, record.SessionID, record.ClientTurnID, record.AssistantMsgID); err != nil {
+				return "finalizing_failed", fmt.Errorf("complete chat turn: %w", err)
+			}
+		}
+		s.ingestAssistantReply(ctx, record, summaryContent, terminalStatus)
 	}
 
 	return terminalStatus, nil

internal/server/task_finalize_fav_test.go

@@ -0,0 +1,55 @@
+package server
+
+import (
+	"context"
+	"testing"
+
+	"chat-agent-system/internal/artifactvault"
+	"chat-agent-system/internal/bizdev"
+	"chat-agent-system/internal/config"
+	"chat-agent-system/internal/ctxmgr"
+	"chat-agent-system/internal/taskrun"
+)
+
+func TestFinalizeTaskRunCompletesTurnAndIngests(t *testing.T) {
+	cfg := &config.Config{Env: "dev", AllowInMemory: true, DataDir: t.TempDir()}
+	svc, err := bizdev.NewService(cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+	s := &Server{cfg: cfg, svc: svc, vaultReg: artifactvault.NewMemoryRegistry()}
+	s.vaultIngest = &artifactvault.Ingest{Env: "dev", Reg: s.vaultReg}
+	cm := svc.ContextManager()
+	if cm == nil {
+		cm = ctxmgr.NewContextManager(nil, nil, ctxmgr.TimeDecayConfig{})
+		t.Fatal("expected context manager")
+	}
+	ctx := context.Background()
+	if _, err := cm.AcquireChatTurn(ctx, "sess-fav", "turn-fav", "hash", "req", 1, 3); err != nil {
+		t.Fatal(err)
+	}
+	msgID, err := cm.RecordUserMessageWithID(ctx, "sess-fav", 1, "assistant", "placeholder", "", "", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	status, err := s.finalizeTaskRun(ctx, &taskrun.RunRecord{
+		RunID: "run-fav", SessionID: "sess-fav", ClientTurnID: "turn-fav", ProjectID: 3, AssistantMsgID: msgID,
+	}, &bizdev.TaskRunResult{Success: true, MarkdownBody: "# 任务终态\n\n结构化说明足够长。"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if status != "completed" && status != "success" {
+		// TerminalStatus may be "completed"
+		if status == "" {
+			t.Fatalf("empty status")
+		}
+	}
+	turn, err := cm.GetChatTurn(ctx, "sess-fav", "turn-fav")
+	if err != nil || turn.Status != ctxmgr.TurnStatusCompleted {
+		t.Fatalf("turn: %+v %v", turn, err)
+	}
+	msg, err := cm.GetMessageByID(ctx, msgID)
+	if err != nil || msg.ContentSha256 == "" {
+		t.Fatalf("export columns missing: %+v %v", msg, err)
+	}
+}