internal/app/app.go

@@ -614,12 +614,13 @@ func Run(opt Options) error {
 	// hostcaps：hostapi 闭包与进程内插件句柄共用的 owner 作用域能力（EXT-1 批注 1）。
 	authzSlot := &projectAuthz{}
 	projectSlot := &projectLifecycleSlot{}
+	workObserverSlot := &projectWorkObserverSlot{}
 	directorySlot, err := newAccountDirectorySlot(st)
 	if err != nil {
 		return err
 	}
 	contextSlot := &projectContextSlot{}
-	caps := hostcaps{projects: projectSlot, contexts: contextSlot, projectAudit: &projectAuditSlot{}, directory: directorySlot, st: st, pipe: pipe, set: setSvc, authz: authzSlot, surfaces: surfaceSvc, allowImplicitLocal: !authOn,
+	caps := hostcaps{projects: projectSlot, workObserver: workObserverSlot, contexts: contextSlot, projectAudit: &projectAuditSlot{}, directory: directorySlot, st: st, pipe: pipe, set: setSvc, authz: authzSlot, surfaces: surfaceSvc, allowImplicitLocal: !authOn,
 		accountGuard: func(owner string) error {
 			valid, err := effectiveAccount(context.Background(), owner)
 			if err != nil {
@@ -646,6 +647,7 @@ func Run(opt Options) error {
 		return err
 	}
 	mgr.SetProjectWorkAuthorization(projectSlot.work())
+	wireProjectWorkGate(mgr, st, caps, workObserverSlot, projectSlot.work())
 	if contextSlot.active() {
 		mgr.SetProjectContextResolver(contextSlot.resolve)
 		mgr.SetProjectEvidenceReader(contextSlot.readEvidence)

internal/app/project_work_observation.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"sync"
 
+	"agent-system-backend/internal/runmgr"
 	"agent-system-backend/internal/store"
 	"agent-system-backend/pkg/extension"
 	core "agent-system-core"
@@ -31,6 +32,33 @@ func (s *projectWorkObserverSlot) get() *extension.ProjectWorkObserver {
 	return &copy
 }
 
+func (s *projectWorkObserverSlot) active() bool { return s.get() != nil }
+
+func wireProjectWorkGate(mgr *runmgr.Manager, st *store.Store, caps hostcaps, slot *projectWorkObserverSlot,
+	work func(context.Context, string, core.ID) (bool, error),
+) {
+	provider := slot.get()
+	if provider == nil || work == nil {
+		return
+	}
+	verify := func(ctx context.Context, row store.ProjectWorkObservationRow) error {
+		switch row.SourceKind {
+		case "retrieval-hit":
+			return verifyProjectKnowledgeObservation(ctx, caps, row)
+		case "tool-result", "visible-analysis":
+			return verifyProjectTransientObservation(ctx, st, row)
+		default:
+			return fmt.Errorf("project work source %s: unsupported run source kind %q", row.EventID, row.SourceKind)
+		}
+	}
+	mgr.SetProjectWorkGate(
+		func(ctx context.Context, runID, action string) (extension.ProjectWorkDecision, error) {
+			return preflightProjectRun(ctx, st, slot, work, verify, runID, action)
+		},
+		provider.Respond,
+	)
+}
+
 func (r compiledRegistry) RegisterProjectWorkObserver(extension.ProjectWorkObserver) error {
 	return errors.New("extension: declarative plugins cannot register project work observers")
 }
@@ -168,8 +196,9 @@ func verifyProjectKnowledgeObservation(_ context.Context, caps hostcaps, row sto
 // 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 == "" {
+	toolSource := row.SourceKind == "tool-result" && row.Metadata["sourceStatus"] == "transient" && row.Metadata["toolName"] != ""
+	visibleAnalysis := row.SourceKind == "visible-analysis" && row.Metadata["stage"] == "assistant-final" && row.Metadata["partType"] != ""
+	if (!toolSource && !visibleAnalysis) || row.RunID == "" {
 		return errors.New("project work source: transient result lacks trusted provenance")
 	}
 	run, err := st.GetRunByID(core.ID(row.RunID))

internal/runmgr/project_work_gate.go

@@ -0,0 +1,124 @@
+package runmgr
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"agent-system-backend/internal/store"
+	"agent-system-backend/pkg/extension"
+	core "agent-system-core"
+)
+
+type projectWorkPreflight func(context.Context, string, string) (extension.ProjectWorkDecision, error)
+type projectRiskRespond func(context.Context, extension.ProjectRiskAnswer) (extension.ProjectWorkDecision, error)
+type projectRiskAsk func(core.HITLCard) (core.HITLResponse, error)
+
+var errProjectWorkPending = errors.New("project work source analysis pending")
+
+func (c *runCtx) stageAssistantWorkSources(parts []core.Part) error {
+	if !c.mgr.projectWorkObservationEnabled || c.run.ProjectID == "" {
+		return nil
+	}
+	for i, part := range parts {
+		var body string
+		switch part.Type {
+		case core.PartText:
+			body = part.Text
+		case core.PartPlan, core.PartCheck, core.PartCard:
+			raw, err := json.Marshal(part)
+			if err != nil {
+				return err
+			}
+			body = string(raw)
+		default:
+			continue
+		}
+		if body == "" {
+			continue
+		}
+		hash := sha256.Sum256([]byte(body))
+		if _, err := c.mgr.st.AppendRunWorkObservation(store.RunWorkObservationInput{
+			RunID: string(c.run.ID), SourceKind: "visible-analysis",
+			SourceID:      fmt.Sprintf("%s:assistant-final:%d", c.run.ID, i),
+			SourceVersion: hex.EncodeToString(hash[:]), Text: body, Delivery: "complete", PartIndex: i,
+			Metadata: map[string]string{"partType": string(part.Type), "stage": "assistant-final"},
+		}); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func resolveProjectWorkDecision(ctx context.Context, actor, projectID, runID, action string,
+	preflight projectWorkPreflight, respond projectRiskRespond, ask projectRiskAsk,
+) (extension.ProjectWorkDecision, error) {
+	pending := extension.ProjectWorkDecision{State: extension.ProjectWorkPending, Reason: "project work risk decision incomplete"}
+	if preflight == nil {
+		return pending, errors.New("project work gate unavailable")
+	}
+	decision, err := preflight(ctx, runID, action)
+	if err != nil {
+		return pending, err
+	}
+	switch decision.State {
+	case extension.ProjectWorkProceed:
+		if decision.DecisionID == "" {
+			return pending, errors.New("project work proceed decision is not committed")
+		}
+		return decision, nil
+	case extension.ProjectWorkDenied:
+		return decision, errors.New("project work action denied: " + decision.Reason)
+	case extension.ProjectWorkPending:
+		return pending, errProjectWorkPending
+	case extension.ProjectWorkClarify:
+		if decision.RiskID == "" || decision.RiskRevision == "" || decision.Question == "" || respond == nil || ask == nil {
+			return pending, errors.New("project work clarification is incomplete")
+		}
+	default:
+		return pending, fmt.Errorf("invalid project work decision %q", decision.State)
+	}
+	choices := make([]core.HITLChoice, 0, len(decision.Options))
+	for i, option := range decision.Options {
+		choices = append(choices, core.HITLChoice{
+			ID: fmt.Sprintf("option-%d", i), Label: option,
+			Description: "澄清这条项目信息的实际含义", Impact: "选择结果将用于重新判断相关项目动作",
+			Recommended: i == 0,
+		})
+	}
+	response, err := ask(core.HITLCard{
+		Question: decision.Question, Kind: core.HITLKindChoice, Choices: choices,
+		AllowFreeInput: true, FreeInputHint: "也可以输入更准确的表述", Blocked: true,
+		BlockReason: decision.Reason,
+	})
+	if err != nil {
+		return pending, err
+	}
+	answer := response.FreeText
+	if response.ChoiceID != "" {
+		for _, choice := range choices {
+			if choice.ID == response.ChoiceID {
+				answer = choice.Label
+				break
+			}
+		}
+	}
+	if answer == "" {
+		return pending, errors.New("project work clarification answer is empty")
+	}
+	idempotency := sha256.Sum256([]byte(decision.RiskID + "\x00" + decision.RiskRevision + "\x00" + runID + "\x00" + answer))
+	resolved, err := respond(ctx, extension.ProjectRiskAnswer{
+		Actor: actor, Owner: actor, ProjectID: core.ID(projectID), RiskID: decision.RiskID,
+		RiskRevision: decision.RiskRevision, Answer: answer, IdempotencyKey: fmt.Sprintf("%x", idempotency),
+	})
+	if err != nil {
+		return pending, err
+	}
+	if resolved.State != extension.ProjectWorkProceed || resolved.DecisionID == "" {
+		return pending, errors.New("project work clarification did not commit a proceed decision")
+	}
+	return resolved, nil
+}

internal/runmgr/project_work_gate_test.go

@@ -0,0 +1,231 @@
+package runmgr
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"agent-system-backend/internal/profiles"
+	backendruntime "agent-system-backend/internal/runtime"
+	"agent-system-backend/pkg/extension"
+	core "agent-system-core"
+)
+
+type projectGateRunner struct{ calls atomic.Int32 }
+
+func (*projectGateRunner) Name() string { return "project-gate-test" }
+
+func (r *projectGateRunner) Execute(backendruntime.RunCtx) error {
+	r.calls.Add(1)
+	return nil
+}
+
+type projectReplyRunner struct{}
+
+func (*projectReplyRunner) Name() string { return "project-reply-test" }
+func (*projectReplyRunner) Execute(rc backendruntime.RunCtx) error {
+	return rc.SetAssistantParts([]core.Part{{Type: core.PartText, Text: "最终项目答复"}})
+}
+
+func enableProjectGateForTest(m *Manager) {
+	m.SetProjectWorkGate(
+		func(context.Context, string, string) (extension.ProjectWorkDecision, error) {
+			return extension.ProjectWorkDecision{State: extension.ProjectWorkProceed, DecisionID: "test-decision"}, nil
+		},
+		func(context.Context, extension.ProjectRiskAnswer) (extension.ProjectWorkDecision, error) {
+			return extension.ProjectWorkDecision{State: extension.ProjectWorkProceed, DecisionID: "test-decision"}, nil
+		},
+	)
+}
+
+func TestProjectWorkGateClarifiesBeforeProceeding(t *testing.T) {
+	preflights, responses, questions := 0, 0, 0
+	preflight := func(context.Context, string, string) (extension.ProjectWorkDecision, error) {
+		preflights++
+		if preflights == 1 {
+			return extension.ProjectWorkDecision{
+				State: extension.ProjectWorkClarify, RiskID: "risk-1", RiskRevision: "r1",
+				Question: "效率低于人工是指更慢吗？", Options: []string{"效率更高且成本更低", "完成时间和成本更低", "确实允许效率更低"},
+			}, nil
+		}
+		return extension.ProjectWorkDecision{State: extension.ProjectWorkPending}, nil
+	}
+	respond := func(_ context.Context, answer extension.ProjectRiskAnswer) (extension.ProjectWorkDecision, error) {
+		responses++
+		if answer.Actor != "alice" || answer.ProjectID != "project-1" || answer.RiskID != "risk-1" || answer.RiskRevision != "r1" || answer.Answer != "效率更高且成本更低" || answer.IdempotencyKey == "" {
+			t.Fatalf("untrusted risk answer: %+v", answer)
+		}
+		return extension.ProjectWorkDecision{State: extension.ProjectWorkProceed, DecisionID: "decision-1"}, nil
+	}
+	ask := func(card core.HITLCard) (core.HITLResponse, error) {
+		questions++
+		if card.Question == "" || len(card.Choices) != 3 || !card.Blocked {
+			t.Fatalf("risk decision did not become a blocking HITL card: %+v", card)
+		}
+		return core.HITLResponse{ChoiceID: "option-0"}, nil
+	}
+	decision, err := resolveProjectWorkDecision(context.Background(), "alice", "project-1", "run-1", "conversation.run", preflight, respond, ask)
+	if err != nil || decision.State != extension.ProjectWorkProceed || preflights != 1 || responses != 1 || questions != 1 {
+		t.Fatalf("gate did not wait for clarification: decision=%+v preflight=%d responses=%d questions=%d err=%v", decision, preflights, responses, questions, err)
+	}
+}
+
+func TestProjectWorkGateNeverTreatsPendingOrMismatchedProceedAsApproval(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		decision extension.ProjectWorkDecision
+	}{
+		{name: "pending", decision: extension.ProjectWorkDecision{State: extension.ProjectWorkPending}},
+		{name: "unbound proceed", decision: extension.ProjectWorkDecision{State: extension.ProjectWorkProceed}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			preflight := func(context.Context, string, string) (extension.ProjectWorkDecision, error) { return tc.decision, nil }
+			decision, err := resolveProjectWorkDecision(context.Background(), "alice", "project-1", "run-1", "conversation.run", preflight, nil, nil)
+			if err == nil || decision.State == extension.ProjectWorkProceed {
+				t.Fatalf("unsafe decision proceeded: %+v err=%v", decision, err)
+			}
+		})
+	}
+}
+
+func TestRunContextUsesHITLForProjectRiskBeforeContinuing(t *testing.T) {
+	rc, st := newExecEnv(t)
+	rc.run.ID = "run-risk-gate"
+	rc.run.Status = core.RunRunning
+	if err := st.CreateRun(&rc.run); err != nil {
+		t.Fatal(err)
+	}
+	rc.h = &runHandle{hitlCh: make(chan hitlAnswer, 1)}
+	rc.h.hitlCh <- hitlAnswer{resp: core.HITLResponse{ChoiceID: "option-0"}}
+	rc.mgr.SetProjectWorkAuthorization(func(context.Context, string, core.ID) (bool, error) { return true, nil })
+	responded := false
+	rc.mgr.SetProjectWorkGate(
+		func(context.Context, string, string) (extension.ProjectWorkDecision, error) {
+			return extension.ProjectWorkDecision{State: extension.ProjectWorkClarify, RiskID: "risk-1", RiskRevision: "r1", Question: "效率具体指什么？", Options: []string{"效率更高"}}, nil
+		},
+		func(_ context.Context, answer extension.ProjectRiskAnswer) (extension.ProjectWorkDecision, error) {
+			responded = answer.Answer == "效率更高"
+			return extension.ProjectWorkDecision{State: extension.ProjectWorkProceed, DecisionID: "decision-1"}, nil
+		},
+	)
+	if err := rc.requireProjectWorkDecision("conversation.run"); err != nil || !responded {
+		t.Fatalf("run did not wait for committed human clarification: responded=%t err=%v", responded, err)
+	}
+	cards, err := st.ActiveHITLCards("alice", rc.run.ConversationID)
+	if err != nil || len(cards) != 1 || cards[0].Question != "效率具体指什么？" {
+		t.Fatalf("clarification card was not persisted: %+v %v", cards, err)
+	}
+}
+
+func TestConversationRunWaitsForProjectSourceAnalysisBeforeRunner(t *testing.T) {
+	rc, st := newExecEnv(t)
+	conv, err := st.GetConversation("alice", rc.run.ConversationID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	message := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "要求效率和成本都低于人工"}}}
+	if err := st.AppendMessageWithRequest(&message, "risk-request"); err != nil {
+		t.Fatal(err)
+	}
+	runner := &projectGateRunner{}
+	rc.mgr.runner = runner
+	rc.mgr.profiles = profiles.New(st)
+	rc.mgr.active = map[string]*runHandle{}
+	rc.mgr.SetProjectWorkAuthorization(func(context.Context, string, core.ID) (bool, error) { return true, nil })
+	ready := atomic.Bool{}
+	rc.mgr.SetProjectWorkGate(
+		func(context.Context, string, string) (extension.ProjectWorkDecision, error) {
+			if !ready.Load() {
+				return extension.ProjectWorkDecision{State: extension.ProjectWorkPending}, nil
+			}
+			return extension.ProjectWorkDecision{State: extension.ProjectWorkProceed, DecisionID: "decision-1"}, nil
+		},
+		func(context.Context, extension.ProjectRiskAnswer) (extension.ProjectWorkDecision, error) {
+			return extension.ProjectWorkDecision{State: extension.ProjectWorkProceed, DecisionID: "decision-1"}, nil
+		},
+	)
+	run, err := rc.mgr.StartConversationRun(conv, message, nil, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(80 * time.Millisecond)
+	if runner.calls.Load() != 0 {
+		t.Fatal("model runner started while project sources were still pending")
+	}
+	linked, err := st.MessageByID(string(message.ID))
+	if err != nil || linked.RunID != run.ID {
+		t.Fatalf("committed user source was not bound before preflight: %+v %v", linked, err)
+	}
+	ready.Store(true)
+	deadline := time.Now().Add(2 * time.Second)
+	for runner.calls.Load() == 0 && time.Now().Before(deadline) {
+		time.Sleep(10 * time.Millisecond)
+	}
+	if runner.calls.Load() != 1 {
+		t.Fatalf("runner calls=%d, want one after committed proceed", runner.calls.Load())
+	}
+}
+
+func TestProjectAssistantTextIsNotStreamedBeforeFinalRiskDecision(t *testing.T) {
+	rc, _ := newExecEnv(t)
+	rc.run.ID = "run-buffered-project-reply"
+	enableProjectGateForTest(rc.mgr)
+	if err := rc.Emit(core.EvMessageDelta, core.MessageDeltaPayload{Index: 0, Text: "未经风险检查的正文"}); err != nil {
+		t.Fatal(err)
+	}
+	rc.mgr.events.mu.Lock()
+	buffered := len(rc.mgr.events.buf[string(rc.run.ConversationID)])
+	rc.mgr.events.mu.Unlock()
+	if buffered != 0 {
+		t.Fatalf("project reply leaked %d message delta(s) before final decision", buffered)
+	}
+}
+
+func TestFinalProjectReplyWaitsForRiskDecisionBeforeCommit(t *testing.T) {
+	rc, st := newExecEnv(t)
+	conv, err := st.GetConversation("alice", rc.run.ConversationID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	message := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "项目问题"}}}
+	if err := st.AppendMessageWithRequest(&message, "final-risk-request"); err != nil {
+		t.Fatal(err)
+	}
+	rc.mgr.runner = &projectReplyRunner{}
+	rc.mgr.profiles = profiles.New(st)
+	rc.mgr.active = map[string]*runHandle{}
+	rc.mgr.SetProjectWorkAuthorization(func(context.Context, string, core.ID) (bool, error) { return true, nil })
+	finalReady := atomic.Bool{}
+	rc.mgr.SetProjectWorkGate(
+		func(_ context.Context, _ string, action string) (extension.ProjectWorkDecision, error) {
+			if action == "assistant.final" && !finalReady.Load() {
+				return extension.ProjectWorkDecision{State: extension.ProjectWorkPending}, nil
+			}
+			return extension.ProjectWorkDecision{State: extension.ProjectWorkProceed, DecisionID: "decision-" + action}, nil
+		},
+		func(context.Context, extension.ProjectRiskAnswer) (extension.ProjectWorkDecision, error) {
+			return extension.ProjectWorkDecision{State: extension.ProjectWorkProceed, DecisionID: "decision-risk"}, nil
+		},
+	)
+	if _, err := rc.mgr.StartConversationRun(conv, message, nil, nil); err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(80 * time.Millisecond)
+	messages, err := st.ListMessages(string(conv.ID))
+	if err != nil || len(messages) != 1 {
+		t.Fatalf("assistant reply committed before final risk decision: %+v %v", messages, err)
+	}
+	finalReady.Store(true)
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		messages, err = st.ListMessages(string(conv.ID))
+		if err == nil && len(messages) == 2 {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	if len(messages) != 2 || messages[1].Role != core.RoleAssistant || messages[1].Parts[0].Text != "最终项目答复" {
+		t.Fatalf("approved final reply was not committed once: %+v", messages)
+	}
+}

internal/runmgr/project_work_retrieval_test.go

@@ -36,7 +36,7 @@ func TestProjectRetrievalRecordsEachActualAuthorizedHit(t *testing.T) {
 		{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)
+	enableProjectGateForTest(mgr)
 	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 {

internal/runmgr/project_work_tool_result_test.go

@@ -14,7 +14,7 @@ func TestProjectToolResultRecordsActualOutputBeforeModelUse(t *testing.T) {
 		t.Fatal(err)
 	}
 	rc.run = run
-	rc.mgr.SetProjectWorkObservationEnabled(true)
+	enableProjectGateForTest(rc.mgr)
 	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)

internal/runmgr/runmgr.go

@@ -70,6 +70,8 @@ type Manager struct {
 	projectEvidenceReader         func(context.Context, extension.ProjectEvidenceRequest) (extension.ProjectEvidence, error)
 	projectSourceRechecker        func(context.Context, extension.ProjectContextSourceRecheckRequest) (extension.ProjectContextSourceRecheckResult, error)
 	projectWorkObservationEnabled bool
+	projectWorkPreflight          projectWorkPreflight
+	projectRiskRespond            projectRiskRespond
 	// commands EXT-14：命令即工具的投影与分发源（nil = 未接线，命令不进工具清单）。
 	commands CommandSource
 	indexer  core.DocumentIndexer // 可选索引入口（nil = store 整文档单块）
@@ -302,6 +304,43 @@ func (m *Manager) SetProjectWorkObservationEnabled(enabled bool) {
 	m.projectWorkObservationEnabled = enabled
 }
 
+// SetProjectWorkGate installs the domain-owned preflight and risk-answer
+// callbacks. Both are required; a partial gate stays disabled.
+func (m *Manager) SetProjectWorkGate(preflight projectWorkPreflight, respond projectRiskRespond) {
+	if preflight == nil || respond == nil {
+		m.projectWorkPreflight, m.projectRiskRespond = nil, nil
+		m.projectWorkObservationEnabled = false
+		return
+	}
+	m.projectWorkPreflight, m.projectRiskRespond = preflight, respond
+	m.projectWorkObservationEnabled = true
+}
+
+func (c *runCtx) requireProjectWorkDecision(action string) error {
+	if !c.mgr.projectWorkObservationEnabled || c.run.ProjectID == "" {
+		return nil
+	}
+	for {
+		decision, err := resolveProjectWorkDecision(c.ctx, string(c.run.Owner), string(c.run.ProjectID), string(c.run.ID), action,
+			c.mgr.projectWorkPreflight, c.mgr.projectRiskRespond, c.AskHITL)
+		if errors.Is(err, errProjectWorkPending) {
+			select {
+			case <-time.After(250 * time.Millisecond):
+				continue
+			case <-c.ctx.Done():
+				return c.ctx.Err()
+			}
+		}
+		if err != nil {
+			return err
+		}
+		if decision.State != extension.ProjectWorkProceed {
+			return errors.New("project work action did not receive a committed proceed decision")
+		}
+		return c.mgr.setRunStatus(&c.run, core.RunRunning, c.run.Phase)
+	}
+}
+
 // 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) {
@@ -773,6 +812,11 @@ func (c *runCtx) Emit(typ core.StreamEventType, payload any) error {
 	if err := c.mgr.authorizeRunWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
 		return err
 	}
+	if c.mgr.projectWorkObservationEnabled && c.run.ProjectID != "" && (typ == core.EvMessageDelta || typ == core.EvThinkingDelta) {
+		// Project reply text is released only through the committed message after
+		// the complete run-source manifest receives a domain risk decision.
+		return nil
+	}
 	_, err := c.mgr.emit(c.run.ConversationID, c.run.ID, typ, payload)
 	return err
 }
@@ -1172,6 +1216,9 @@ func (c *runCtx) Retrieve(q core.RetrievalQuery) ([]core.RetrievalHit, error) {
 			return nil, err
 		}
 	}
+	if err := c.requireProjectWorkDecision("model.consume.retrieval"); err != nil {
+		return nil, err
+	}
 	return hits, nil
 }
 
@@ -1328,6 +1375,9 @@ func (c *runCtx) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult {
 		}); err != nil {
 			return core.ToolCallResult{CallID: call.CallID, Status: "error", Output: "project tool result could not be recorded"}
 		}
+		if err := c.requireProjectWorkDecision("model.consume.tool-result"); err != nil {
+			return core.ToolCallResult{CallID: call.CallID, Status: "error", Output: err.Error()}
+		}
 	}
 	// 超长截断并转工件引用（4.10）。
 	limit := c.prof.Context.MaxToolResultChars
