Persist private project message observations atomically

Why: Project user messages and final assistant replies need exact, replayable source parts before a domain observer can safely analyze them.

Evidence: go test -p 1 ./... passed; isolated PostgreSQL outbox, audit, migration, and work-context tests passed against a temporary cluster; request dedup, fenced settlement rollback, authorization recheck, and receipt idempotency tests passed.

Impact: Add migration 44 and a private host outbox for committed message parts; actual observer delivery and action preflight are not yet wired into the product path.

Refs: TT-N03, TT-R01; approved X3 stage-five plan r1

cmd/storage-migrate/main.go

@@ -285,7 +285,7 @@ func cmdBackup(args []string) error {
 // admission_leases 是运行态执行许可（迁移无意义），明确排除。
 var copyOrder = []string{
 	"projects", "users", "account_directory_revision", "profiles", "settings", "plugins",
-	"conversations", "runs", "messages", "events", "artifacts",
+	"conversations", "runs", "messages", "project_work_outbox", "events", "artifacts",
 	"hitl_cards", "plans", "todos", "knowledge", "knowledge_vectors",
 	"activity_cards", "project_event_shares", "command_audit", "project_audit_links", "prompt_modules",
 	"sessions", "api_keys",
@@ -953,7 +953,7 @@ func advanceSequences(db *sql.DB, table string) error {
 // 除外）。未知 owner 阻塞迁移（U4/R7：不静默带过未知归属）。
 func checkOwners(db *sql.DB, src *sql.DB) error {
 	ownerTables := []string{"projects", "conversations", "runs", "artifacts", "knowledge", "hitl_cards", "profiles", "activity_cards",
-		"project_event_shares", "command_audit", "project_audit_links", "work_task_bindings", "context_manifests", "workflow_step_sources",
+		"project_event_shares", "command_audit", "project_audit_links", "project_work_outbox", "work_task_bindings", "context_manifests", "workflow_step_sources",
 		"sessions", "api_keys", "login_keys", "account_events", "recovery_proposals"}
 	for _, t := range ownerTables {
 		if !sqliteHasTable(src, t) {

internal/app/project_work_observation.go

@@ -1,10 +1,14 @@
 package app
 
 import (
+	"context"
 	"errors"
+	"fmt"
 	"sync"
 
+	"agent-system-backend/internal/store"
 	"agent-system-backend/pkg/extension"
+	core "agent-system-core"
 )
 
 type projectWorkObserverSlot struct {
@@ -54,3 +58,57 @@ func (r recordRegistry) RegisterProjectWorkObserver(p extension.ProjectWorkObser
 	}
 	return reg.RegisterProjectWorkObserver(p)
 }
+
+// deliverProjectWorkObservations replays only committed source parts, rechecking
+// their owner and current project work authority before disclosing text to RDM.
+// An Observe ACK is an intake receipt, not a preflight or publication decision.
+// Failures leave the row pending for a later idempotent retry.
+func deliverProjectWorkObservations(ctx context.Context, st *store.Store, slot *projectWorkObserverSlot,
+	work func(context.Context, string, core.ID) (bool, error), limit int,
+) (int, error) {
+	provider := slot.get()
+	if provider == nil || work == nil {
+		return 0, errors.New("extension: trusted project work observer unavailable")
+	}
+	rows, err := st.PendingProjectWorkObservations(limit)
+	if err != nil {
+		return 0, err
+	}
+	completed := 0
+	for _, row := range rows {
+		if err := ctx.Err(); err != nil {
+			return completed, err
+		}
+		conv, err := st.GetConversation(row.Owner, row.ConversationID)
+		if err != nil || string(conv.ProjectID) != row.ProjectID {
+			return completed, fmt.Errorf("project work source %s: current conversation binding unavailable", row.EventID)
+		}
+		allowed, err := work(ctx, row.Owner, core.ID(row.ProjectID))
+		if err != nil || !allowed {
+			return completed, fmt.Errorf("project work source %s: current project authorization denied", row.EventID)
+		}
+		observation := extension.WorkObservation{
+			EventID: row.EventID, Phase: "source-committed", Actor: row.Owner, Owner: row.Owner,
+			ProjectID: core.ID(row.ProjectID), ConversationID: core.ID(row.ConversationID), RunID: core.ID(row.RunID),
+			SourceKind: row.SourceKind, SourceID: row.SourceID, SourceVersion: row.SourceVersion,
+			SourceSHA256: row.SourceSHA256, PartIndex: row.PartIndex, SpanStart: row.SpanStart, SpanEnd: row.SpanEnd,
+			Text: row.Text, Scope: row.Scope, Delivery: row.Delivery, ObservedAt: row.ObservedAt,
+		}
+		result, err := provider.Observe(ctx, observation)
+		if err != nil {
+			return completed, err
+		}
+		switch result.State {
+		case extension.WorkObservationPending:
+			continue
+		case extension.WorkObservationAccepted, extension.WorkObservationPartial, extension.WorkObservationDenied:
+			if err := st.FinishProjectWorkObservation(row.EventID, result.ReceiptID, string(result.State)); err != nil {
+				return completed, err
+			}
+			completed++
+		default:
+			return completed, fmt.Errorf("project work source %s: invalid observer result %q", row.EventID, result.State)
+		}
+	}
+	return completed, nil
+}

internal/app/project_work_observation_test.go

@@ -2,9 +2,12 @@ package app
 
 import (
 	"context"
+	"path/filepath"
 	"testing"
 
+	"agent-system-backend/internal/store"
 	"agent-system-backend/pkg/extension"
+	core "agent-system-core"
 )
 
 func workObserverForTest() extension.ProjectWorkObserver {
@@ -21,6 +24,55 @@ func workObserverForTest() extension.ProjectWorkObserver {
 	}
 }
 
+func TestProjectWorkDeliveryRechecksCurrentAuthorityAndACKsOnce(t *testing.T) {
+	st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "work.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer st.Close()
+	if err := st.EnsureUser("alice", "Alice"); err != nil {
+		t.Fatal(err)
+	}
+	project, err := st.CreateProject("alice", "P")
+	if err != nil {
+		t.Fatal(err)
+	}
+	conv, err := st.CreateConversation("alice", string(project.ID), "Work", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	msg := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "要求效率和成本都低于人工"}}}
+	if err := st.AppendMessageWithRequestAccountChecked("alice", false, &msg, "req-1"); err != nil {
+		t.Fatal(err)
+	}
+	called := 0
+	slot := &projectWorkObserverSlot{}
+	observer := workObserverForTest()
+	observer.Observe = func(_ context.Context, source extension.WorkObservation) (extension.WorkObservationResult, error) {
+		called++
+		if source.Owner != "alice" || source.Actor != "alice" || source.ProjectID != project.ID || source.SourceID != string(msg.ID) || source.Text != msg.Parts[0].Text || source.Scope != "private" {
+			t.Fatalf("provider received forged or wrong source: %+v", source)
+		}
+		return extension.WorkObservationResult{State: extension.WorkObservationAccepted, ReceiptID: "rdm-receipt-1"}, nil
+	}
+	if err := (pluginRegistry{pluginID: "rdm", caps: hostcaps{workObserver: slot}}).RegisterProjectWorkObserver(observer); err != nil {
+		t.Fatal(err)
+	}
+	denied := func(context.Context, string, core.ID) (bool, error) { return false, nil }
+	if _, err := deliverProjectWorkObservations(context.Background(), st, slot, denied, 10); err == nil || called != 0 {
+		t.Fatalf("revoked source was disclosed: calls=%d err=%v", called, err)
+	}
+	allowed := func(_ context.Context, actor string, id core.ID) (bool, error) {
+		return actor == "alice" && id == project.ID, nil
+	}
+	if delivered, err := deliverProjectWorkObservations(context.Background(), st, slot, allowed, 10); err != nil || delivered != 1 || called != 1 {
+		t.Fatalf("authorized source delivery failed: delivered=%d calls=%d err=%v", delivered, called, err)
+	}
+	if delivered, err := deliverProjectWorkObservations(context.Background(), st, slot, allowed, 10); err != nil || delivered != 0 || called != 1 {
+		t.Fatalf("delivered source was replayed without need: delivered=%d calls=%d err=%v", delivered, called, err)
+	}
+}
+
 func TestProjectWorkObservationRegistrationIsTrustedAndUnique(t *testing.T) {
 	slot := &projectWorkObserverSlot{}
 	trusted := pluginRegistry{pluginID: "rdm", caps: hostcaps{workObserver: slot}}

internal/store/messages.go

@@ -48,6 +48,9 @@ func (s *Store) appendMessageWithRequest(m *core.Message, requestID, actor strin
 	if m.ID == "" {
 		m.ID = newID("msg")
 	}
+	if m.CreatedAt.IsZero() {
+		m.CreatedAt = time.Now().UTC()
+	}
 	m.CreatedAt = m.CreatedAt.UTC()
 	insert := `INSERT INTO messages(id,conversation_id,run_id,role,parts,seq,created_at,request_id)
 		SELECT ?,?,?,?,?,COALESCE(MAX(seq),0)+1,?,? FROM messages WHERE conversation_id=?`
@@ -113,7 +116,10 @@ func (s *Store) appendMessageRows(tx *sql.Tx, m *core.Message, requestID, insert
 	if err := s.dialect.lockConversation(tx, string(m.ConversationID)); err != nil {
 		return err
 	}
-	if _, err := tx.Exec(s.q(insert), m.ID, m.ConversationID, m.RunID, string(m.Role), marshalJSON(m.Parts), nowISO(), requestID, m.ConversationID); err != nil {
+	if _, err := tx.Exec(s.q(insert), m.ID, m.ConversationID, m.RunID, string(m.Role), marshalJSON(m.Parts), m.CreatedAt.Format(time.RFC3339Nano), requestID, m.ConversationID); err != nil {
+		return err
+	}
+	if err := s.appendProjectMessageObservationsTx(tx, m); err != nil {
 		return err
 	}
 	_, err := tx.Exec(s.q(`UPDATE conversations SET updated_at=? WHERE id=?`), nowISO(), m.ConversationID)

internal/store/migrate.go

@@ -1085,6 +1085,15 @@ func init() {
 					`DROP FUNCTION project_audit_writers_guarded(regclass)`,
 				},
 			}},
+		migration{version: 44, name: "project_work_observation_outbox",
+			up: map[string][]string{
+				"sqlite":   {`CREATE TABLE project_work_outbox (event_id TEXT PRIMARY KEY,owner TEXT NOT NULL,project_id TEXT NOT NULL,conversation_id TEXT NOT NULL,run_id TEXT NOT NULL DEFAULT '',source_kind TEXT NOT NULL,source_id TEXT NOT NULL,source_version TEXT NOT NULL,source_sha256 TEXT NOT NULL,part_index INTEGER NOT NULL,span_start INTEGER NOT NULL,span_end INTEGER NOT NULL,delivery TEXT NOT NULL,scope TEXT NOT NULL,observed_at TEXT NOT NULL,state TEXT NOT NULL DEFAULT 'pending',receipt_id TEXT NOT NULL DEFAULT '',UNIQUE(source_kind,source_id,part_index))`, `CREATE INDEX idx_project_work_outbox_pending ON project_work_outbox(state,observed_at,event_id)`},
+				"postgres": {`CREATE TABLE project_work_outbox (event_id TEXT PRIMARY KEY,owner TEXT NOT NULL,project_id TEXT NOT NULL,conversation_id TEXT NOT NULL,run_id TEXT NOT NULL DEFAULT '',source_kind TEXT NOT NULL,source_id TEXT NOT NULL,source_version TEXT NOT NULL,source_sha256 TEXT NOT NULL,part_index INTEGER NOT NULL,span_start INTEGER NOT NULL,span_end INTEGER NOT NULL,delivery TEXT NOT NULL,scope TEXT NOT NULL,observed_at TEXT NOT NULL,state TEXT NOT NULL DEFAULT 'pending',receipt_id TEXT NOT NULL DEFAULT '',UNIQUE(source_kind,source_id,part_index))`, `CREATE INDEX idx_project_work_outbox_pending ON project_work_outbox(state,observed_at,event_id)`},
+			},
+			down: map[string][]string{
+				"sqlite":   {`DROP INDEX IF EXISTS idx_project_work_outbox_pending`, `DROP TABLE IF EXISTS project_work_outbox`},
+				"postgres": {`DROP INDEX IF EXISTS idx_project_work_outbox_pending`, `DROP TABLE IF EXISTS project_work_outbox`},
+			}},
 	)
 }
 

internal/store/migrate_compat.go

@@ -48,6 +48,7 @@ var migrationBreaksOldReaders = map[int]bool{
 	41: false, // project_audit_committed_links：新表，旧读者不受影响
 	42: false, // project_audit_committed_prefix：PG 审计 ID 默认函数，旧读者兼容
 	43: false, // project_audit_all_writers_guard：PG 在线审计写入门禁，旧读者兼容
+	44: false, // project_work_observation_outbox：独立私域投递表，旧读者不读取
 }
 
 // SchemaTarget 是本二进制会迁移到的 schema 版本。

internal/store/project_work_outbox.go

@@ -0,0 +1,181 @@
+package store
+
+import (
+	"crypto/sha256"
+	"database/sql"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"time"
+
+	core "agent-system-core"
+)
+
+// ProjectWorkObservationRow is a private host delivery record. A project ID is
+// a trusted locator here, not authorization to reveal the original message to
+// other project members. The source body is fetched from the message row only
+// for the current provider delivery and is never part of project audit DTOs.
+type ProjectWorkObservationRow struct {
+	EventID        string
+	Owner          string
+	ProjectID      string
+	ConversationID string
+	RunID          string
+	SourceKind     string
+	SourceID       string
+	SourceVersion  string
+	SourceSHA256   string
+	PartIndex      int
+	SpanStart      int
+	SpanEnd        int
+	Delivery       string
+	Scope          string
+	ObservedAt     time.Time
+	State          string
+	ReceiptID      string
+	Text           string
+	Reason         string
+}
+
+func (s *Store) appendProjectMessageObservationsTx(tx *sql.Tx, m *core.Message) error {
+	if m == nil || (m.Role != core.RoleUser && m.Role != core.RoleAssistant) {
+		return nil
+	}
+	var owner, projectID string
+	if err := tx.QueryRow(s.q(`SELECT owner,project_id FROM conversations WHERE id=?`), m.ConversationID).Scan(&owner, &projectID); err != nil {
+		return err
+	}
+	if projectID == "" {
+		return nil
+	}
+	kind := string(m.Role) + "-message"
+	observed := m.CreatedAt.UTC().Format(time.RFC3339Nano)
+	for i, part := range m.Parts {
+		// A tool summary, artifact reference, card, or hidden thinking part is not
+		// evidence of the original resource body. The separate retrieval/tool
+		// adapters will supply their actual source IDs and spans later.
+		if part.Type == core.PartThinking {
+			continue
+		}
+		delivery, text := "partial", ""
+		var material []byte
+		if part.Type == core.PartText && part.Text != "" {
+			delivery, text = "complete", part.Text
+			material = []byte(text)
+		} else {
+			material = []byte(marshalJSON(part))
+		}
+		hash := sha256.Sum256(material)
+		idDigest := sha256.Sum256([]byte(fmt.Sprintf("%s:%s:%d", kind, m.ID, i)))
+		eventID := "obs_" + hex.EncodeToString(idDigest[:16])
+		if _, err := tx.Exec(s.q(`INSERT INTO project_work_outbox(event_id,owner,project_id,conversation_id,run_id,source_kind,source_id,source_version,source_sha256,part_index,span_start,span_end,delivery,scope,observed_at)
+			VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`), eventID, owner, projectID, m.ConversationID, m.RunID,
+			kind, m.ID, m.ID, hex.EncodeToString(hash[:]), i, 0, len(text), delivery, "private", observed); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// PendingProjectWorkObservations returns committed, not-yet-ACKed source parts.
+// It rechecks that the immutable source bytes still match the recorded hash.
+// Missing or changed source bytes become partial, never invented complete text.
+func (s *Store) PendingProjectWorkObservations(limit int) ([]ProjectWorkObservationRow, error) {
+	if limit <= 0 || limit > 200 {
+		limit = 50
+	}
+	rows, err := s.db.Query(s.q(`SELECT event_id,owner,project_id,conversation_id,run_id,source_kind,source_id,source_version,source_sha256,part_index,span_start,span_end,delivery,scope,observed_at,state,receipt_id
+		FROM project_work_outbox WHERE state='pending' ORDER BY observed_at,event_id LIMIT ?`), limit)
+	if err != nil {
+		return nil, err
+	}
+	out := make([]ProjectWorkObservationRow, 0, limit)
+	for rows.Next() {
+		var row ProjectWorkObservationRow
+		var observed string
+		if err := rows.Scan(&row.EventID, &row.Owner, &row.ProjectID, &row.ConversationID, &row.RunID,
+			&row.SourceKind, &row.SourceID, &row.SourceVersion, &row.SourceSHA256, &row.PartIndex,
+			&row.SpanStart, &row.SpanEnd, &row.Delivery, &row.Scope, &observed, &row.State, &row.ReceiptID); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		row.ObservedAt, err = time.Parse(time.RFC3339Nano, observed)
+		if err != nil {
+			rows.Close()
+			return nil, err
+		}
+		out = append(out, row)
+	}
+	err = rows.Err()
+	rows.Close()
+	if err != nil {
+		return nil, err
+	}
+	for i := range out {
+		row := &out[i]
+		if row.Delivery != "complete" {
+			continue
+		}
+		msg, err := s.MessageByID(row.SourceID)
+		if errors.Is(err, ErrNotFound) {
+			row.Delivery, row.Reason = "partial", "source-missing"
+			continue
+		}
+		if err != nil {
+			return nil, err
+		}
+		if msg.ConversationID != core.ID(row.ConversationID) || row.PartIndex < 0 || row.PartIndex >= len(msg.Parts) || msg.Parts[row.PartIndex].Type != core.PartText {
+			row.Delivery, row.Reason = "partial", "source-changed"
+			continue
+		}
+		text := msg.Parts[row.PartIndex].Text
+		hash := sha256.Sum256([]byte(text))
+		if hex.EncodeToString(hash[:]) != row.SourceSHA256 || row.SpanEnd != len(text) {
+			row.Delivery, row.Reason = "partial", "source-changed"
+			continue
+		}
+		row.Text = text
+	}
+	return out, nil
+}
+
+// AckProjectWorkObservation records only the provider's durable intake receipt.
+// It does not assert that source analysis, risk checks, or project publication
+// finished. A different receipt for the same event is a conflict, not a retry.
+func (s *Store) AckProjectWorkObservation(eventID, receiptID string) error {
+	return s.FinishProjectWorkObservation(eventID, receiptID, "accepted")
+}
+
+// FinishProjectWorkObservation records the provider's terminal intake result.
+// "accepted" is only durable receipt; "partial" and "denied" are truthful
+// terminal treatment of this source part, not a project fact publication.
+func (s *Store) FinishProjectWorkObservation(eventID, receiptID, state string) error {
+	if eventID == "" || receiptID == "" {
+		return errors.New("project work observation: empty event or receipt")
+	}
+	if state != "accepted" && state != "partial" && state != "denied" {
+		return errors.New("project work observation: invalid terminal state")
+	}
+	res, err := s.db.Exec(s.q(`UPDATE project_work_outbox SET state=?,receipt_id=? WHERE event_id=? AND state='pending'`), state, receiptID, eventID)
+	if err != nil {
+		return err
+	}
+	changed, err := res.RowsAffected()
+	if err != nil {
+		return err
+	}
+	if changed == 1 {
+		return nil
+	}
+	var currentReceipt, currentState string
+	if err := s.db.QueryRow(s.q(`SELECT receipt_id,state FROM project_work_outbox WHERE event_id=?`), eventID).Scan(&currentReceipt, &currentState); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return ErrNotFound
+		}
+		return err
+	}
+	if currentReceipt != receiptID || currentState != state {
+		return fmt.Errorf("project work observation: receipt conflict for %s", eventID)
+	}
+	return nil
+}

internal/store/project_work_outbox_test.go

@@ -0,0 +1,124 @@
+package store
+
+import (
+	"crypto/sha256"
+	"fmt"
+	"testing"
+	"time"
+
+	core "agent-system-core"
+)
+
+func TestProjectMessageOutboxFollowsCommittedMessage(t *testing.T) {
+	st := openTestStore(t)
+	if err := st.EnsureUser("alice", "Alice"); err != nil {
+		t.Fatal(err)
+	}
+	p, err := st.CreateProject("alice", "Project")
+	if err != nil {
+		t.Fatal(err)
+	}
+	projectConv, err := st.CreateConversation("alice", string(p.ID), "Work", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	privateConv, err := st.CreateConversation("alice", string(p.ID), "Private", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	msg := core.Message{ConversationID: projectConv.ID, Role: core.RoleUser, Parts: []core.Part{
+		{Type: core.PartText, Text: "要求效率和成本都低于人工"},
+		{Type: core.PartArtifactRef, Artifact: &core.ArtifactRef{ArtifactID: "file-1"}},
+	}}
+	if err := st.AppendMessageWithRequestAccountChecked("alice", false, &msg, "request-1"); err != nil {
+		t.Fatal(err)
+	}
+	rows, err := st.PendingProjectWorkObservations(10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(rows) != 2 {
+		t.Fatalf("message and artifact parts were not each recorded: %+v", rows)
+	}
+	byPart := map[int]ProjectWorkObservationRow{}
+	for _, row := range rows {
+		byPart[row.PartIndex] = row
+	}
+	wantHash := fmt.Sprintf("%x", sha256.Sum256([]byte(msg.Parts[0].Text)))
+	textPart := byPart[0]
+	if textPart.ProjectID != string(p.ID) || textPart.Owner != "alice" || textPart.SourceID != string(msg.ID) || textPart.SourceSHA256 != wantHash || textPart.Text != msg.Parts[0].Text || textPart.Delivery != "complete" || textPart.RunID != "" {
+		t.Fatalf("incorrect trusted user message observation: %+v", textPart)
+	}
+	artifactPart := byPart[1]
+	if artifactPart.Delivery != "partial" || artifactPart.Text != "" || artifactPart.SourceID != string(msg.ID) {
+		t.Fatalf("unparsed attachment was presented as analyzed text: %+v", artifactPart)
+	}
+	run := core.Run{ConversationID: projectConv.ID, ProjectID: p.ID, Owner: "alice", Kind: core.RunConversation, Status: core.RunPending, Phase: core.PhaseThinking}
+	if err := st.CreateRun(&run); err != nil {
+		t.Fatal(err)
+	}
+	if err := st.SetMessageRunIDAccountChecked("alice", string(msg.ID), string(run.ID), false); err != nil {
+		t.Fatal(err)
+	}
+	linked, err := st.PendingProjectWorkObservations(10)
+	if err != nil || len(linked) != 2 || linked[0].RunID != string(run.ID) || linked[1].RunID != string(run.ID) {
+		t.Fatalf("later trusted run link was not saved for every source part: %+v %v", linked, err)
+	}
+	retry := core.Message{ConversationID: projectConv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "changed"}}}
+	if err := st.AppendMessageWithRequestAccountChecked("alice", false, &retry, "request-1"); err == nil {
+		t.Fatal("duplicate request did not surface existing message")
+	}
+	private := core.Message{ConversationID: privateConv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "private"}}}
+	if err := st.AppendMessageWithRequestAccountChecked("alice", false, &private, "request-2"); err != nil {
+		t.Fatal(err)
+	}
+	again, err := st.PendingProjectWorkObservations(10)
+	if err != nil || len(again) != 3 {
+		t.Fatalf("duplicate request generated an extra source or private source was lost: %+v %v", again, err)
+	}
+	for _, row := range again {
+		if row.Scope != "private" {
+			t.Fatalf("project locator alone changed a message's private scope: %+v", row)
+		}
+	}
+	if err := st.AckProjectWorkObservation(textPart.EventID, "receipt-1"); err != nil {
+		t.Fatal(err)
+	}
+	if err := st.AckProjectWorkObservation(textPart.EventID, "receipt-1"); err != nil {
+		t.Fatalf("same committed receipt was not idempotent: %v", err)
+	}
+	if err := st.AckProjectWorkObservation(textPart.EventID, "different"); err == nil {
+		t.Fatal("different observer receipt replaced the first one")
+	}
+	pending, err := st.PendingProjectWorkObservations(10)
+	if err != nil || len(pending) != 2 {
+		t.Fatalf("ACK did not remove only the delivered source: %+v %v", pending, err)
+	}
+}
+
+func TestAssistantObservationCommitsOnlyWithFencedRunSettlement(t *testing.T) {
+	st := openTestStore(t)
+	conv := seedConversation(t, st)
+	run := seedRun(t, st, conv, core.RunRunning)
+	now := time.Now().UTC()
+	run.Status, run.Phase, run.EndedAt = core.RunCompleted, core.PhaseDone, &now
+	answer := core.Message{ConversationID: conv.ID, RunID: run.ID, Role: core.RoleAssistant, Parts: []core.Part{{Type: core.PartText, Text: "需要澄清效率的含义"}}}
+	updated, err := st.SettleRun(&run, &answer)
+	if err != nil || !updated {
+		t.Fatalf("run settlement failed: updated=%t err=%v", updated, err)
+	}
+	rows, err := st.PendingProjectWorkObservations(10)
+	if err != nil || len(rows) != 1 || rows[0].SourceKind != "assistant-message" || rows[0].SourceID != string(answer.ID) || rows[0].RunID != string(run.ID) || rows[0].Text != answer.Parts[0].Text {
+		t.Fatalf("final assistant observation was not committed with the run: %+v %v", rows, err)
+	}
+	secondRun := seedRun(t, st, conv, core.RunRunning)
+	secondRun.Status, secondRun.Phase, secondRun.EndedAt = core.RunCompleted, core.PhaseDone, &now
+	failed := core.Message{ID: answer.ID, ConversationID: conv.ID, RunID: secondRun.ID, Role: core.RoleAssistant, Parts: []core.Part{{Type: core.PartText, Text: "different"}}}
+	if _, err := st.SettleRun(&secondRun, &failed); err == nil {
+		t.Fatal("duplicate final message did not fail")
+	}
+	again, err := st.PendingProjectWorkObservations(10)
+	if err != nil || len(again) != 1 {
+		t.Fatalf("failed settlement added a second observation: %+v %v", again, err)
+	}
+}

internal/store/recovery.go

@@ -256,9 +256,15 @@ func (s *Store) settleRunOnce(r *core.Run, msg *core.Message, fence *fenceRef, a
 		return false, false, err
 	}
 	if msg != nil {
+		if msg.CreatedAt.IsZero() {
+			msg.CreatedAt = time.Now().UTC()
+		}
 		insert := `INSERT INTO messages(id,conversation_id,run_id,role,parts,seq,created_at,request_id)
 			SELECT ?,?,?,?,?,COALESCE(MAX(seq),0)+1,?,? FROM messages WHERE conversation_id=?`
-		if _, err := tx.Exec(s.q(insert), msg.ID, msg.ConversationID, msg.RunID, string(msg.Role), marshalJSON(msg.Parts), nowISO(), "", msg.ConversationID); err != nil {
+		if _, err := tx.Exec(s.q(insert), msg.ID, msg.ConversationID, msg.RunID, string(msg.Role), marshalJSON(msg.Parts), msg.CreatedAt.UTC().Format(time.RFC3339Nano), "", msg.ConversationID); err != nil {
+			return false, false, err
+		}
+		if err := s.appendProjectMessageObservationsTx(tx, msg); err != nil {
 			return false, false, err
 		}
 		if msg.VariantGroup != "" || msg.Excluded {

internal/store/run_account_write.go

@@ -96,6 +96,9 @@ func (s *Store) SetMessageRunIDAccountChecked(actor, messageID, runID string, al
 		} else if n != 1 {
 			return ErrNotFound
 		}
+		if _, err := tx.Exec(s.q(`UPDATE project_work_outbox SET run_id=? WHERE source_kind='user-message' AND source_id=?`), runID, messageID); err != nil {
+			return err
+		}
 		return nil
 	})
 }