Capture actual project work sources before reuse

Why: Project retrieval hits, tool results, and saved visible analysis must be individually traceable and fully analyzed before later project actions consume them.

Evidence: RED tests first showed missing run-source capture, tool-result capture, and run-wide preflight; go test -p 1 ./... and go build ./... passed; isolated PostgreSQL source, migration, audit, and context tests passed.

Impact: Persist actual run-bound sources with current-status checks and exact manifests behind the disabled X3 assembly gate; no live product path, migration, deployment, or release is enabled.

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

internal/app/project_work_observation.go

@@ -66,7 +66,7 @@ func (r recordRegistry) RegisterProjectWorkObserver(p extension.ProjectWorkObser
 // 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,
+	work func(context.Context, string, core.ID) (bool, error), limit int, verify ...func(context.Context, store.ProjectWorkObservationRow) error,
 ) (int, error) {
 	provider := slot.get()
 	if provider == nil || work == nil {
@@ -76,17 +76,26 @@ func deliverProjectWorkObservations(ctx context.Context, st *store.Store, slot *
 	if err != nil {
 		return 0, err
 	}
-	return deliverProjectWorkRows(ctx, st, provider, work, rows)
+	return deliverProjectWorkRows(ctx, st, provider, work, rows, verify...)
 }
 
 func deliverProjectWorkRows(ctx context.Context, st *store.Store, provider *extension.ProjectWorkObserver,
 	work func(context.Context, string, core.ID) (bool, error), rows []store.ProjectWorkObservationRow,
+	verify ...func(context.Context, store.ProjectWorkObservationRow) error,
 ) (int, error) {
 	completed := 0
 	for _, row := range rows {
 		if row.State != "pending" {
 			continue
 		}
+		if row.SourceKind != "user-message" && row.SourceKind != "assistant-message" {
+			if len(verify) != 1 || verify[0] == nil {
+				return completed, fmt.Errorf("project work source %s: current resource status verifier unavailable", row.EventID)
+			}
+			if err := verify[0](ctx, row); err != nil {
+				return completed, err
+			}
+		}
 		if err := ctx.Err(); err != nil {
 			return completed, err
 		}
@@ -103,7 +112,7 @@ func deliverProjectWorkRows(ctx context.Context, st *store.Store, provider *exte
 			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, DeliveryReason: row.Reason, ObservedAt: row.ObservedAt,
+			Text: row.Text, Scope: row.Scope, Delivery: row.Delivery, DeliveryReason: row.Reason, Metadata: row.Metadata, ObservedAt: row.ObservedAt,
 		}
 		result, err := provider.Observe(ctx, observation)
 		if err != nil {
@@ -124,6 +133,52 @@ func deliverProjectWorkRows(ctx context.Context, st *store.Store, provider *exte
 	return completed, nil
 }
 
+func verifyProjectKnowledgeObservation(_ context.Context, caps hostcaps, row store.ProjectWorkObservationRow) error {
+	if row.SourceKind != "retrieval-hit" || caps.st == nil {
+		return errors.New("project work source: unsupported resource verification")
+	}
+	doc, err := caps.st.ProjectWorkKnowledgeSource(row.SourceID)
+	if err != nil {
+		return errors.New("project work source: current knowledge source unavailable")
+	}
+	hash := sha256.Sum256([]byte(doc.Chunk))
+	actualHash := fmt.Sprintf("%x", hash)
+	if actualHash != row.SourceVersion || actualHash != row.SourceSHA256 || doc.Chunk != row.Text ||
+		(row.Metadata["ref"] != "" && row.Metadata["ref"] != doc.Ref) ||
+		(row.Metadata["source"] != "" && row.Metadata["source"] != doc.Source) ||
+		(doc.ProjectID != "" && string(doc.ProjectID) != row.ProjectID) {
+		return errors.New("project work source: knowledge version or scope changed")
+	}
+	if caps.authz != nil && caps.authz.hasV2() {
+		allowed, err := caps.authz.AuthorizeResource(extension.ResourceAuthzQuery{
+			Actor: row.Owner, Action: "model-context", ResourceType: "knowledge", ResourceID: doc.ID,
+			Owner: string(doc.Owner), ProjectID: doc.ProjectID, Source: doc.Source,
+		})
+		if err != nil || !allowed {
+			return errors.New("project work source: current knowledge authorization denied")
+		}
+	} else if string(doc.Owner) != row.Owner {
+		return errors.New("project work source: shared knowledge authorization unavailable")
+	}
+	return nil
+}
+
+// A transient tool result has no external source authority to recheck. Its
+// committed private outbox bytes and stored run are the only reliable source;
+// RDM must keep it a candidate until a separately verifiable evidence source
+// or human confirmation supports a formal project claim.
+func verifyProjectTransientObservation(_ context.Context, st *store.Store, row store.ProjectWorkObservationRow) error {
+	if row.SourceKind != "tool-result" || row.Metadata["sourceStatus"] != "transient" ||
+		row.Metadata["toolName"] == "" || row.RunID == "" {
+		return errors.New("project work source: transient result lacks trusted provenance")
+	}
+	run, err := st.GetRunByID(core.ID(row.RunID))
+	if err != nil || string(run.Owner) != row.Owner || string(run.ProjectID) != row.ProjectID || string(run.ConversationID) != row.ConversationID {
+		return errors.New("project work source: transient run binding unavailable")
+	}
+	return nil
+}
+
 // preflightProjectMessage delivers this committed message's pending parts
 // inline, then asks RDM to prove that every exact source part was analyzed for
 // the pending action. A durable intake ACK alone never permits the action.
@@ -198,3 +253,80 @@ func preflightProjectMessage(ctx context.Context, st *store.Store, slot *project
 	}
 	return decision, nil
 }
+
+// preflightProjectRun binds all committed user, retrieval, tool, and visible
+// analysis parts of one run before a later model or business action consumes
+// them. Missing/partial parts keep the action pending. The domain provider
+// decides whether the complete source set contains a blocking risk.
+func preflightProjectRun(ctx context.Context, st *store.Store, slot *projectWorkObserverSlot,
+	work func(context.Context, string, core.ID) (bool, error), verify func(context.Context, store.ProjectWorkObservationRow) error,
+	runID, action string,
+) (extension.ProjectWorkDecision, error) {
+	pending := extension.ProjectWorkDecision{State: extension.ProjectWorkPending, Reason: "run sources awaiting analysis"}
+	provider := slot.get()
+	if provider == nil || work == nil || verify == nil {
+		return pending, errors.New("extension: trusted project run observer unavailable")
+	}
+	run, err := st.GetRunByID(core.ID(runID))
+	if err != nil || run.ProjectID == "" {
+		return pending, errors.New("extension: trusted project run unavailable")
+	}
+	rows, err := st.ProjectRunObservations(runID)
+	if err != nil {
+		return pending, err
+	}
+	if len(rows) == 0 {
+		return pending, errors.New("extension: project run has no trusted source manifest")
+	}
+	if _, err := deliverProjectWorkRows(ctx, st, provider, work, rows, verify); err != nil {
+		return pending, err
+	}
+	rows, err = st.ProjectRunObservations(runID)
+	if err != nil {
+		return pending, err
+	}
+	manifest := make([]extension.ProjectWorkSource, 0, len(rows))
+	for _, row := range rows {
+		if row.State != "accepted" || row.Delivery != "complete" {
+			return pending, nil
+		}
+		if row.RunID != runID || row.Owner != string(run.Owner) || row.ProjectID != string(run.ProjectID) || row.ConversationID != string(run.ConversationID) {
+			return pending, errors.New("extension: mixed run source manifest")
+		}
+		manifest = append(manifest, extension.ProjectWorkSource{
+			EventID: row.EventID, SourceKind: row.SourceKind, SourceID: row.SourceID,
+			SourceVersion: row.SourceVersion, SourceSHA256: row.SourceSHA256, PartIndex: row.PartIndex,
+		})
+	}
+	conv, err := st.GetConversation(string(run.Owner), string(run.ConversationID))
+	if err != nil || conv.ProjectID != run.ProjectID {
+		return pending, errors.New("extension: run project binding changed before preflight")
+	}
+	allowed, err := work(ctx, string(run.Owner), run.ProjectID)
+	if err != nil || !allowed {
+		return pending, errors.New("extension: current run project work denied")
+	}
+	raw, err := json.Marshal(manifest)
+	if err != nil {
+		return pending, err
+	}
+	watermark := fmt.Sprintf("%x", sha256.Sum256(raw))
+	decision, err := provider.Preflight(ctx, extension.ProjectWorkAction{
+		Actor: string(run.Owner), Owner: string(run.Owner), ProjectID: run.ProjectID,
+		ConversationID: run.ConversationID, RunID: run.ID, Action: action,
+		Manifest: manifest, SourceWatermark: watermark,
+	})
+	if err != nil {
+		return pending, err
+	}
+	switch decision.State {
+	case extension.ProjectWorkProceed:
+		if decision.DecisionID == "" || decision.SourceWatermark != watermark {
+			return pending, errors.New("extension: run preflight decision did not bind current sources")
+		}
+	case extension.ProjectWorkClarify, extension.ProjectWorkPending, extension.ProjectWorkDenied:
+	default:
+		return pending, errors.New("extension: invalid project run preflight decision")
+	}
+	return decision, nil
+}

internal/app/project_work_observation_test.go

@@ -2,6 +2,8 @@ package app
 
 import (
 	"context"
+	"crypto/sha256"
+	"fmt"
 	"path/filepath"
 	"testing"
 
@@ -140,6 +142,167 @@ func TestProjectMessagePreflightRequiresAnalysisAndExactManifest(t *testing.T) {
 	}
 }
 
+func TestRetrievalObservationRechecksCurrentSourceBeforeDisclosure(t *testing.T) {
+	st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "source.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)
+	}
+	run := core.Run{ConversationID: conv.ID, ProjectID: project.ID, Owner: "alice", Kind: core.RunConversation, Status: core.RunRunning, Phase: core.PhaseThinking}
+	if err := st.CreateRun(&run); err != nil {
+		t.Fatal(err)
+	}
+	doc := core.KnowledgeDoc{ID: "knowledge-1", Owner: "alice", ProjectID: project.ID, Source: "upload", Ref: "https://example.test/a", Chunk: "actual source"}
+	if err := st.AddKnowledge(doc); err != nil {
+		t.Fatal(err)
+	}
+	observation := store.RunWorkObservationInput{RunID: string(run.ID), SourceKind: "retrieval-hit", SourceID: string(doc.ID),
+		SourceVersion: fmt.Sprintf("%x", sha256.Sum256([]byte(doc.Chunk))), Text: doc.Chunk, Delivery: "complete",
+		Metadata: map[string]string{"ref": doc.Ref, "source": doc.Source}}
+	if _, err := st.AppendRunWorkObservation(observation); err != nil {
+		t.Fatal(err)
+	}
+	called := 0
+	provider := workObserverForTest()
+	provider.Observe = func(context.Context, extension.WorkObservation) (extension.WorkObservationResult, error) {
+		called++
+		return extension.WorkObservationResult{State: extension.WorkObservationAccepted, ReceiptID: "rdm-1"}, nil
+	}
+	slot := &projectWorkObserverSlot{provider: &provider}
+	work := func(context.Context, string, core.ID) (bool, error) { return true, nil }
+	verify := func(ctx context.Context, row store.ProjectWorkObservationRow) error {
+		return verifyProjectKnowledgeObservation(ctx, hostcaps{st: st}, row)
+	}
+	if n, err := deliverProjectWorkObservations(context.Background(), st, slot, work, 10, verify); err != nil || n != 1 || called != 1 {
+		t.Fatalf("current owned source was not delivered: n=%d calls=%d err=%v", n, called, err)
+	}
+	if _, err := st.AppendRunWorkObservation(observation); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := st.DB().Exec(`DELETE FROM knowledge WHERE id=?`, doc.ID); err != nil {
+		t.Fatal(err)
+	}
+	if n, err := deliverProjectWorkObservations(context.Background(), st, slot, work, 10, verify); err == nil || n != 0 || called != 1 {
+		t.Fatalf("deleted source text crossed into observer: n=%d calls=%d err=%v", n, called, err)
+	}
+}
+
+func TestTransientToolSourceKeepsCurrentRunAndCandidateStatus(t *testing.T) {
+	st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "tool.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)
+	}
+	run := core.Run{ConversationID: conv.ID, ProjectID: project.ID, Owner: "alice", Kind: core.RunConversation, Status: core.RunRunning, Phase: core.PhaseThinking}
+	if err := st.CreateRun(&run); err != nil {
+		t.Fatal(err)
+	}
+	_, err = st.AppendRunWorkObservation(store.RunWorkObservationInput{RunID: string(run.ID), SourceKind: "tool-result", SourceID: "call-1", SourceVersion: "v1",
+		Text: "实际工具返回", Delivery: "complete", Metadata: map[string]string{"sourceStatus": "transient", "toolName": "lookup"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	rows, err := st.PendingProjectWorkObservations(10)
+	if err != nil || len(rows) != 1 {
+		t.Fatalf("tool source missing: %+v %v", rows, err)
+	}
+	if err := verifyProjectTransientObservation(context.Background(), st, rows[0]); err != nil {
+		t.Fatalf("committed private tool source rejected: %v", err)
+	}
+	if _, err := st.DB().Exec(`DELETE FROM runs WHERE id=?`, run.ID); err != nil {
+		t.Fatal(err)
+	}
+	if err := verifyProjectTransientObservation(context.Background(), st, rows[0]); err == nil {
+		t.Fatal("deleted run still authorized its transient source")
+	}
+}
+
+func TestRunPreflightWaitsForAllCommittedSourceParts(t *testing.T) {
+	st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "run-preflight.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)
+	}
+	run := core.Run{ConversationID: conv.ID, ProjectID: project.ID, Owner: "alice", Kind: core.RunConversation, Status: core.RunRunning, Phase: core.PhaseThinking}
+	if err := st.CreateRun(&run); 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"); err != nil {
+		t.Fatal(err)
+	}
+	if err := st.SetMessageRunIDAccountChecked("alice", string(msg.ID), string(run.ID), false); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := st.AppendRunWorkObservation(store.RunWorkObservationInput{RunID: string(run.ID), SourceKind: "tool-result", SourceID: "call-1",
+		SourceVersion: "v1", Text: "工具中的测量结果", Delivery: "complete", Metadata: map[string]string{"sourceStatus": "transient", "toolName": "measure"}}); err != nil {
+		t.Fatal(err)
+	}
+	ready, preflights, observed := false, 0, 0
+	provider := workObserverForTest()
+	provider.Observe = func(context.Context, extension.WorkObservation) (extension.WorkObservationResult, error) {
+		observed++
+		return extension.WorkObservationResult{State: extension.WorkObservationAccepted, ReceiptID: fmt.Sprintf("receipt-%d", observed)}, nil
+	}
+	provider.Preflight = func(_ context.Context, action extension.ProjectWorkAction) (extension.ProjectWorkDecision, error) {
+		preflights++
+		if action.RunID != run.ID || action.ProjectID != project.ID || len(action.Manifest) != 2 {
+			t.Fatalf("run manifest omitted a committed source: %+v", action)
+		}
+		if !ready {
+			return extension.ProjectWorkDecision{State: extension.ProjectWorkPending}, nil
+		}
+		return extension.ProjectWorkDecision{State: extension.ProjectWorkProceed, DecisionID: "approved-risk-check", SourceWatermark: action.SourceWatermark}, nil
+	}
+	slot := &projectWorkObserverSlot{provider: &provider}
+	work := func(context.Context, string, core.ID) (bool, error) { return true, nil }
+	verify := func(ctx context.Context, row store.ProjectWorkObservationRow) error {
+		return verifyProjectTransientObservation(ctx, st, row)
+	}
+	decision, err := preflightProjectRun(context.Background(), st, slot, work, verify, string(run.ID), "model.consume.tool")
+	if err != nil || decision.State != extension.ProjectWorkPending || observed != 2 || preflights != 1 {
+		t.Fatalf("intake ACK was mistaken for finished risk analysis: %+v observed=%d preflights=%d err=%v", decision, observed, preflights, err)
+	}
+	ready = true
+	decision, err = preflightProjectRun(context.Background(), st, slot, work, verify, string(run.ID), "model.consume.tool")
+	if err != nil || decision.State != extension.ProjectWorkProceed || observed != 2 || preflights != 2 {
+		t.Fatalf("fully analyzed run was not released once: %+v observed=%d preflights=%d err=%v", decision, observed, preflights, err)
+	}
+}
+
 func TestProjectWorkObservationRegistrationIsTrustedAndUnique(t *testing.T) {
 	slot := &projectWorkObserverSlot{}
 	trusted := pluginRegistry{pluginID: "rdm", caps: hostcaps{workObserver: slot}}

internal/runmgr/project_work_retrieval_test.go

@@ -0,0 +1,58 @@
+package runmgr
+
+import (
+	"context"
+	"path/filepath"
+	"testing"
+
+	"agent-system-backend/internal/store"
+	core "agent-system-core"
+)
+
+func TestProjectRetrievalRecordsEachActualAuthorizedHit(t *testing.T) {
+	st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "retrieval.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)
+	}
+	run := core.Run{ConversationID: conv.ID, ProjectID: project.ID, Owner: "alice", Kind: core.RunConversation, Status: core.RunRunning, Phase: core.PhaseThinking}
+	if err := st.CreateRun(&run); err != nil {
+		t.Fatal(err)
+	}
+	mgr := New(st, nil, nil, nil, nil, nil, nil, nil, nil)
+	fr := &sharedFakeRetriever{own: []core.RetrievalHit{
+		{Doc: core.KnowledgeDoc{ID: "k1", Owner: "alice", ProjectID: project.ID, Ref: "https://example.test/a", Chunk: "source A"}, Kind: "fulltext"},
+		{Doc: core.KnowledgeDoc{ID: "k2", Owner: "alice", ProjectID: project.ID, Ref: "https://example.test/b", Chunk: "source B"}, Kind: "fulltext"},
+	}}
+	mgr.SetRetriever(fr)
+	mgr.SetProjectWorkObservationEnabled(true)
+	rc := &runCtx{mgr: mgr, run: run, ctx: context.Background()}
+	hits, err := rc.Retrieve(core.RetrievalQuery{Text: "test", ProjectID: project.ID})
+	if err != nil || len(hits) != 2 {
+		t.Fatalf("authorized results not returned: %+v %v", hits, err)
+	}
+	rows, err := st.PendingProjectWorkObservations(10)
+	if err != nil || len(rows) != 2 {
+		t.Fatalf("actual hits not individually captured: %+v %v", rows, err)
+	}
+	for _, row := range rows {
+		if row.SourceKind != "retrieval-hit" || row.Owner != "alice" || row.RunID != string(run.ID) || row.Metadata["query"] != "test" || row.Text == "" {
+			t.Fatalf("retrieval source missing trusted identity or actual text: %+v", row)
+		}
+	}
+	fr.own = []core.RetrievalHit{{Doc: core.KnowledgeDoc{Chunk: "no stable ID"}}}
+	if hits, err := rc.Retrieve(core.RetrievalQuery{Text: "test", ProjectID: project.ID}); err == nil || hits != nil {
+		t.Fatalf("source without stable ID entered model input: %+v %v", hits, err)
+	}
+}

internal/runmgr/project_work_tool_result_test.go

@@ -0,0 +1,30 @@
+package runmgr
+
+import (
+	"testing"
+
+	core "agent-system-core"
+)
+
+func TestProjectToolResultRecordsActualOutputBeforeModelUse(t *testing.T) {
+	rc, st := newExecEnv(t)
+	run := core.Run{ConversationID: rc.run.ConversationID, ProjectID: rc.run.ProjectID, Owner: rc.run.Owner,
+		Kind: core.RunConversation, Status: core.RunRunning, Phase: core.PhaseThinking}
+	if err := st.CreateRun(&run); err != nil {
+		t.Fatal(err)
+	}
+	rc.run = run
+	rc.mgr.SetProjectWorkObservationEnabled(true)
+	result := rc.ExecuteTool(core.ToolCallRequest{CallID: "tool-1", Name: "time_now", Arguments: []byte(`{}`)})
+	if result.Status != "ok" || result.Output == "" {
+		t.Fatalf("tool did not return its actual output: %+v", result)
+	}
+	rows, err := st.PendingProjectWorkObservations(10)
+	if err != nil || len(rows) != 1 || rows[0].SourceKind != "tool-result" || rows[0].SourceID != "tool-1" || rows[0].Text != result.Output || rows[0].Metadata["toolName"] != "time_now" {
+		t.Fatalf("actual tool output was not captured before return: %+v %v", rows, err)
+	}
+	missingID := rc.ExecuteTool(core.ToolCallRequest{Name: "time_now", Arguments: []byte(`{}`)})
+	if missingID.Status != "error" {
+		t.Fatalf("tool output without stable call ID reached model: %+v", missingID)
+	}
+}

internal/runmgr/runmgr.go

@@ -5,6 +5,8 @@ package runmgr
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -67,6 +69,7 @@ type Manager struct {
 	projectContextResolver        func(context.Context, extension.ProjectContextRequest) (extension.ProjectContextBundle, error)
 	projectEvidenceReader         func(context.Context, extension.ProjectEvidenceRequest) (extension.ProjectEvidence, error)
 	projectSourceRechecker        func(context.Context, extension.ProjectContextSourceRecheckRequest) (extension.ProjectContextSourceRecheckResult, error)
+	projectWorkObservationEnabled bool
 	// commands EXT-14：命令即工具的投影与分发源（nil = 未接线，命令不进工具清单）。
 	commands CommandSource
 	indexer  core.DocumentIndexer // 可选索引入口（nil = store 整文档单块）
@@ -292,6 +295,13 @@ func (m *Manager) SetProjectWorkAuthorization(fn func(context.Context, string, c
 	m.projectWorkAuthorization = fn
 }
 
+// SetProjectWorkObservationEnabled is set at startup after the trusted
+// observer/preflight path is assembled. The disabled default preserves older
+// hosts and cannot be mistaken for successful source treatment.
+func (m *Manager) SetProjectWorkObservationEnabled(enabled bool) {
+	m.projectWorkObservationEnabled = enabled
+}
+
 // SetAccountEffectiveAuthorization binds the current account-state guard. It is
 // independent of the optional domain project provider.
 func (m *Manager) SetAccountEffectiveAuthorization(fn func(context.Context, string) (bool, error), allowImplicitLocal ...bool) {
@@ -1143,6 +1153,29 @@ func indexTextOf(name, mime string, content []byte) string {
 }
 
 func (c *runCtx) Retrieve(q core.RetrievalQuery) ([]core.RetrievalHit, error) {
+	hits, err := c.retrieveRaw(q)
+	if err != nil || !c.mgr.projectWorkObservationEnabled || c.run.ProjectID == "" {
+		return hits, err
+	}
+	for _, hit := range hits {
+		if hit.Doc.ID == "" {
+			return nil, errors.New("project retrieval hit lacks a stable source ID")
+		}
+		hash := sha256.Sum256([]byte(hit.Doc.Chunk))
+		_, err := c.mgr.st.AppendRunWorkObservation(store.RunWorkObservationInput{
+			RunID: string(c.run.ID), SourceKind: "retrieval-hit", SourceID: string(hit.Doc.ID),
+			SourceVersion: hex.EncodeToString(hash[:]), Text: hit.Doc.Chunk, Delivery: "complete",
+			Metadata: map[string]string{"query": q.Text, "ref": hit.Doc.Ref, "title": hit.Doc.Title,
+				"source": hit.Doc.Source, "hitKind": hit.Kind, "retrievedAt": time.Now().UTC().Format(time.RFC3339Nano)},
+		})
+		if err != nil {
+			return nil, err
+		}
+	}
+	return hits, nil
+}
+
+func (c *runCtx) retrieveRaw(q core.RetrievalQuery) ([]core.RetrievalHit, error) {
 	if q.TopK <= 0 {
 		q.TopK = int(c.eff.GetNumber("retrieval.topK", 5))
 	}
@@ -1240,6 +1273,9 @@ func (c *runCtx) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult {
 			if ct.ID != call.Name {
 				continue
 			}
+			if c.mgr.projectWorkObservationEnabled && c.run.ProjectID != "" {
+				return core.ToolCallResult{CallID: call.CallID, Status: "error", Output: "project command tool awaits trusted source preflight"}
+			}
 			var params map[string]any
 			if len(call.Arguments) > 0 {
 				_ = json.Unmarshal(call.Arguments, &params)
@@ -1276,6 +1312,23 @@ func (c *runCtx) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult {
 		last := tc.arts[len(tc.arts)-1]
 		res.Artifact = &last
 	}
+	if c.mgr.projectWorkObservationEnabled && c.run.ProjectID != "" {
+		if call.CallID == "" {
+			return core.ToolCallResult{CallID: call.CallID, Status: "error", Output: "project tool result lacks a stable call ID"}
+		}
+		hash := sha256.Sum256([]byte(res.Output))
+		meta := map[string]string{"toolName": call.Name, "status": res.Status, "sourceStatus": "transient",
+			"observedAt": time.Now().UTC().Format(time.RFC3339Nano)}
+		if res.Artifact != nil {
+			meta["artifactId"] = string(res.Artifact.ID)
+		}
+		if _, err := c.mgr.st.AppendRunWorkObservation(store.RunWorkObservationInput{
+			RunID: string(c.run.ID), SourceKind: "tool-result", SourceID: call.CallID,
+			SourceVersion: hex.EncodeToString(hash[:]), Text: res.Output, Delivery: "complete", Metadata: meta,
+		}); err != nil {
+			return core.ToolCallResult{CallID: call.CallID, Status: "error", Output: "project tool result could not be recorded"}
+		}
+	}
 	// 超长截断并转工件引用（4.10）。
 	limit := c.prof.Context.MaxToolResultChars
 	if limit <= 0 {

internal/store/migrate.go

@@ -1087,8 +1087,8 @@ func init() {
 			}},
 		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)`},
+				"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 '',payload_text TEXT NOT NULL DEFAULT '',metadata_json TEXT NOT NULL DEFAULT '{}')`, `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 '',payload_text TEXT NOT NULL DEFAULT '',metadata_json TEXT NOT NULL DEFAULT '{}')`, `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`},

internal/store/project_work_outbox.go

@@ -4,6 +4,7 @@ import (
 	"crypto/sha256"
 	"database/sql"
 	"encoding/hex"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"time"
@@ -34,9 +35,74 @@ type ProjectWorkObservationRow struct {
 	State          string
 	ReceiptID      string
 	Text           string
+	Metadata       map[string]string
 	Reason         string
 }
 
+// RunWorkObservationInput identifies an actual retrieval/tool/visible-analysis
+// result seen by a trusted run. Owner, project, and conversation are resolved
+// from the stored run; callers cannot supply or override those identities.
+type RunWorkObservationInput struct {
+	RunID         string
+	SourceKind    string // retrieval-hit|tool-result|visible-analysis
+	SourceID      string
+	SourceVersion string
+	Text          string
+	Delivery      string // complete|summary-only|partial
+	PartIndex     int
+	Metadata      map[string]string
+}
+
+func (s *Store) AppendRunWorkObservation(input RunWorkObservationInput) (string, error) {
+	if input.RunID == "" || input.SourceID == "" || input.SourceVersion == "" || input.PartIndex < 0 {
+		return "", errors.New("project work source: missing trusted source identity")
+	}
+	switch input.SourceKind {
+	case "retrieval-hit", "tool-result", "visible-analysis":
+	default:
+		return "", errors.New("project work source: unsupported kind")
+	}
+	if input.Delivery != "complete" && input.Delivery != "summary-only" && input.Delivery != "partial" {
+		return "", errors.New("project work source: unsupported delivery state")
+	}
+	if len(input.Text) > 64*1024 {
+		return "", errors.New("project work source: fragment exceeds bounded intake")
+	}
+	meta, err := json.Marshal(input.Metadata)
+	if err != nil || len(meta) > 8*1024 {
+		return "", errors.New("project work source: invalid or oversized metadata")
+	}
+	tx, err := s.db.Begin()
+	if err != nil {
+		return "", err
+	}
+	defer tx.Rollback()
+	var owner, projectID, conversationID, convOwner, convProject string
+	if err := tx.QueryRow(s.q(`SELECT r.owner,r.project_id,r.conversation_id,c.owner,c.project_id FROM runs r JOIN conversations c ON c.id=r.conversation_id WHERE r.id=?`), input.RunID).
+		Scan(&owner, &projectID, &conversationID, &convOwner, &convProject); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return "", ErrNotFound
+		}
+		return "", err
+	}
+	if owner != convOwner || projectID == "" || projectID != convProject {
+		return "", errors.New("project work source: run binding does not match current conversation")
+	}
+	hash := sha256.Sum256([]byte(input.Text))
+	eventID := newID("obs")
+	observed := time.Now().UTC().Format(time.RFC3339Nano)
+	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,payload_text,metadata_json)
+		VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`), eventID, owner, projectID, conversationID, input.RunID,
+		input.SourceKind, input.SourceID, input.SourceVersion, hex.EncodeToString(hash[:]), input.PartIndex,
+		0, len(input.Text), input.Delivery, "private", observed, input.Text, string(meta)); err != nil {
+		return "", err
+	}
+	if err := tx.Commit(); err != nil {
+		return "", err
+	}
+	return eventID, nil
+}
+
 func (s *Store) appendProjectMessageObservationsTx(tx *sql.Tx, m *core.Message) error {
 	if m == nil || (m.Role != core.RoleUser && m.Role != core.RoleAssistant) {
 		return nil
@@ -57,20 +123,26 @@ func (s *Store) appendProjectMessageObservationsTx(tx *sql.Tx, m *core.Message)
 		if part.Type == core.PartThinking {
 			continue
 		}
-		delivery, text := "partial", ""
+		delivery := "partial"
 		var material []byte
-		if part.Type == core.PartText && part.Text != "" {
-			delivery, text = "complete", part.Text
-			material = []byte(text)
-		} else {
+		switch part.Type {
+		case core.PartText:
+			material = []byte(part.Text)
+		case core.PartPlan, core.PartCheck, core.PartCard:
+			material = []byte(marshalJSON(part)) // saved, visible analysis artefact
+		default:
 			material = []byte(marshalJSON(part))
 		}
+		if (part.Type == core.PartText || part.Type == core.PartPlan || part.Type == core.PartCheck || part.Type == core.PartCard) && len(material) > 0 && len(material) <= 64*1024 {
+			delivery = "complete"
+		}
 		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 {
+		meta, _ := json.Marshal(map[string]string{"partType": string(part.Type)})
+		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,metadata_json)
+			VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`), eventID, owner, projectID, m.ConversationID, m.RunID,
+			kind, m.ID, m.ID, hex.EncodeToString(hash[:]), i, 0, len(material), delivery, "private", observed, string(meta)); err != nil {
 			return err
 		}
 	}
@@ -81,7 +153,7 @@ func (s *Store) appendProjectMessageObservationsTx(tx *sql.Tx, m *core.Message)
 // 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) {
-	return s.projectWorkObservations("", limit)
+	return s.projectWorkObservations("", "", limit)
 }
 
 // ProjectMessageObservations returns every committed part of one message,
@@ -91,26 +163,46 @@ func (s *Store) ProjectMessageObservations(messageID string) ([]ProjectWorkObser
 		return nil, errors.New("project work observation: empty message ID")
 	}
 	var count int
-	if err := s.db.QueryRow(s.q(`SELECT COUNT(*) FROM project_work_outbox WHERE source_id=?`), messageID).Scan(&count); err != nil {
+	if err := s.db.QueryRow(s.q(`SELECT COUNT(*) FROM project_work_outbox WHERE source_id=? AND source_kind IN ('user-message','assistant-message')`), messageID).Scan(&count); err != nil {
 		return nil, err
 	}
 	if count > 200 {
 		return nil, errors.New("project work observation: message parts exceed safe preflight limit")
 	}
-	return s.projectWorkObservations(messageID, 200)
+	return s.projectWorkObservations(messageID, "", 200)
 }
 
-func (s *Store) projectWorkObservations(messageID string, limit int) ([]ProjectWorkObservationRow, error) {
+// ProjectRunObservations is the exact committed source set linked to a run.
+// An over-limit set is an explicit incomplete manifest, never a truncated pass.
+func (s *Store) ProjectRunObservations(runID string) ([]ProjectWorkObservationRow, error) {
+	if runID == "" {
+		return nil, errors.New("project work observation: empty run ID")
+	}
+	var count int
+	if err := s.db.QueryRow(s.q(`SELECT COUNT(*) FROM project_work_outbox WHERE run_id=?`), runID).Scan(&count); err != nil {
+		return nil, err
+	}
+	if count > 200 {
+		return nil, errors.New("project work observation: run sources exceed safe preflight limit")
+	}
+	return s.projectWorkObservations("", runID, 200)
+}
+
+func (s *Store) projectWorkObservations(messageID, runID string, limit int) ([]ProjectWorkObservationRow, error) {
 	if limit <= 0 || limit > 200 {
 		limit = 50
 	}
-	query := `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
+	query := `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,payload_text,metadata_json
 		FROM project_work_outbox WHERE state='pending' ORDER BY observed_at,event_id LIMIT ?`
 	args := []any{limit}
 	if messageID != "" {
-		query = `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 source_id=? ORDER BY part_index LIMIT ?`
+		query = `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,payload_text,metadata_json
+			FROM project_work_outbox WHERE source_id=? AND source_kind IN ('user-message','assistant-message') ORDER BY part_index LIMIT ?`
 		args = []any{messageID, limit}
+	} else if runID != "" {
+		query = `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,payload_text,metadata_json
+			FROM project_work_outbox WHERE run_id=? ORDER BY observed_at,event_id LIMIT ?`
+		args = []any{runID, limit}
 	}
 	rows, err := s.db.Query(s.q(query), args...)
 	if err != nil {
@@ -119,13 +211,18 @@ func (s *Store) projectWorkObservations(messageID string, limit int) ([]ProjectW
 	out := make([]ProjectWorkObservationRow, 0, limit)
 	for rows.Next() {
 		var row ProjectWorkObservationRow
-		var observed string
+		var observed, payload, metadata 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 {
+			&row.SpanStart, &row.SpanEnd, &row.Delivery, &row.Scope, &observed, &row.State, &row.ReceiptID, &payload, &metadata); err != nil {
 			rows.Close()
 			return nil, err
 		}
+		if err := json.Unmarshal([]byte(metadata), &row.Metadata); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		row.Text = payload
 		row.ObservedAt, err = time.Parse(time.RFC3339Nano, observed)
 		if err != nil {
 			rows.Close()
@@ -143,6 +240,13 @@ func (s *Store) projectWorkObservations(messageID string, limit int) ([]ProjectW
 		if row.Delivery != "complete" {
 			continue
 		}
+		if row.SourceKind != "user-message" && row.SourceKind != "assistant-message" {
+			hash := sha256.Sum256([]byte(row.Text))
+			if hex.EncodeToString(hash[:]) != row.SourceSHA256 || row.SpanEnd != len(row.Text) {
+				row.Delivery, row.Reason, row.Text = "partial", "source-changed", ""
+			}
+			continue
+		}
 		msg, err := s.MessageByID(row.SourceID)
 		if errors.Is(err, ErrNotFound) {
 			row.Delivery, row.Reason = "partial", "source-missing"
@@ -151,11 +255,21 @@ func (s *Store) projectWorkObservations(messageID string, limit int) ([]ProjectW
 		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 {
+		if msg.ConversationID != core.ID(row.ConversationID) || row.PartIndex < 0 || row.PartIndex >= len(msg.Parts) {
+			row.Delivery, row.Reason = "partial", "source-changed"
+			continue
+		}
+		part := msg.Parts[row.PartIndex]
+		var text string
+		switch part.Type {
+		case core.PartText:
+			text = part.Text
+		case core.PartPlan, core.PartCheck, core.PartCard:
+			text = marshalJSON(part)
+		default:
 			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"

internal/store/project_work_outbox_test.go

@@ -64,6 +64,10 @@ func TestProjectMessageOutboxFollowsCommittedMessage(t *testing.T) {
 	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)
 	}
+	runSources, err := st.ProjectRunObservations(string(run.ID))
+	if err != nil || len(runSources) != 2 {
+		t.Fatalf("run preflight did not find its linked user parts: %+v %v", runSources, 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")
@@ -106,15 +110,25 @@ func TestAssistantObservationCommitsOnlyWithFencedRunSettlement(t *testing.T) {
 	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: "需要澄清效率的含义"}}}
+	answer := core.Message{ConversationID: conv.ID, RunID: run.ID, Role: core.RoleAssistant, Parts: []core.Part{
+		{Type: core.PartText, Text: "需要澄清效率的含义"},
+		{Type: core.PartPlan, Plan: &core.PlanSnapshot{Title: "可见分析计划", Steps: []core.PlanStep{{ID: "clarify", Text: "确认比较方向", Status: "pending"}}}},
+	}}
 	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 {
+	if err != nil || len(rows) != 2 {
 		t.Fatalf("final assistant observation was not committed with the run: %+v %v", rows, err)
 	}
+	byPart := map[int]ProjectWorkObservationRow{}
+	for _, row := range rows {
+		byPart[row.PartIndex] = row
+	}
+	if byPart[0].SourceKind != "assistant-message" || byPart[0].SourceID != string(answer.ID) || byPart[0].RunID != string(run.ID) || byPart[0].Text != answer.Parts[0].Text || byPart[1].Delivery != "complete" || byPart[1].Metadata["partType"] != string(core.PartPlan) || byPart[1].Text == "" {
+		t.Fatalf("visible analysis or final answer was lost: %+v", rows)
+	}
 	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"}}}
@@ -122,7 +136,7 @@ func TestAssistantObservationCommitsOnlyWithFencedRunSettlement(t *testing.T) {
 		t.Fatal("duplicate final message did not fail")
 	}
 	again, err := st.PendingProjectWorkObservations(10)
-	if err != nil || len(again) != 1 {
+	if err != nil || len(again) != 2 {
 		t.Fatalf("failed settlement added a second observation: %+v %v", again, err)
 	}
 }

internal/store/project_work_run_source_test.go

@@ -0,0 +1,52 @@
+package store
+
+import (
+	"testing"
+
+	core "agent-system-core"
+)
+
+func TestRunWorkSourceUsesStoredRunAndRetainsEachActualHit(t *testing.T) {
+	st := openTestStore(t)
+	if err := st.EnsureUser("alice", "Alice"); err != nil {
+		t.Fatal(err)
+	}
+	p, err := st.CreateProject("alice", "P")
+	if err != nil {
+		t.Fatal(err)
+	}
+	conv, err := st.CreateConversation("alice", string(p.ID), "Work", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	run := core.Run{ConversationID: conv.ID, ProjectID: p.ID, Owner: "alice", Kind: core.RunConversation, Status: core.RunRunning, Phase: core.PhaseThinking}
+	if err := st.CreateRun(&run); err != nil {
+		t.Fatal(err)
+	}
+	input := RunWorkObservationInput{RunID: string(run.ID), SourceKind: "retrieval-hit", SourceID: "knowledge-1", SourceVersion: "v1", Text: "实际命中的资料", Delivery: "complete", Metadata: map[string]string{"query": "效率", "ref": "https://example.test/a"}}
+	first, err := st.AppendRunWorkObservation(input)
+	if err != nil {
+		t.Fatal(err)
+	}
+	second, err := st.AppendRunWorkObservation(input)
+	if err != nil || first == second {
+		t.Fatalf("repeated actual hit lost occurrence identity: first=%q second=%q err=%v", first, second, err)
+	}
+	rows, err := st.PendingProjectWorkObservations(10)
+	if err != nil || len(rows) != 2 {
+		t.Fatalf("run observations missing: %+v %v", rows, err)
+	}
+	for _, row := range rows {
+		if row.Owner != "alice" || row.ProjectID != string(p.ID) || row.ConversationID != string(conv.ID) || row.RunID != string(run.ID) || row.SourceID != "knowledge-1" || row.Text != input.Text || row.Metadata["query"] != "效率" {
+			t.Fatalf("source was not derived from trusted run or actual hit: %+v", row)
+		}
+	}
+	runRows, err := st.ProjectRunObservations(string(run.ID))
+	if err != nil || len(runRows) != 2 {
+		t.Fatalf("run manifest omitted actual retrieval occurrences: %+v %v", runRows, err)
+	}
+	input.RunID = "run-forged"
+	if _, err := st.AppendRunWorkObservation(input); err == nil {
+		t.Fatal("unbound run locator became a project source")
+	}
+}

internal/store/project_work_source_status.go

@@ -0,0 +1,28 @@
+package store
+
+import (
+	"database/sql"
+	"errors"
+
+	core "agent-system-core"
+)
+
+// ProjectWorkKnowledgeSource returns a stored knowledge source to the trusted
+// host verifier. This is not a client read API: callers must independently
+// enforce current owner, project, and resource publication permissions before
+// disclosing its body to a project observer or model.
+func (s *Store) ProjectWorkKnowledgeSource(id string) (core.KnowledgeDoc, error) {
+	var doc core.KnowledgeDoc
+	var owner, created string
+	err := s.db.QueryRow(s.q(`SELECT id,owner,project_id,source,title,ref,chunk,created_at FROM knowledge WHERE id=?`), id).
+		Scan(&doc.ID, &owner, &doc.ProjectID, &doc.Source, &doc.Title, &doc.Ref, &doc.Chunk, &created)
+	if errors.Is(err, sql.ErrNoRows) {
+		return doc, ErrNotFound
+	}
+	if err != nil {
+		return doc, err
+	}
+	doc.Owner = core.ID(owner)
+	doc.CreatedAt, err = parseTime(created)
+	return doc, err
+}

pkg/extension/project_work.go

@@ -12,26 +12,27 @@ import (
 // Actor, ProjectID, source identity, or the source authorization epoch.
 // RunID may be empty for a newly committed user message or chat command.
 type WorkObservation struct {
-	EventID            string    `json:"eventId"`
-	Phase              string    `json:"phase"`
-	Actor              string    `json:"actor"`
-	Owner              string    `json:"owner"`
-	ProjectID          core.ID   `json:"projectId"`
-	ConversationID     core.ID   `json:"conversationId,omitempty"`
-	RunID              core.ID   `json:"runId,omitempty"`
-	SourceKind         string    `json:"sourceKind"`
-	SourceID           string    `json:"sourceId"`
-	SourceVersion      string    `json:"sourceVersion"`
-	SourceSHA256       string    `json:"sourceSha256"`
-	PartIndex          int       `json:"partIndex"`
-	SpanStart          int       `json:"spanStart"` // UTF-8 byte offset in the exact source
-	SpanEnd            int       `json:"spanEnd"`
-	Text               string    `json:"text,omitempty"` // only the actual authorized fragment
-	Scope              string    `json:"scope"`          // private|project|controlled
-	Delivery           string    `json:"delivery"`       // complete|summary-only|partial
-	DeliveryReason     string    `json:"deliveryReason,omitempty"`
-	AuthorizationEpoch string    `json:"authorizationEpoch"`
-	ObservedAt         time.Time `json:"observedAt"`
+	EventID            string            `json:"eventId"`
+	Phase              string            `json:"phase"`
+	Actor              string            `json:"actor"`
+	Owner              string            `json:"owner"`
+	ProjectID          core.ID           `json:"projectId"`
+	ConversationID     core.ID           `json:"conversationId,omitempty"`
+	RunID              core.ID           `json:"runId,omitempty"`
+	SourceKind         string            `json:"sourceKind"`
+	SourceID           string            `json:"sourceId"`
+	SourceVersion      string            `json:"sourceVersion"`
+	SourceSHA256       string            `json:"sourceSha256"`
+	PartIndex          int               `json:"partIndex"`
+	SpanStart          int               `json:"spanStart"` // UTF-8 byte offset in the exact source
+	SpanEnd            int               `json:"spanEnd"`
+	Text               string            `json:"text,omitempty"` // only the actual authorized fragment
+	Scope              string            `json:"scope"`          // private|project|controlled
+	Delivery           string            `json:"delivery"`       // complete|summary-only|partial
+	DeliveryReason     string            `json:"deliveryReason,omitempty"`
+	Metadata           map[string]string `json:"metadata,omitempty"`
+	AuthorizationEpoch string            `json:"authorizationEpoch"`
+	ObservedAt         time.Time         `json:"observedAt"`
 }
 
 type WorkObservationState string