@@ -1647,6 +1697,26 @@ func (m *Manager) startClaimedRun(conv core.Conversation, userMessage core.Messa
 		m.releaseConversation(conv.ID)
 		return run, err
 	}
+	if m.projectWorkObservationEnabled {
+		if userMessage.ID == "" {
+			m.releaseConversation(conv.ID)
+			return run, errors.New("project run requires a committed user message source")
+		}
+		var linkErr error
+		if m.accountEffectiveAuthorization != nil {
+			linkErr = m.st.SetMessageRunIDAccountChecked(string(run.Owner), string(userMessage.ID), string(run.ID), m.accountAllowImplicitLocal)
+		} else {
+			linkErr = m.st.SetMessageRunID(string(userMessage.ID), string(run.ID))
+		}
+		if linkErr != nil {
+			run.Status, run.Phase = core.RunFailed, core.PhaseDone
+			run.Error = &core.ErrorPart{Code: "run.project_source_link_failed", Message: linkErr.Error()}
+			_ = m.st.UpdateRun(&run)
+			m.releaseConversation(conv.ID)
+			return run, linkErr
+		}
+		userMessage.RunID = run.ID
+	}
 	ctx, cancel := context.WithCancel(context.Background())
 	h := &runHandle{owner: run.Owner, cancel: cancel, hitlCh: make(chan hitlAnswer, 1), started: time.Now()}
 	m.mu.Lock()
@@ -1735,6 +1805,10 @@ func (m *Manager) execute(rc *runCtx, runner runtime.Runner) {
 		return
 	}
 	m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunPhase, core.RunPhasePayload{Phase: core.PhaseThinking})
+	if err := rc.requireProjectWorkDecision("conversation.run"); err != nil {
+		m.finishRun(rc, core.RunFailed, core.Usage{}, &core.ErrorPart{Code: "run.project_work_blocked", Message: err.Error()})
+		return
+	}
 
 	execErr := func() (err error) {
 		defer func() {
@@ -1804,6 +1878,22 @@ func (m *Manager) finishRun(rc *runCtx, status core.RunStatus, usage core.Usage,
 			msg.VariantIndex = index
 		}
 	}
+	if msg != nil && status == core.RunCompleted && m.projectWorkObservationEnabled && rc.run.ProjectID != "" {
+		gateErr := rc.stageAssistantWorkSources(msg.Parts)
+		if gateErr == nil {
+			gateErr = rc.requireProjectWorkDecision("assistant.final")
+		}
+		if gateErr != nil {
+			status = core.RunFailed
+			errPart = &core.ErrorPart{Code: "run.project_reply_blocked", Message: gateErr.Error()}
+			msg.Parts = []core.Part{{Type: core.PartError, Err: errPart}}
+		}
+	}
+	// HITL/preflight may have moved the persisted run through waiting/running.
+	// Restore the intended terminal state immediately before the fenced commit.
+	rc.run.Status = status
+	rc.run.Phase = core.PhaseDone
+	rc.run.Error = errPart
 	updated, fenced, err := m.settleRunWithFence(rc, msg)
 	if err != nil {
 		log.Printf("[runmgr] settle run %s: %v", rc.run.ID, err)