feat(app): add composition root, stageflow, ContextUpdate, and SSE transport

Move construction and document OnOutbox out of server, give HTTP/CLI a shared task use case, and block already-cleared import/event arch rules.

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

.doc/system-design/module-boundaries.yaml

@@ -56,12 +56,36 @@ imports:
     - storage
   db: []
   docduty: []
-  app: []
+  app:
+    - agent
+    - app/chatflow
+    - app/docnotify
+    - app/stageflow
+    - app/uow
+    - artifactvault
+    - auth
+    - authz
+    - bizdev
+    - config
+    - coordinator
+    - docgov
+    - intent
+    - notification
+    - observability
+    - projectinstance
+    - projectworkflow
+    - taskrun
+    - workflowdef
+    - workspace
   app/chatflow:
     - bizdev
     - coordinator
     - ctxmgr
     - projectinstance
+  app/docnotify:
+    - docgov
+    - docrevision
+    - notification
   app/docflow:
     - app/uow
     - docgov
@@ -69,6 +93,8 @@ imports:
     - docrevision
     - events
   app/uow: []
+  app/stageflow:
+    - bizdev
   docquery:
     - docgov
     - docrevision
@@ -105,8 +131,10 @@ imports:
   replyexport: []
   server:
     - agent
+    - app
     - app/chatflow
     - app/docflow
+    - app/stageflow
     - app/uow
     - auth
     - authz
@@ -128,6 +156,7 @@ imports:
     - replyexport
     - squad
     - stream
+    - stream/transport
     - taskrun
     - upload
     - workflowdef
@@ -142,6 +171,8 @@ imports:
     - workspace
   storage: []
   stream: []
+  stream/transport:
+    - stream
   taskrun:
     - app/uow
     - artifactvault
@@ -256,7 +287,7 @@ event_producers:
 
 event_consumers:
   outbox.notify:
-    - server
+    - app/docnotify
     - notification
 
 unwrap_owners:
@@ -443,3 +474,9 @@ baseline_counts:
   sql_write: 5
   path: 16
   event: 0
+
+# U12: kinds already at zero production findings fail --mode block.
+# sql_write/path keep dated exceptions until their owner milestones clear them.
+block_kinds:
+  - import
+  - event

.github/workflows/ci.yml

@@ -45,6 +45,9 @@ jobs:
           TEST_DATABASE_URL: postgres://app:ci_secret_db_pass@localhost:5432/agent_system_ci?sslmode=disable
         run: go test -short -timeout 20m ./...
 
+      - name: arch-check (block cleared kinds)
+        run: ./scripts/arch-check.sh --mode block
+
   backup-restore-drill:
     runs-on: ubuntu-latest
     services:

cmd/bizdev-cli/main.go

@@ -10,6 +10,7 @@ import (
 	"strings"
 	"time"
 
+	"chat-agent-system/internal/app/stageflow"
 	"chat-agent-system/internal/bizdev"
 	"chat-agent-system/internal/chat"
 	"chat-agent-system/internal/config"
@@ -240,7 +241,7 @@ func main() {
 		return
 	}
 
-	res, err := svc.ExecuteTask(context.Background(), bizdev.TaskRunRequest{
+	res, err := stageflow.New(svc).ExecuteTask(context.Background(), bizdev.TaskRunRequest{
 		TaskKey:         *taskKey,
 		ProjectName:     *project,
 		CustomerName:    *customer,
@@ -347,7 +348,7 @@ func handleStageRun(
 		ctx, _ = stream.WithListener(ctx, verboseMode, listener)
 	}
 
-	stageRes, err := svc.ExecuteStage(ctx, bizdev.StageRunRequest{
+	stageRes, err := stageflow.New(svc).ExecuteStage(ctx, bizdev.StageRunRequest{
 		Stage:           bizdev.MaturityStage(stageName),
 		ProjectName:     project,
 		CustomerName:    customer,

cmd/server/main.go

@@ -10,7 +10,7 @@ import (
 	"syscall"
 	"time"
 
-	"chat-agent-system/internal/bizdev"
+	"chat-agent-system/internal/app"
 	"chat-agent-system/internal/config"
 	"chat-agent-system/internal/server"
 )
@@ -38,12 +38,16 @@ func main() {
 		cfg.AgentBackend = *backend
 	}
 
-	svc, err := bizdev.NewService(cfg)
+	rt, err := app.New(cfg)
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "业务服务初始化失败: %v\n", err)
 		os.Exit(1)
 	}
-	srv := server.NewServer(cfg, svc)
+	if err := rt.Start(context.Background()); err != nil {
+		fmt.Fprintf(os.Stderr, "后台 worker 启动失败: %v\n", err)
+		os.Exit(1)
+	}
+	srv := server.NewFromRuntime(rt)
 
 	addr := fmt.Sprintf("%s:%d", cfg.HTTPHost, cfg.HTTPPort)
 	fmt.Println("==================================================================")

internal/app/app.go

@@ -0,0 +1,305 @@
+package app
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"sync"
+	"time"
+
+	"chat-agent-system/internal/agent"
+	"chat-agent-system/internal/app/chatflow"
+	"chat-agent-system/internal/app/docnotify"
+	"chat-agent-system/internal/app/stageflow"
+	"chat-agent-system/internal/app/uow"
+	"chat-agent-system/internal/artifactvault"
+	"chat-agent-system/internal/auth"
+	"chat-agent-system/internal/authz"
+	"chat-agent-system/internal/bizdev"
+	"chat-agent-system/internal/config"
+	"chat-agent-system/internal/coordinator"
+	"chat-agent-system/internal/docgov"
+	"chat-agent-system/internal/intent"
+	"chat-agent-system/internal/notification"
+	"chat-agent-system/internal/observability"
+	"chat-agent-system/internal/projectinstance"
+	"chat-agent-system/internal/projectworkflow"
+	"chat-agent-system/internal/taskrun"
+	"chat-agent-system/internal/workflowdef"
+	"chat-agent-system/internal/workspace"
+)
+
+const defaultReminderPollInterval = 15 * time.Minute
+
+// Runtime is the unique composition root. cmd/server, tests and workers
+// construct it once; HTTP adapters must not re-wire domain services.
+type Runtime struct {
+	Cfg        *config.Config
+	Biz        *bizdev.Service
+	Engine     *coordinator.CoordinatorEngine
+	Auth       *auth.Service
+	PW         *projectworkflow.Service
+	TaskRuns   *taskrun.Manager
+	Authorizer *authz.Authorizer
+	Authz      *authz.Service
+	WFDef      *workflowdef.Service
+	Inst       *projectinstance.Service
+	Notify     *notification.Service
+	Resolver   *workspace.Resolver
+	WSCap      *workspace.Capability
+	Commands   *taskrun.Commands
+	Rec        intent.Recognizer
+	Stageflow  *stageflow.UseCase
+	DocNotify  *docnotify.Subscriber
+	ChatLog    chatflow.ChatLog
+	Assembler  chatflow.ContextAssembler
+
+	mu             sync.Mutex
+	running        bool
+	subscribers    int
+	starts         int
+	notifyBound    bool
+	hookBound      bool
+	reminderCancel context.CancelFunc
+}
+
+func New(cfg *config.Config) (*Runtime, error) {
+	return Assemble(cfg, nil)
+}
+
+// Assemble wires domain services. svc may be injected by tests.
+func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) {
+	if cfg == nil {
+		cfg, _ = config.Load("")
+	}
+	if cfg != nil {
+		observability.InitLogger(cfg.Env)
+	}
+	if svc == nil {
+		var err error
+		svc, err = bizdev.NewService(cfg)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	rec := intent.NewRecognizer()
+	dialogueBackend := agent.NewBackend(cfg.AgentBackend, cfg)
+	coordEngine := coordinator.NewCoordinatorEngineWithDialogue(svc, dialogueBackend, cfg.DataDir)
+	pool := svc.Pool()
+	stores, storeErr := auth.NewStores(context.Background(), pool, cfg)
+	if storeErr != nil {
+		return nil, fmt.Errorf("auth store: %w", storeErr)
+	}
+	authSvc := auth.NewService(stores.Auth, stores.Sessions, cfg)
+	pwSvc := svc.ProjectWorkflow()
+	if pwSvc == nil {
+		pwSvc = projectworkflow.NewService(projectworkflow.NewStore(nil), nil)
+	}
+
+	authzSvc := authz.NewService(pool)
+	wfDefSvc := workflowdef.NewService(pool, svc.CAS())
+	instStore := projectinstance.NewStore(pool)
+	var instSvc *projectinstance.Service
+	if wfDefSvc.Available() {
+		docChecker := projectinstance.NewDocGovChecker(svc.DocGov())
+		instSvc = projectinstance.NewService(instStore, wfDefSvc.Store(), pwSvc.Store(), docChecker)
+		if pool != nil {
+			instSvc.SetActiveMemberLookup(projectinstance.NewPGMemberLookup(pool))
+		}
+	}
+	taskMgr := taskrun.NewManager(pwSvc, svc.ContextManager())
+	if pool != nil {
+		taskMgr.SetStore(taskrun.NewPGStore(pool))
+	}
+	wsCap := workspace.NewCapability(cfg.DataDir, workspace.NewLeaseTable())
+	cmds := &taskrun.Commands{
+		Ctx:             svc.ContextManager(),
+		PW:              pwSvc,
+		Env:             cfg.Env,
+		StorageIdentity: "env:" + cfg.Env,
+	}
+	if pool != nil {
+		cmds.UoW = uow.NewPG(pool)
+		cmds.Fav = &artifactvault.Ingest{Env: cfg.Env}
+	}
+	taskMgr.SetCommands(cmds)
+
+	stages := stageflow.New(svc)
+	coordEngine.SetStageTaskRunner(stages)
+	if instSvc != nil && instSvc.Available() {
+		coordEngine.SetInstanceService(instSvc)
+	}
+
+	rt := &Runtime{
+		Cfg:        cfg,
+		Biz:        svc,
+		Engine:     coordEngine,
+		Auth:       authSvc,
+		PW:         pwSvc,
+		TaskRuns:   taskMgr,
+		Authorizer: authz.NewAuthorizer(pool),
+		Authz:      authzSvc,
+		WFDef:      wfDefSvc,
+		Inst:       instSvc,
+		Notify:     notification.NewService(pool),
+		Resolver:   workspace.NewResolver(cfg.DataDir, pool),
+		WSCap:      wsCap,
+		Commands:   cmds,
+		Rec:        rec,
+		Stageflow:  stages,
+	}
+	rt.DocNotify = &docnotify.Subscriber{Notify: rt.Notify, Docs: svc.DocGov()}
+	rt.bindDocumentNotify()
+	rt.bindWorkflowApprovalHook()
+	return rt, nil
+}
+
+func (r *Runtime) bindDocumentNotify() {
+	if r == nil || r.notifyBound {
+		return
+	}
+	if r.Biz != nil && r.Biz.DocRevision() != nil {
+		if docnotify.Bind(r.Biz.DocRevision(), r.DocNotify) {
+			r.subscribers++
+		}
+	}
+	r.notifyBound = true
+}
+
+func (r *Runtime) bindWorkflowApprovalHook() {
+	if r == nil || r.hookBound || r.Biz == nil || r.Biz.DocGov() == nil || r.Inst == nil || !r.Inst.Available() {
+		r.hookBound = true
+		return
+	}
+	r.Biz.DocGov().AddOnApprovedHook(r.onWorkflowDocumentApproved)
+	r.hookBound = true
+	r.subscribers++
+}
+
+func (r *Runtime) onWorkflowDocumentApproved(ctx context.Context, doc *docgov.DocumentRecord) {
+	if doc == nil || doc.TaskKey == "" || doc.ProjectID <= 0 || r.Inst == nil {
+		return
+	}
+	result, err := r.Inst.HandleDocumentApproved(ctx, doc.ProjectID, doc.TaskKey)
+	if err != nil || result == nil || !result.StageReady {
+		return
+	}
+	if r.Notify == nil || !r.Notify.Available() {
+		return
+	}
+	if r.PW == nil {
+		return
+	}
+	project, err := r.PW.GetProject(ctx, doc.ProjectID, 0)
+	if err != nil || project == nil || project.CreatorUserID <= 0 {
+		return
+	}
+	_, _ = r.Notify.EmitStageReadyToAdvance(ctx, project.CreatorUserID, notification.StageReadyPayload{
+		ProjectID:          doc.ProjectID,
+		WorkflowInstanceID: result.InstanceID,
+		StageInstanceID:    result.StageInstanceID,
+		StageKey:           result.StageKey,
+		StageName:          result.StageName,
+		ConfigRevision:     result.ConfigRevision,
+		RequiredDocCount:   result.RequiredDocCount,
+		ApprovedDocCount:   result.ApprovedDocCount,
+	})
+}
+
+// Start launches background workers. Repeat calls on a running runtime are no-ops.
+func (r *Runtime) Start(_ context.Context) error {
+	if r == nil {
+		return fmt.Errorf("app: nil runtime")
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.running {
+		return nil
+	}
+	r.bindDocumentNotify()
+	r.bindWorkflowApprovalHook()
+	ctx, cancel := context.WithCancel(context.Background())
+	r.reminderCancel = cancel
+	go r.runReminderLoop(ctx)
+	r.running = true
+	r.starts++
+	if r.subscribers == 0 {
+		r.subscribers = 1
+	}
+	return nil
+}
+
+// Stop cancels workers. A later Start recreates the reminder loop without
+// double-registering document subscribers.
+func (r *Runtime) Stop(_ context.Context) error {
+	if r == nil {
+		return nil
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if !r.running {
+		return nil
+	}
+	if r.reminderCancel != nil {
+		r.reminderCancel()
+		r.reminderCancel = nil
+	}
+	r.running = false
+	return nil
+}
+
+func (r *Runtime) SubscriberCount() int {
+	if r == nil {
+		return 0
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return r.subscribers
+}
+
+func (r *Runtime) StartCount() int {
+	if r == nil {
+		return 0
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return r.starts
+}
+
+func (r *Runtime) Running() bool {
+	if r == nil {
+		return false
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return r.running
+}
+
+func (r *Runtime) runReminderLoop(ctx context.Context) {
+	r.runStageAdvanceReminders(ctx)
+	ticker := time.NewTicker(defaultReminderPollInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			r.runStageAdvanceReminders(ctx)
+		}
+	}
+}
+
+func (r *Runtime) runStageAdvanceReminders(ctx context.Context) {
+	if r.Notify == nil || !r.Notify.Available() {
+		return
+	}
+	n, err := r.Notify.ProcessStageAdvanceReminders(ctx)
+	if err != nil {
+		slog.Warn("stage advance reminder scan failed", "error", err)
+		return
+	}
+	if n > 0 {
+		slog.Info("stage advance 24h reminders sent", "count", n)
+	}
+}

internal/app/app_test.go

@@ -0,0 +1,44 @@
+package app
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+func TestStartStopRestartIdempotent(t *testing.T) {
+	rt := &Runtime{}
+	ctx := context.Background()
+	if err := rt.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if err := rt.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if rt.StartCount() != 1 {
+		t.Fatalf("duplicate Start spawned workers: starts=%d", rt.StartCount())
+	}
+	if got := rt.SubscriberCount(); got != 1 {
+		t.Fatalf("subscribers=%d", got)
+	}
+	if !rt.Running() {
+		t.Fatal("expected running")
+	}
+	if err := rt.Stop(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if rt.Running() {
+		t.Fatal("expected stopped")
+	}
+	if err := rt.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if rt.StartCount() != 2 {
+		t.Fatalf("restart did not relaunch worker: starts=%d", rt.StartCount())
+	}
+	if rt.SubscriberCount() != 1 {
+		t.Fatalf("restart re-registered subscribers: %d", rt.SubscriberCount())
+	}
+	_ = rt.Stop(ctx)
+	time.Sleep(10 * time.Millisecond)
+}

internal/app/chatflow/assembler.go

@@ -70,16 +70,22 @@ func (a *Assembler) Assemble(ctx context.Context, p Principal, req AssembleReque
 	if restored == nil {
 		return userCtx, nil
 	}
-	if restored.ProjectID > 0 && userCtx.Project.ID == "" {
-		userCtx.Project.ID = fmt.Sprintf("%d", restored.ProjectID)
+	upd := coordinator.ContextUpdate{
+		ExpectedVersion:   userCtx.Version,
+		BlackboardFill:    restored.Facts,
+		SetRecentDialogue: true,
+		RecentDialogue:    authorizedTurns(restored),
 	}
-	if userCtx.BlackboardFacts == nil {
-		userCtx.BlackboardFacts = make(map[string]string)
+	if restored.ProjectID > 0 {
+		upd.ProjectIDIfEmpty = fmt.Sprintf("%d", restored.ProjectID)
 	}
-	for k, v := range restored.Facts {
-		if _, exists := userCtx.BlackboardFacts[k]; !exists {
-			userCtx.BlackboardFacts[k] = v
+	_ = coordinator.ApplyContextUpdate(userCtx, upd)
+	return userCtx, nil
 }
+
+func authorizedTurns(restored *ctxmgr.RestoredSessionContext) []coordinator.DialogueTurn {
+	if restored == nil {
+		return nil
 	}
 	turns := make([]coordinator.DialogueTurn, 0, len(restored.Messages))
 	records := make([]ctxmgr.ChatMessageRecord, 0, len(restored.Messages))
@@ -101,6 +107,5 @@ func (a *Assembler) Assemble(ctx context.Context, p Principal, req AssembleReque
 			turns = append(turns, coordinator.DialogueTurn{Role: m.Role, Content: m.Content})
 		}
 	}
-	userCtx.RecentDialogue = turns
-	return userCtx, nil
+	return turns
 }

internal/app/docnotify/subscriber.go

@@ -0,0 +1,94 @@
+package docnotify
+
+import (
+	"context"
+	"fmt"
+
+	"chat-agent-system/internal/docgov"
+	"chat-agent-system/internal/docrevision"
+	"chat-agent-system/internal/notification"
+)
+
+// DocumentLookup resolves a document for recipient and project fill-in.
+type DocumentLookup interface {
+	GetDocument(ctx context.Context, docID string) (*docgov.DocumentRecord, error)
+}
+
+// Subscriber is the unique consumer for document OnOutbox notification events.
+// Bind once from the app composition root; do not assign OnOutbox in server.
+type Subscriber struct {
+	Notify notificationEmitter
+	Docs   DocumentLookup
+}
+
+type notificationEmitter interface {
+	Available() bool
+	EmitDocumentEvent(ctx context.Context, eventType, notifType, title, body, dedupeKey string, recipientUserID int64, payload notification.DocumentEventPayload) (bool, error)
+}
+
+// Bind assigns the subscriber exactly once. Repeat calls keep the first handler.
+func Bind(rev *docrevision.Service, sub *Subscriber) bool {
+	if rev == nil || sub == nil || rev.OnOutbox != nil {
+		return false
+	}
+	rev.OnOutbox = sub.Handle
+	return true
+}
+
+// Handle maps one document outbox event onto a permission-safe inbox notification.
+func (s *Subscriber) Handle(ctx context.Context, eventType string, payload map[string]any) error {
+	if s == nil || s.Notify == nil || !s.Notify.Available() {
+		return nil
+	}
+	docID, _ := payload["doc_id"].(string)
+	projectID, _ := payload["project_id"].(int64)
+	risk, _ := payload["risk"].(string)
+	var recipient int64
+	if s.Docs != nil && docID != "" {
+		if doc, err := s.Docs.GetDocument(ctx, docID); err == nil && doc != nil && doc.AuthorID > 0 {
+			recipient = doc.AuthorID
+			projectID = doc.ProjectID
+		}
+	}
+	if recipient == 0 {
+		return nil
+	}
+	p := notification.DocumentEventPayload{
+		ProjectID: projectID, DocID: docID, RiskLevel: risk,
+	}
+	switch eventType {
+	case notification.EventDocAgentCandidateReady:
+		if v, ok := payload["candidate_id"].(int64); ok {
+			p.CandidateID = v
+		}
+		_, err := s.Notify.EmitDocumentEvent(ctx, eventType, notification.TypeDocAgentCandidateReady,
+			"智能体变更待确认", "文档有高风险智能体候选变更，请确认后生效。",
+			fmt.Sprintf("doc_candidate:%s:%d:%d", docID, p.CandidateID, recipient), recipient, p)
+		return err
+	case notification.EventDocAgentVersionCommitted:
+		if v, ok := payload["version_no"].(int); ok {
+			p.VersionNo = v
+		}
+		_, err := s.Notify.EmitDocumentEvent(ctx, eventType, notification.TypeDocAgentVersionCommitted,
+			"智能体版本已生效", "新的智能体文档版本已提交。",
+			fmt.Sprintf("doc_agent_version:%s:%d:%d", docID, p.VersionNo, recipient), recipient, p)
+		return err
+	case notification.EventDocDownstreamImpact:
+		impacted, _ := payload["impacted_doc_id"].(string)
+		p.ImpactedDocID = impacted
+		_, err := s.Notify.EmitDocumentEvent(ctx, eventType, notification.TypeDocDownstreamImpact,
+			"下游文档受影响", "上游文档变更，请复核关联文档。",
+			fmt.Sprintf("doc_impact:%s:%s:%d", docID, impacted, recipient), recipient, p)
+		return err
+	case notification.EventDocChangeReportDegraded:
+		if v, ok := payload["report_id"].(int64); ok {
+			p.ReportID = v
+		}
+		_, err := s.Notify.EmitDocumentEvent(ctx, eventType, notification.TypeDocChangeReportDegraded,
+			"变更报告未完整评估", "变更报告降级，受控文档暂不可送审。",
+			fmt.Sprintf("doc_report_degraded:%s:%d:%d", docID, p.ReportID, recipient), recipient, p)
+		return err
+	default:
+		return nil
+	}
+}

internal/app/docnotify/subscriber_test.go

@@ -0,0 +1,73 @@
+package docnotify
+
+import (
+	"context"
+	"testing"
+
+	"chat-agent-system/internal/docgov"
+	"chat-agent-system/internal/docrevision"
+	"chat-agent-system/internal/notification"
+)
+
+type stubNotify struct {
+	available bool
+	calls     int
+	lastType  string
+}
+
+func (s *stubNotify) Available() bool { return s != nil && s.available }
+
+func (s *stubNotify) EmitDocumentEvent(_ context.Context, eventType, _, _, _, _ string, _ int64, _ notification.DocumentEventPayload) (bool, error) {
+	s.calls++
+	s.lastType = eventType
+	return true, nil
+}
+
+type stubDocs struct {
+	doc *docgov.DocumentRecord
+}
+
+func (s stubDocs) GetDocument(_ context.Context, _ string) (*docgov.DocumentRecord, error) {
+	return s.doc, nil
+}
+
+func TestBindOnce(t *testing.T) {
+	rev := &docrevision.Service{}
+	sub := &Subscriber{Notify: &stubNotify{available: true}}
+	if !Bind(rev, sub) {
+		t.Fatal("first bind")
+	}
+	if Bind(rev, sub) {
+		t.Fatal("second bind must be a no-op")
+	}
+	if rev.OnOutbox == nil {
+		t.Fatal("hook missing")
+	}
+}
+
+func TestHandleCandidateReady(t *testing.T) {
+	n := &stubNotify{available: true}
+	sub := &Subscriber{
+		Notify: n,
+		Docs:   stubDocs{doc: &docgov.DocumentRecord{AuthorID: 9, ProjectID: 4}},
+	}
+	if err := sub.Handle(context.Background(), notification.EventDocAgentCandidateReady, map[string]any{
+		"doc_id": "d1", "candidate_id": int64(3),
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if n.calls != 1 || n.lastType != notification.EventDocAgentCandidateReady {
+		t.Fatalf("calls=%d type=%s", n.calls, n.lastType)
+	}
+}
+
+func TestHandleSkipsUnknownAndNoRecipient(t *testing.T) {
+	n := &stubNotify{available: true}
+	sub := &Subscriber{Notify: n}
+	if err := sub.Handle(context.Background(), "document.unknown", map[string]any{"doc_id": "d1"}); err != nil {
+		t.Fatal(err)
+	}
+	if n.calls != 0 {
+		t.Fatalf("unknown event must not emit, calls=%d", n.calls)
+	}
+}

internal/app/stageflow/usecase.go

@@ -0,0 +1,40 @@
+package stageflow
+
+import (
+	"context"
+	"errors"
+
+	"chat-agent-system/internal/bizdev"
+)
+
+var ErrClosed = errors.New("stageflow: executor not configured")
+
+// Executor is the unique application owner for task and stage runs.
+// HTTP, CLI and coordinator must call this instead of bizdev directly.
+type Executor interface {
+	ExecuteTask(ctx context.Context, req bizdev.TaskRunRequest) (*bizdev.TaskRunResult, error)
+	ExecuteStage(ctx context.Context, req bizdev.StageRunRequest) (*bizdev.StageRunResult, error)
+}
+
+// UseCase adapts bizdev orchestration without moving domain tables.
+type UseCase struct {
+	Biz Executor
+}
+
+func New(biz Executor) *UseCase {
+	return &UseCase{Biz: biz}
+}
+
+func (u *UseCase) ExecuteTask(ctx context.Context, req bizdev.TaskRunRequest) (*bizdev.TaskRunResult, error) {
+	if u == nil || u.Biz == nil {
+		return nil, ErrClosed
+	}
+	return u.Biz.ExecuteTask(ctx, req)
+}
+
+func (u *UseCase) ExecuteStage(ctx context.Context, req bizdev.StageRunRequest) (*bizdev.StageRunResult, error) {
+	if u == nil || u.Biz == nil {
+		return nil, ErrClosed
+	}
+	return u.Biz.ExecuteStage(ctx, req)
+}

internal/app/stageflow/usecase_test.go

@@ -0,0 +1,46 @@
+package stageflow
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"chat-agent-system/internal/bizdev"
+)
+
+type stubBiz struct {
+	tasks  int
+	stages int
+}
+
+func (s *stubBiz) ExecuteTask(_ context.Context, req bizdev.TaskRunRequest) (*bizdev.TaskRunResult, error) {
+	s.tasks++
+	return &bizdev.TaskRunResult{Success: true, ArtifactTitle: req.TaskKey}, nil
+}
+
+func (s *stubBiz) ExecuteStage(_ context.Context, req bizdev.StageRunRequest) (*bizdev.StageRunResult, error) {
+	s.stages++
+	return &bizdev.StageRunResult{Success: true, StageTitle: string(req.Stage)}, nil
+}
+
+func TestUseCaseDelegates(t *testing.T) {
+	biz := &stubBiz{}
+	httpUC := New(biz)
+	cliUC := New(biz)
+	if _, err := httpUC.ExecuteTask(context.Background(), bizdev.TaskRunRequest{TaskKey: "intake"}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := cliUC.ExecuteStage(context.Background(), bizdev.StageRunRequest{Stage: bizdev.StageV01_Triage}); err != nil {
+		t.Fatal(err)
+	}
+	if biz.tasks != 1 || biz.stages != 1 {
+		t.Fatalf("shared executor counts task=%d stage=%d", biz.tasks, biz.stages)
+	}
+}
+
+func TestUseCaseClosed(t *testing.T) {
+	var u *UseCase
+	if _, err := u.ExecuteTask(context.Background(), bizdev.TaskRunRequest{}); !errors.Is(err, ErrClosed) {
+		t.Fatalf("got %v", err)
+	}
+}

internal/archtest/arch_test.go

@@ -26,7 +26,25 @@ func TestArchitectureBoundaries(t *testing.T) {
 	if ev.Failed() {
 		t.Fatal(ev.Report())
 	}
-	if os.Getenv("ARCHTEST_STRICT") != "" && len(ev.Excepted) > 0 {
+	if os.Getenv("ARCHTEST_STRICT") == "" {
+		return
+	}
+	if len(b.BlockKinds) == 0 {
+		if len(ev.Excepted) > 0 {
 			t.Fatalf("strict/block mode: %d baseline exceptions remain\n%s", len(ev.Excepted), ev.Report())
 		}
+		return
+	}
+	for _, f := range ev.Excepted {
+		for _, kind := range b.BlockKinds {
+			if f.Kind == kind {
+				t.Fatalf("strict/block mode: cleared kind %s still has exception %s\n%s", kind, f.Key(), ev.Report())
+			}
+		}
+	}
+	for _, kind := range b.BlockKinds {
+		if ev.Counts[kind] > 0 {
+			t.Fatalf("strict/block mode: cleared kind %s count=%d\n%s", kind, ev.Counts[kind], ev.Report())
+		}
+	}
 }

internal/archtest/boundaries.go

@@ -27,6 +27,7 @@ type Boundaries struct {
 	UnwrapOwners    []string            `yaml:"unwrap_owners"`
 	Exceptions      []Exception         `yaml:"exceptions"`
 	BaselineCounts  map[string]int      `yaml:"baseline_counts"`
+	BlockKinds      []string            `yaml:"block_kinds"`
 }
 
 type Exception struct {

internal/chat/session.go

@@ -81,7 +81,11 @@ func NewSession(id string, userCtx *coordinator.UserProjectContext, rec intent.R
 	}
 	// 若上下文尚未挂起预期且当前已具有阶段，则初始化默认选项草稿
 	if userCtx != nil && userCtx.PendingExpectation == nil && userCtx.Project.CurrentStage != "" {
-		userCtx.PendingExpectation = coordinator.BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name)
+		_ = coordinator.ApplyContextUpdate(userCtx, coordinator.ContextUpdate{
+			ExpectedVersion:       userCtx.Version,
+			SetPendingExpectation: true,
+			PendingExpectation:    coordinator.BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name),
+		})
 	}
 
 	sess := &Session{
@@ -115,7 +119,11 @@ func (s *Session) SyncRecentDialogue() {
 		}
 		turns = append(turns, coordinator.DialogueTurn{Role: m.Role, Content: m.Content})
 	}
-	s.UserCtx.RecentDialogue = turns
+	_ = coordinator.ApplyContextUpdate(s.UserCtx, coordinator.ContextUpdate{
+		ExpectedVersion:   s.UserCtx.Version,
+		SetRecentDialogue: true,
+		RecentDialogue:    turns,
+	})
 
 	// 若发生了截断且存在早期消息，同步计算或加载滚动摘要
 	if truncated && len(early) > 0 && s.CtxMgr != nil {
@@ -416,11 +424,17 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str
 	}
 
 	if coordRes.ClearExpectation {
-		s.UserCtx.PendingExpectation = nil
+		_ = coordinator.ApplyContextUpdate(s.UserCtx, coordinator.ContextUpdate{
+			ExpectedVersion: s.UserCtx.Version, SetPendingExpectation: true,
+		})
 	} else if coordRes.NextExpectation != nil {
-		s.UserCtx.PendingExpectation = coordRes.NextExpectation
+		_ = coordinator.ApplyContextUpdate(s.UserCtx, coordinator.ContextUpdate{
+			ExpectedVersion: s.UserCtx.Version, SetPendingExpectation: true, PendingExpectation: coordRes.NextExpectation,
+		})
 	} else if recRes.Branch == intent.BranchExpectationHit {
-		s.UserCtx.PendingExpectation = nil
+		_ = coordinator.ApplyContextUpdate(s.UserCtx, coordinator.ContextUpdate{
+			ExpectedVersion: s.UserCtx.Version, SetPendingExpectation: true,
+		})
 	}
 	// 若为 BranchNewIntentPreempt 且未产生新预期，原挂起预期在焦点底静默保留，不阻拦用户
 

internal/coordinator/context.go

@@ -8,6 +8,7 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"chat-agent-system/internal/bizdev"
@@ -76,6 +77,8 @@ type DocumentSummaryReader func(ctx context.Context, target string) (*ReadableDo
 
 // UserProjectContext 用户项目级全量统一上下文 (对话系统与执行系统共同消费)
 type UserProjectContext struct {
+	mu                 sync.Mutex
+	Version            int64                         `json:"version"`
 	SnapshotID         string                        `json:"snapshot_id"`
 	AssembledAt        time.Time                     `json:"assembled_at"`
 	User               UserInfo                      `json:"user"`

internal/coordinator/context_update.go

@@ -0,0 +1,144 @@
+package coordinator
+
+import (
+	"errors"
+
+	"chat-agent-system/internal/intent"
+)
+
+var (
+	ErrNilContext    = errors.New("coordinator: nil user project context")
+	ErrStaleContext  = errors.New("coordinator: stale context version")
+	ErrInvalidUpdate = errors.New("coordinator: empty context update")
+)
+
+// ContextUpdate is the only typed write against UserProjectContext.
+// Callers must supply ExpectedVersion from a prior read; a mismatch fails closed.
+type ContextUpdate struct {
+	ExpectedVersion int64
+
+	BlackboardFill map[string]string
+	BlackboardSet  map[string]string
+
+	SetPendingExpectation bool
+	PendingExpectation    *intent.PendingExpectation
+	ExpirePending         bool
+
+	SetInteractiveCard bool
+	InteractiveCard    *intent.InteractiveCardPayload
+	TickInteractive    bool
+	DismissCardStatus  string // cancelled | resolved | expired; then nil
+
+	SetPersonalizedGuidance bool
+	PersonalizedGuidance    string
+
+	SetRecentDialogue bool
+	RecentDialogue    []DialogueTurn
+
+	ProjectIDIfEmpty string
+}
+
+// ApplyContextUpdate applies one versioned patch. Concurrent writers with the
+// same ExpectedVersion: exactly one succeeds, the rest get ErrStaleContext.
+func ApplyContextUpdate(ctx *UserProjectContext, u ContextUpdate) error {
+	if ctx == nil {
+		return ErrNilContext
+	}
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+	if u.ExpectedVersion != ctx.Version {
+		return ErrStaleContext
+	}
+	applied := false
+	if len(u.BlackboardFill) > 0 {
+		if ctx.BlackboardFacts == nil {
+			ctx.BlackboardFacts = make(map[string]string, len(u.BlackboardFill))
+		}
+		for k, v := range u.BlackboardFill {
+			if _, exists := ctx.BlackboardFacts[k]; !exists {
+				ctx.BlackboardFacts[k] = v
+			}
+		}
+		applied = true
+	}
+	if len(u.BlackboardSet) > 0 {
+		if ctx.BlackboardFacts == nil {
+			ctx.BlackboardFacts = make(map[string]string, len(u.BlackboardSet))
+		}
+		for k, v := range u.BlackboardSet {
+			ctx.BlackboardFacts[k] = v
+		}
+		applied = true
+	}
+	if u.ExpirePending && ctx.PendingExpectation != nil && !ctx.PendingExpectation.ExpiresAt.IsZero() {
+		// caller decides freshness; apply just clears
+		ctx.PendingExpectation = nil
+		applied = true
+	}
+	if u.SetPendingExpectation {
+		ctx.PendingExpectation = u.PendingExpectation
+		applied = true
+	}
+	if u.TickInteractive && ctx.InteractiveCard != nil {
+		ctx.InteractiveCard.TTLTurns--
+		if ctx.InteractiveCard.TTLTurns <= 0 || ctx.InteractiveCard.CurrentStatus != "active" {
+			ctx.InteractiveCard.CurrentStatus = "expired"
+			ctx.InteractiveCard = nil
+		}
+		applied = true
+	}
+	if u.DismissCardStatus != "" && ctx.InteractiveCard != nil {
+		ctx.InteractiveCard.CurrentStatus = u.DismissCardStatus
+		ctx.InteractiveCard = nil
+		applied = true
+	}
+	if u.SetInteractiveCard {
+		ctx.InteractiveCard = u.InteractiveCard
+		applied = true
+	}
+	if u.SetPersonalizedGuidance {
+		ctx.PersonalizedGuidance = u.PersonalizedGuidance
+		applied = true
+	}
+	if u.SetRecentDialogue {
+		ctx.RecentDialogue = u.RecentDialogue
+		applied = true
+	}
+	if u.ProjectIDIfEmpty != "" && ctx.Project.ID == "" {
+		ctx.Project.ID = u.ProjectIDIfEmpty
+		applied = true
+	}
+	if !applied {
+		return ErrInvalidUpdate
+	}
+	ctx.Version++
+	return nil
+}
+
+func (c *UserProjectContext) applyPending(exp *intent.PendingExpectation) {
+	if c == nil {
+		return
+	}
+	_ = ApplyContextUpdate(c, ContextUpdate{ExpectedVersion: c.Version, SetPendingExpectation: true, PendingExpectation: exp})
+}
+
+func (c *UserProjectContext) applyCard(card *intent.InteractiveCardPayload) {
+	if c == nil {
+		return
+	}
+	_ = ApplyContextUpdate(c, ContextUpdate{ExpectedVersion: c.Version, SetInteractiveCard: true, InteractiveCard: card})
+}
+
+func (c *UserProjectContext) dismissCard(status string) {
+	if c == nil {
+		return
+	}
+	_ = ApplyContextUpdate(c, ContextUpdate{ExpectedVersion: c.Version, DismissCardStatus: status})
+}
+
+func (c *UserProjectContext) applyBlackboard(key, value string) {
+	if c == nil {
+		return
+	}
+	_ = ApplyContextUpdate(c, ContextUpdate{ExpectedVersion: c.Version, BlackboardSet: map[string]string{key: value}})
+}

internal/coordinator/context_update_test.go

@@ -0,0 +1,55 @@
+package coordinator
+
+import (
+	"errors"
+	"sync"
+	"sync/atomic"
+	"testing"
+)
+
+func TestContextUpdateConcurrentNoLostUpdate(t *testing.T) {
+	ctx := &UserProjectContext{
+		Version:         1,
+		BlackboardFacts: map[string]string{"market": "国内"},
+	}
+	var ok, stale int32
+	var wg sync.WaitGroup
+	for i := 0; i < 2; i++ {
+		wg.Add(1)
+		go func(label string) {
+			defer wg.Done()
+			err := ApplyContextUpdate(ctx, ContextUpdate{
+				ExpectedVersion: 1,
+				BlackboardSet:   map[string]string{"market": label},
+			})
+			switch {
+			case err == nil:
+				atomic.AddInt32(&ok, 1)
+			case errors.Is(err, ErrStaleContext):
+				atomic.AddInt32(&stale, 1)
+			default:
+				t.Errorf("unexpected %v", err)
+			}
+		}("v" + string(rune('a'+i)))
+	}
+	wg.Wait()
+	if ok != 1 || stale != 1 {
+		t.Fatalf("ok=%d stale=%d version=%d facts=%v", ok, stale, ctx.Version, ctx.BlackboardFacts)
+	}
+	if ctx.Version != 2 {
+		t.Fatalf("version %d", ctx.Version)
+	}
+	if ctx.BlackboardFacts["market"] == "国内" {
+		t.Fatal("winner did not apply")
+	}
+}
+
+func TestContextUpdateFillDoesNotOverwrite(t *testing.T) {
+	ctx := &UserProjectContext{Version: 0, BlackboardFacts: map[string]string{"a": "1"}}
+	if err := ApplyContextUpdate(ctx, ContextUpdate{ExpectedVersion: 0, BlackboardFill: map[string]string{"a": "x", "b": "2"}}); err != nil {
+		t.Fatal(err)
+	}
+	if ctx.BlackboardFacts["a"] != "1" || ctx.BlackboardFacts["b"] != "2" {
+		t.Fatalf("%v", ctx.BlackboardFacts)
+	}
+}

internal/coordinator/engine.go

@@ -73,12 +73,36 @@ type CoordinationResult struct {
 // CoordinatorEngine L2 协调引擎
 type CoordinatorEngine struct {
 	bizdevSvc       bizdev.BizdevService
+	stageTasks      StageTaskRunner
 	instSvc         *projectinstance.Service
 	assembler       *ContextAssembler
 	dialogueBackend agent.Backend
 	dataDir         string
 }
 
+// StageTaskRunner is the application owner for ExecuteTask / ExecuteStage.
+// Wired from app/stageflow; tests may leave it nil and fall back to bizdevSvc.
+type StageTaskRunner interface {
+	ExecuteTask(ctx context.Context, req bizdev.TaskRunRequest) (*bizdev.TaskRunResult, error)
+	ExecuteStage(ctx context.Context, req bizdev.StageRunRequest) (*bizdev.StageRunResult, error)
+}
+
+func (e *CoordinatorEngine) SetStageTaskRunner(r StageTaskRunner) {
+	if e != nil {
+		e.stageTasks = r
+	}
+}
+
+func (e *CoordinatorEngine) tasks() StageTaskRunner {
+	if e == nil {
+		return nil
+	}
+	if e.stageTasks != nil {
+		return e.stageTasks
+	}
+	return e.bizdevSvc
+}
+
 // SetInstanceService 注入项目工作流实例服务（对话推进与实例门禁合流）。
 func (e *CoordinatorEngine) SetInstanceService(svc *projectinstance.Service) {
 	if e != nil {
@@ -118,14 +142,10 @@ func (e *CoordinatorEngine) ProcessIntent(
 
 	// 交互卡片 TTL 轮次消退与生命周期清理
 	if userCtx.InteractiveCard != nil {
-		userCtx.InteractiveCard.TTLTurns--
-		if userCtx.InteractiveCard.TTLTurns <= 0 || userCtx.InteractiveCard.CurrentStatus != "active" {
-			userCtx.InteractiveCard.CurrentStatus = "expired"
-			userCtx.InteractiveCard = nil
-		}
+		_ = ApplyContextUpdate(userCtx, ContextUpdate{ExpectedVersion: userCtx.Version, TickInteractive: true})
 	}
 	if userCtx.PendingExpectation != nil && !userCtx.PendingExpectation.ExpiresAt.IsZero() && time.Now().After(userCtx.PendingExpectation.ExpiresAt) {
-		userCtx.PendingExpectation = nil
+		userCtx.applyPending(nil)
 	}
 
 	top := recRes.TopIntent
@@ -355,10 +375,9 @@ func (e *CoordinatorEngine) handleDeclineExpectation(userCtx *UserProjectContext
 	if projName == "" {
 		projName = "当前项目"
 	}
-	userCtx.PendingExpectation = nil
+	userCtx.applyPending(nil)
 	if userCtx.InteractiveCard != nil {
-		userCtx.InteractiveCard.CurrentStatus = "cancelled"
-		userCtx.InteractiveCard = nil
+		userCtx.dismissCard("cancelled")
 	}
 
 	reply := fmt.Sprintf("收到，听您的！那我们先不急着推进。\n针对「%s」，您手头还有哪些细节、技术顾虑或商务考量想先聊聊？我随时待命为您解答。", projName)
@@ -418,7 +437,7 @@ func (e *CoordinatorEngine) handleAmbiguousConsult(ctx context.Context, cand *in
 	nextExp := BuildNextExpectation(currentStage, projName)
 	// 路由层：复杂自由任务画像命中时，先给即时初答，再附升级建议（复用挂起期望机制，含每会话冷却）
 	reply, nextExp = maybeAttachUpgradeHint(rawInput, userCtx, reply, nextExp)
-	userCtx.PendingExpectation = nextExp
+	userCtx.applyPending(nextExp)
 
 	return &CoordinationResult{
 		Status:          StatusAccepted,
@@ -436,7 +455,7 @@ func (e *CoordinatorEngine) handleQueryProgress(ctx context.Context, cand *inten
 	brief := userCtx.LatestAdvisorBrief
 	topic := cand.FilledSlots[intent.SlotQueryTopic]
 	nextExp := BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name)
-	userCtx.PendingExpectation = nextExp
+	userCtx.applyPending(nextExp)
 
 	// 1. 若为测试方案/难点/验证咨询，直接调取测试经理与参谋专业知识库解答
 	if topic == "testing_analysis" {
@@ -624,7 +643,7 @@ func (e *CoordinatorEngine) handleSupplementInfo(cand *intent.CandidateIntent, u
 
 	summaryKey := ""
 	if factKey != "" && factVal != "" {
-		userCtx.BlackboardFacts[factKey] = factVal
+		userCtx.applyBlackboard(factKey, factVal)
 		summaryKey = fmt.Sprintf("「%s」已更新为「%s」", factKey, factVal)
 		if factKey == "target_market" {
 			userCtx.Project.TargetMarket = factVal
@@ -657,7 +676,7 @@ func (e *CoordinatorEngine) handleSupplementInfo(cand *intent.CandidateIntent, u
 	}
 
 	nextExp := BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name)
-	userCtx.PendingExpectation = nextExp
+	userCtx.applyPending(nextExp)
 
 	var sb strings.Builder
 	sb.WriteString(fmt.Sprintf("已收到并为您记录最新信息：%s。\n", summaryKey))
@@ -757,7 +776,7 @@ func (e *CoordinatorEngine) handleAdvanceStage(ctx context.Context, cand *intent
 	if len(cand.PatchSlots) > 0 {
 		var patches []string
 		for k, v := range cand.PatchSlots {
-			userCtx.BlackboardFacts[k] = v
+			userCtx.applyBlackboard(k, v)
 			patches = append(patches, fmt.Sprintf("%s=%s", k, v))
 			if k == "target_market" {
 				userCtx.Project.TargetMarket = v
@@ -795,11 +814,11 @@ func (e *CoordinatorEngine) handleAdvanceStage(ctx context.Context, cand *intent
 		}, nil
 	}
 
-	if e.bizdevSvc == nil {
+	if e.tasks() == nil {
 		userCtx.Project.CurrentStage = targetStage
 		userCtx.AvailableActions = ComputeAvailableActions(targetStage, userCtx.User.ProjectRole)
 		nextExp := BuildNextExpectation(targetStage, userCtx.Project.Name)
-		userCtx.PendingExpectation = nextExp
+		userCtx.applyPending(nextExp)
 
 		reply := fmt.Sprintf("已为您模拟推进到阶段【%s】。", targetStage)
 		if patchSummary != "" {
@@ -816,7 +835,7 @@ func (e *CoordinatorEngine) handleAdvanceStage(ctx context.Context, cand *intent
 	}
 
 	// 3. 调用底层 L3 调度器执行阶段跃迁
-	stageRes, err := e.bizdevSvc.ExecuteStage(ctx, bizdev.StageRunRequest{
+	stageRes, err := e.tasks().ExecuteStage(ctx, bizdev.StageRunRequest{
 		Stage:        targetStage,
 		ProjectName:  userCtx.Project.Name,
 		CustomerName: userCtx.Project.CustomerName,
@@ -850,7 +869,7 @@ func (e *CoordinatorEngine) handleAdvanceStage(ctx context.Context, cand *intent
 	}
 
 	nextExp := BuildNextExpectation(targetStage, userCtx.Project.Name)
-	userCtx.PendingExpectation = nextExp
+	userCtx.applyPending(nextExp)
 
 	var sb strings.Builder
 	if patchSummary != "" {
@@ -889,9 +908,9 @@ func (e *CoordinatorEngine) handleTriggerTask(ctx context.Context, cand *intent.
 	}
 
 	nextExp := BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name)
-	userCtx.PendingExpectation = nextExp
+	userCtx.applyPending(nextExp)
 
-	if e.bizdevSvc == nil {
+	if e.tasks() == nil {
 		return &CoordinationResult{
 			Status:          StatusAccepted,
 			Intent:          cand.Intent,
@@ -910,7 +929,7 @@ func (e *CoordinatorEngine) handleTriggerTask(ctx context.Context, cand *intent.
 		fmt.Sscanf(userCtx.User.ID, "%d", &userID)
 	}
 
-	taskRes, err := e.bizdevSvc.ExecuteTask(ctx, bizdev.TaskRunRequest{
+	taskRes, err := e.tasks().ExecuteTask(ctx, bizdev.TaskRunRequest{
 		TaskKey:      taskKey,
 		ProjectName:  userCtx.Project.Name,
 		CustomerName: userCtx.Project.CustomerName,
@@ -1026,9 +1045,9 @@ func (e *CoordinatorEngine) handleTriggerAdhocTask(ctx context.Context, cand *in
 	taskKey := sanitizeAdhocTaskKey(title)
 
 	nextExp := BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name)
-	userCtx.PendingExpectation = nextExp
+	userCtx.applyPending(nextExp)
 
-	if e.bizdevSvc == nil {
+	if e.tasks() == nil {
 		return &CoordinationResult{
 			Status:          StatusAccepted,
 			Intent:          cand.Intent,
@@ -1058,7 +1077,7 @@ func (e *CoordinatorEngine) handleTriggerAdhocTask(ctx context.Context, cand *in
 		Content:  []byte(fmt.Sprintf("# 任务简报（用户原始诉求，权威等级：primary_evidence）\n\n%s\n", brief)),
 	}
 
-	taskRes, err := e.bizdevSvc.ExecuteTask(ctx, bizdev.TaskRunRequest{
+	taskRes, err := e.tasks().ExecuteTask(ctx, bizdev.TaskRunRequest{
 		TaskKey:      taskKey,
 		Title:        title,
 		ProjectName:  userCtx.Project.Name,
@@ -1141,7 +1160,7 @@ func (e *CoordinatorEngine) handleCasualChat(ctx context.Context, userCtx *UserP
 	}
 
 	nextExp := BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name)
-	userCtx.PendingExpectation = nextExp
+	userCtx.applyPending(nextExp)
 
 	nameSuffix := ""
 	if userCtx.User.Name != "" {
@@ -1157,7 +1176,7 @@ func (e *CoordinatorEngine) handleCasualChat(ctx context.Context, userCtx *UserP
 
 	// 路由层：复杂自由任务画像命中时，先给即时初答，再附升级建议（复用挂起期望机制，含每会话冷却）
 	reply, nextExp = maybeAttachUpgradeHint(rawInput, userCtx, reply, nextExp)
-	userCtx.PendingExpectation = nextExp
+	userCtx.applyPending(nextExp)
 
 	return &CoordinationResult{
 		Status:          StatusAccepted,
@@ -1254,8 +1273,8 @@ func (e *CoordinatorEngine) handleDocGovReject(ctx context.Context, cand *intent
 		}, nil
 	}
 
-	userCtx.InteractiveCard = resolved.Card
-	userCtx.PendingExpectation = resolved.Expectation
+	userCtx.applyCard(resolved.Card)
+	userCtx.applyPending(resolved.Expectation)
 
 	return &CoordinationResult{
 		Status:          StatusAccepted,
@@ -1299,7 +1318,7 @@ func (e *CoordinatorEngine) handleQueryRejectReason(ctx context.Context, cand *i
 
 	resolved := ResolveRejectCard(role, "", docName, docName, "", reason, RejectUtteranceQuery)
 	if resolved.Card != nil {
-		userCtx.InteractiveCard = resolved.Card
+		userCtx.applyCard(resolved.Card)
 	}
 
 	return &CoordinationResult{
@@ -1442,8 +1461,8 @@ func (e *CoordinatorEngine) handleDocGovApprove(ctx context.Context, cand *inten
 			},
 		}
 
-		userCtx.InteractiveCard = cardPayload
-		userCtx.PendingExpectation = nextExp
+		userCtx.applyCard(cardPayload)
+		userCtx.applyPending(nextExp)
 
 		reply := fmt.Sprintf("已为您起草针对《%s》的终审批准确认单。终审批准将使文档正式归档并永久锁定，请审阅下方卡片并在确认无误后点击确认或回复‘确认’。", docName)
 
@@ -1559,10 +1578,9 @@ func (e *CoordinatorEngine) handleDocGovProposalDecide(ctx context.Context, cand
 		}
 
 		if userCtx.InteractiveCard != nil {
-			userCtx.InteractiveCard.CurrentStatus = "resolved"
-			userCtx.InteractiveCard = nil
+			userCtx.dismissCard("resolved")
 		}
-		userCtx.PendingExpectation = nil
+		userCtx.applyPending(nil)
 
 		rollback := docgov.StateDrafting
 		if userCtx.InteractiveCard != nil && strings.Contains(userCtx.InteractiveCard.ImpactRisk, "reviewing") {
@@ -1587,10 +1605,9 @@ func (e *CoordinatorEngine) handleDocGovProposalDecide(ctx context.Context, cand
 
 	if targetAction == "confirm_final_approve" {
 		if userCtx.InteractiveCard != nil {
-			userCtx.InteractiveCard.CurrentStatus = "resolved"
-			userCtx.InteractiveCard = nil
+			userCtx.dismissCard("resolved")
 		}
-		userCtx.PendingExpectation = nil
+		userCtx.applyPending(nil)
 
 		return &CoordinationResult{
 			Status:           StatusAccepted,
@@ -1608,10 +1625,9 @@ func (e *CoordinatorEngine) handleDocGovProposalDecide(ctx context.Context, cand
 
 	if targetAction == "cancel" {
 		if userCtx.InteractiveCard != nil {
-			userCtx.InteractiveCard.CurrentStatus = "cancelled"
-			userCtx.InteractiveCard = nil
+			userCtx.dismissCard("cancelled")
 		}
-		userCtx.PendingExpectation = nil
+		userCtx.applyPending(nil)
 
 		return &CoordinationResult{
 			Status:           StatusAccepted,

internal/coordinator/engine_run.go

@@ -50,7 +50,7 @@ func (e *CoordinatorEngine) handleRunResume(_ context.Context, cand *intent.Cand
 	}
 
 	if isRunActionConfirmed(userCtx, "run.resume") {
-		userCtx.PendingExpectation = nil
+		userCtx.applyPending(nil)
 		return &CoordinationResult{
 			Status:         StatusAccepted,
 			Intent:         intent.IntentRunResume,
@@ -96,7 +96,7 @@ func (e *CoordinatorEngine) handleRunResume(_ context.Context, cand *intent.Cand
 
 func (e *CoordinatorEngine) handleRunRestart(_ context.Context, cand *intent.CandidateIntent, userCtx *UserProjectContext) (*CoordinationResult, error) {
 	if isRunActionConfirmed(userCtx, "run.restart") {
-		userCtx.PendingExpectation = nil
+		userCtx.applyPending(nil)
 		taskKey := cand.FilledSlots["task_key"]
 		if taskKey == "" && userCtx.TaskState != nil {
 			taskKey = userCtx.TaskState.ActiveTask

internal/server/handlers.go

@@ -253,7 +253,7 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) {
 
 	// 1. 如果指定了 stage，则执行成熟度阶段编排
 	if req.Stage != "" {
-		stageRes, err := s.svc.ExecuteStage(r.Context(), bizdev.StageRunRequest{
+		stageRes, err := s.taskUseCase().ExecuteStage(r.Context(), bizdev.StageRunRequest{
 			Stage:           bizdev.MaturityStage(req.Stage),
 			ProjectName:     req.ProjectName,
 			CustomerName:    req.CustomerName,
@@ -301,7 +301,7 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) {
 	if req.TaskKey == "" {
 		req.TaskKey = "intake"
 	}
-	taskRes, err := s.svc.ExecuteTask(r.Context(), bizdev.TaskRunRequest{
+	taskRes, err := s.taskUseCase().ExecuteTask(r.Context(), bizdev.TaskRunRequest{
 		TaskKey:         req.TaskKey,
 		ProjectName:     req.ProjectName,
 		CustomerName:    req.CustomerName,

internal/server/server.go

@@ -10,10 +10,9 @@ import (
 	"sync"
 	"time"
 
-	"chat-agent-system/internal/agent"
+	"chat-agent-system/internal/app"
 	"chat-agent-system/internal/app/chatflow"
-	"chat-agent-system/internal/app/uow"
-	"chat-agent-system/internal/artifactvault"
+	"chat-agent-system/internal/app/stageflow"
 	"chat-agent-system/internal/auth"
 	"chat-agent-system/internal/authz"
 	"chat-agent-system/internal/bizdev"
@@ -23,7 +22,6 @@ import (
 	"chat-agent-system/internal/ctxmgr"
 	"chat-agent-system/internal/intent"
 	"chat-agent-system/internal/notification"
-	"chat-agent-system/internal/observability"
 	"chat-agent-system/internal/projectinstance"
 	"chat-agent-system/internal/projectworkflow"
 	"chat-agent-system/internal/taskrun"
@@ -33,6 +31,7 @@ import (
 
 // Server 提供基于 HTTP RESTful 与 SSE 的商务与研发参谋接口服务
 type Server struct {
+	rt         *app.Runtime
 	cfg        *config.Config
 	svc        *bizdev.Service
 	pwSvc      *projectworkflow.Service
@@ -52,156 +51,55 @@ type Server struct {
 	cmds       *taskrun.Commands
 	chatLog    chatflow.ChatLog
 	assembler  chatflow.ContextAssembler
+	stageflow  *stageflow.UseCase
 	httpServer *http.Server
 }
 
-// NewServer 构造函数
-func NewServer(cfg *config.Config, svc *bizdev.Service) *Server {
-	if cfg == nil {
-		cfg, _ = config.Load("")
+// NewFromRuntime attaches the HTTP adapter to an already-wired composition root.
+func NewFromRuntime(rt *app.Runtime) *Server {
+	if rt == nil {
+		panic("server: nil runtime")
 	}
-	if cfg != nil {
-		observability.InitLogger(cfg.Env)
-	}
-	if svc == nil {
-		var err error
-		svc, err = bizdev.NewService(cfg)
-		if err != nil {
-			panic("server: bizdev service init failed: " + err.Error())
-		}
-	}
-
-	rec := intent.NewRecognizer()
-	dialogueBackend := agent.NewBackend(cfg.AgentBackend, cfg)
-	coordEngine := coordinator.NewCoordinatorEngineWithDialogue(svc, dialogueBackend, cfg.DataDir)
-	pool := svc.Pool()
-	stores, storeErr := auth.NewStores(context.Background(), pool, cfg)
-	if storeErr != nil {
-		panic("server: auth store init failed: " + storeErr.Error())
-	}
-	authSvc := auth.NewService(stores.Auth, stores.Sessions, cfg)
-	pwSvc := svc.ProjectWorkflow()
-	if pwSvc == nil {
-		pwSvc = projectworkflow.NewService(projectworkflow.NewStore(nil), nil)
-	}
-
-	authzSvc := authz.NewService(pool)
-	wfDefSvc := workflowdef.NewService(pool, svc.CAS())
-	instStore := projectinstance.NewStore(pool)
-	var instSvc *projectinstance.Service
-	if wfDefSvc.Available() {
-		docChecker := projectinstance.NewDocGovChecker(svc.DocGov())
-		instSvc = projectinstance.NewService(instStore, wfDefSvc.Store(), pwSvc.Store(), docChecker)
-		if pool != nil {
-			instSvc.SetActiveMemberLookup(projectinstance.NewPGMemberLookup(pool))
-		}
-	}
-	taskMgr := taskrun.NewManager(pwSvc, svc.ContextManager())
-	if pool != nil {
-		taskMgr.SetStore(taskrun.NewPGStore(pool))
-	}
-	wsCap := workspace.NewCapability(cfg.DataDir, workspace.NewLeaseTable())
-	cmds := &taskrun.Commands{
-		Ctx:             svc.ContextManager(),
-		PW:              pwSvc,
-		Env:             cfg.Env,
-		StorageIdentity: "env:" + cfg.Env,
-	}
-	if pool != nil {
-		cmds.UoW = uow.NewPG(pool)
-		cmds.Fav = &artifactvault.Ingest{Env: cfg.Env}
-	}
-	taskMgr.SetCommands(cmds)
-
 	s := &Server{
-		cfg:        cfg,
-		svc:        svc,
-		pwSvc:      pwSvc,
-		rec:        rec,
-		engine:     coordEngine,
-		authSvc:    authSvc,
+		rt:         rt,
+		cfg:        rt.Cfg,
+		svc:        rt.Biz,
+		pwSvc:      rt.PW,
+		rec:        rt.Rec,
+		engine:     rt.Engine,
+		authSvc:    rt.Auth,
 		sessions:   make(map[string]*chat.Session),
-		taskRuns:   taskMgr,
-		authorizer: authz.NewAuthorizer(pool),
-		authzSvc:   authzSvc,
-		wfDefSvc:   wfDefSvc,
-		instSvc:    instSvc,
-		notifySvc:  notification.NewService(pool),
-		resolver:   workspace.NewResolver(cfg.DataDir, pool),
-		wsCap:      wsCap,
-		cmds:       cmds,
+		taskRuns:   rt.TaskRuns,
+		authorizer: rt.Authorizer,
+		authzSvc:   rt.Authz,
+		wfDefSvc:   rt.WFDef,
+		instSvc:    rt.Inst,
+		notifySvc:  rt.Notify,
+		resolver:   rt.Resolver,
+		wsCap:      rt.WSCap,
+		cmds:       rt.Commands,
+		stageflow:  rt.Stageflow,
 	}
-	s.chatLog = chatflow.NewAdapter(svc.ContextManager(), s.resolveStageAccess, s.failClosedAuthz())
+	s.chatLog = chatflow.NewAdapter(s.svc.ContextManager(), s.resolveStageAccess, s.failClosedAuthz())
 	s.assembler = chatflow.NewAssembler(coordinator.NewContextAssembler(), s.chatLog)
-	if instSvc != nil && instSvc.Available() {
-		svc.SetStageAccessGate(newBizdevStageGate(instSvc, pwSvc))
-		coordEngine.SetInstanceService(instSvc)
+	rt.ChatLog = s.chatLog
+	rt.Assembler = s.assembler
+	if s.instSvc != nil && s.instSvc.Available() && s.svc != nil {
+		s.svc.SetStageAccessGate(newBizdevStageGate(s.instSvc, s.pwSvc))
 	}
-	s.registerWorkflowApprovalHook()
-	s.startNotificationReminderLoop()
 	s.initTaskRunExecutors()
-	if svc != nil && svc.DocRevision() != nil && s.notifySvc != nil {
-		rev := svc.DocRevision()
-		rev.OnOutbox = func(ctx context.Context, eventType string, payload map[string]any) error {
-			if !s.notifySvc.Available() {
-				return nil
-			}
-			docID, _ := payload["doc_id"].(string)
-			projectID, _ := payload["project_id"].(int64)
-			risk, _ := payload["risk"].(string)
-			var recipient int64
-			if s.svc.DocGov() != nil {
-				if doc, err := s.svc.DocGov().GetDocument(ctx, docID); err == nil && doc != nil && doc.AuthorID > 0 {
-					recipient = doc.AuthorID
-					projectID = doc.ProjectID
-				}
-			}
-			if recipient == 0 {
-				return nil
-			}
-			p := notification.DocumentEventPayload{
-				ProjectID: projectID, DocID: docID, RiskLevel: risk,
-			}
-			switch eventType {
-			case notification.EventDocAgentCandidateReady:
-				if v, ok := payload["candidate_id"].(int64); ok {
-					p.CandidateID = v
-				}
-				_, err := s.notifySvc.EmitDocumentEvent(ctx, eventType, notification.TypeDocAgentCandidateReady,
-					"智能体变更待确认", "文档有高风险智能体候选变更，请确认后生效。",
-					fmt.Sprintf("doc_candidate:%s:%d:%d", docID, p.CandidateID, recipient), recipient, p)
-				return err
-			case notification.EventDocAgentVersionCommitted:
-				if v, ok := payload["version_no"].(int); ok {
-					p.VersionNo = v
-				}
-				_, err := s.notifySvc.EmitDocumentEvent(ctx, eventType, notification.TypeDocAgentVersionCommitted,
-					"智能体版本已生效", "新的智能体文档版本已提交。",
-					fmt.Sprintf("doc_agent_version:%s:%d:%d", docID, p.VersionNo, recipient), recipient, p)
-				return err
-			case notification.EventDocDownstreamImpact:
-				impacted, _ := payload["impacted_doc_id"].(string)
-				p.ImpactedDocID = impacted
-				_, err := s.notifySvc.EmitDocumentEvent(ctx, eventType, notification.TypeDocDownstreamImpact,
-					"下游文档受影响", "上游文档变更，请复核关联文档。",
-					fmt.Sprintf("doc_impact:%s:%s:%d", docID, impacted, recipient), recipient, p)
-				return err
-			case notification.EventDocChangeReportDegraded:
-				if v, ok := payload["report_id"].(int64); ok {
-					p.ReportID = v
-				}
-				_, err := s.notifySvc.EmitDocumentEvent(ctx, eventType, notification.TypeDocChangeReportDegraded,
-					"变更报告未完整评估", "变更报告降级，受控文档暂不可送审。",
-					fmt.Sprintf("doc_report_degraded:%s:%d:%d", docID, p.ReportID, recipient), recipient, p)
-				return err
-			default:
-				return nil
-			}
-		}
-	}
 	return s
 }
 
+// NewServer 构造函数（测试兼容：内部走 app.Assemble）
+func NewServer(cfg *config.Config, svc *bizdev.Service) *Server {
+	rt, err := app.Assemble(cfg, svc)
+	if err != nil {
+		panic("server: runtime init failed: " + err.Error())
+	}
+	return NewFromRuntime(rt)
+}
+
 // AuthService 获取认证服务
 func (s *Server) AuthService() *auth.Service {
 	return s.authSvc
@@ -426,6 +324,9 @@ func (s *Server) serveSPA(w http.ResponseWriter, r *http.Request) {
 
 // Start 启动 HTTP 守护进程
 func (s *Server) Start() error {
+	if s.rt != nil {
+		_ = s.rt.Start(context.Background())
+	}
 	addr := fmt.Sprintf("%s:%d", s.cfg.HTTPHost, s.cfg.HTTPPort)
 	s.httpServer = &http.Server{
 		Addr:              addr,
@@ -441,6 +342,9 @@ func (s *Server) Start() error {
 
 // Shutdown 优雅停机
 func (s *Server) Shutdown(ctx context.Context) error {
+	if s.rt != nil {
+		_ = s.rt.Stop(ctx)
+	}
 	if s.httpServer != nil {
 		return s.httpServer.Shutdown(ctx)
 	}
@@ -558,11 +462,18 @@ func (s *Server) getOrCreateSession(
 			BlackboardFacts:  make(map[string]string),
 			AvailableActions: coordinator.ComputeAvailableActions(bizdev.StageV01_Triage, u.ProjectRole),
 		}
+		if targetMarket != "" || deviceType != "" {
+			fill := map[string]string{}
 			if targetMarket != "" {
-			userCtx.BlackboardFacts["target_market"] = targetMarket
+				fill["target_market"] = targetMarket
 			}
 			if deviceType != "" {
-			userCtx.BlackboardFacts["device_type"] = deviceType
+				fill["device_type"] = deviceType
+			}
+			_ = coordinator.ApplyContextUpdate(userCtx, coordinator.ContextUpdate{
+				ExpectedVersion: userCtx.Version,
+				BlackboardFill:  fill,
+			})
 		}
 	}
 
@@ -586,10 +497,11 @@ func (s *Server) getOrCreateSession(
 		if restored.ProjectID > 0 {
 			sess.ProjectID = restored.ProjectID
 		}
-		for k, v := range restored.Facts {
-			if _, exists := userCtx.BlackboardFacts[k]; !exists {
-				userCtx.BlackboardFacts[k] = v
-			}
+		if len(restored.Facts) > 0 {
+			_ = coordinator.ApplyContextUpdate(userCtx, coordinator.ContextUpdate{
+				ExpectedVersion: userCtx.Version,
+				BlackboardFill:  restored.Facts,
+			})
 		}
 		if s.chatLog == nil {
 			for _, rm := range restored.Messages {
@@ -607,7 +519,11 @@ func (s *Server) getOrCreateSession(
 
 		if sess.UserID > 0 && sess.CtxMgr != nil {
 			if guidance, err := sess.CtxMgr.CompilePromptGuidance(context.Background(), sess.UserID); err == nil && guidance != "" {
-				userCtx.PersonalizedGuidance = guidance
+				_ = coordinator.ApplyContextUpdate(userCtx, coordinator.ContextUpdate{
+					ExpectedVersion:         userCtx.Version,
+					SetPersonalizedGuidance: true,
+					PersonalizedGuidance:    guidance,
+				})
 			}
 		}
 	}
@@ -641,6 +557,16 @@ func (s *Server) hydrateVisibleHistory(ctx context.Context, sess *chat.Session,
 	sess.SyncRecentDialogue()
 }
 
+func (s *Server) taskUseCase() *stageflow.UseCase {
+	if s == nil {
+		return stageflow.New(nil)
+	}
+	if s.stageflow != nil {
+		return s.stageflow
+	}
+	return stageflow.New(s.svc)
+}
+
 func (s *Server) corsMiddleware(next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		s.writeCORS(w, r)

internal/server/sse.go

@@ -13,6 +13,7 @@ import (
 	"chat-agent-system/internal/ctxmgr"
 	"chat-agent-system/internal/squad"
 	"chat-agent-system/internal/stream"
+	"chat-agent-system/internal/stream/transport"
 	"chat-agent-system/internal/taskrun"
 )
 
@@ -55,7 +56,7 @@ func extendSSEWriteDeadline(w http.ResponseWriter) {
 
 func writeSSEHeartbeat(w http.ResponseWriter, flusher http.Flusher) error {
 	extendSSEWriteDeadline(w)
-	if _, err := fmt.Fprintf(w, ": heartbeat %d\n\n", time.Now().Unix()); err != nil {
+	if err := (transport.Encoder{W: w}).Heartbeat(time.Now()); err != nil {
 		return err
 	}
 	flusher.Flush()
@@ -63,11 +64,7 @@ func writeSSEHeartbeat(w http.ResponseWriter, flusher http.Flusher) error {
 }
 
 func writeSSE(w http.ResponseWriter, flusher http.Flusher, ev stream.StreamEvent) {
-	data, err := json.Marshal(ev)
-	if err != nil {
-		return
-	}
-	fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Type, string(data))
+	_ = (transport.Encoder{W: w}).Write(ev)
 	flusher.Flush()
 }
 
@@ -236,14 +233,7 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) {
 	listener := func(ev stream.StreamEvent) {
 		writeMu.Lock()
 		defer writeMu.Unlock()
-
-		data, err := json.Marshal(ev)
-		if err != nil {
-			return
-		}
-		// 按照标准 SSE 协议规范发射: event: <type>\ndata: <json>\n\n
-		fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Type, string(data))
-		flusher.Flush()
+		writeSSE(w, flusher, ev)
 	}
 
 	clientTurnID := req.ClientTurnID
@@ -302,15 +292,12 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) {
 	if err != nil {
 		writeMu.Lock()
 		info := stream.ClassifyError(err, "chat_stream", requestID, sess.ID, "")
-		errEv := stream.StreamEvent{
+		writeSSE(w, flusher, stream.StreamEvent{
 			Type:      stream.EventError,
 			Timestamp: time.Now(),
 			Error:     info.UserMessage,
 			ErrorInfo: &info,
-		}
-		data, _ := json.Marshal(errEv)
-		fmt.Fprintf(w, "event: %s\ndata: %s\n\n", errEv.Type, string(data))
-		flusher.Flush()
+		})
 		writeMu.Unlock()
 	}
 }
@@ -408,12 +395,7 @@ func (s *Server) handleRunTaskStream(w http.ResponseWriter, r *http.Request) {
 	defer s.taskRuns.Unsubscribe(created.RunID, ch)
 
 	writeStored := func(stored taskrun.StoredEvent) {
-		data, marshalErr := json.Marshal(stored.Event)
-		if marshalErr != nil {
-			return
-		}
-		fmt.Fprintf(w, "id: %d\n", stored.Seq)
-		fmt.Fprintf(w, "event: %s\ndata: %s\n\n", stored.Event.Type, string(data))
+		_ = (transport.Encoder{W: w}).WriteWithID(stored.Seq, stored.Event)
 		flusher.Flush()
 	}
 

internal/server/task_run_handlers.go

@@ -708,7 +708,7 @@ func (s *Server) buildBizdevStageRequest(dto TaskRunRequestDTO, sessionID, clien
 func (s *Server) initTaskRunExecutors() {
 	s.taskRuns.SetExecutor(
 		func(ctx context.Context, _ string, req bizdev.TaskRunRequest, _ *stream.Streamer) (*bizdev.TaskRunResult, error) {
-			return s.svc.ExecuteTask(ctx, req)
+			return s.taskUseCase().ExecuteTask(ctx, req)
 		},
 		func(ctx context.Context, _ string, workDir string, _ *stream.Streamer) (*bizdev.TaskRunResult, error) {
 			res, err := s.svc.ResumeOrModifyWorkspace(ctx, squad.ModifyWorkflowRequest{WorkDir: workDir})
@@ -722,7 +722,7 @@ func (s *Server) initTaskRunExecutors() {
 			return out, err
 		},
 		func(ctx context.Context, _ string, req bizdev.StageRunRequest, _ *stream.Streamer) (*bizdev.TaskRunResult, error) {
-			stageRes, err := s.svc.ExecuteStage(ctx, req)
+			stageRes, err := s.taskUseCase().ExecuteStage(ctx, req)
 			if stageRes == nil {
 				return nil, err
 			}

internal/stream/transport/sse.go

@@ -0,0 +1,47 @@
+package transport
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"time"
+
+	"chat-agent-system/internal/stream"
+)
+
+// Encoder writes neutral SSE frames. It does not own retries, leases, or task state.
+type Encoder struct {
+	W io.Writer
+}
+
+func (e Encoder) Write(ev stream.StreamEvent) error {
+	return e.WriteWithID(0, ev)
+}
+
+func (e Encoder) WriteWithID(id int64, ev stream.StreamEvent) error {
+	if e.W == nil {
+		return nil
+	}
+	if ev.Timestamp.IsZero() {
+		ev.Timestamp = time.Now()
+	}
+	data, err := json.Marshal(ev)
+	if err != nil {
+		return err
+	}
+	if id > 0 {
+		if _, err = fmt.Fprintf(e.W, "id: %d\n", id); err != nil {
+			return err
+		}
+	}
+	_, err = fmt.Fprintf(e.W, "event: %s\ndata: %s\n\n", ev.Type, string(data))
+	return err
+}
+
+func (e Encoder) Heartbeat(now time.Time) error {
+	if e.W == nil {
+		return nil
+	}
+	_, err := fmt.Fprintf(e.W, ": heartbeat %d\n\n", now.Unix())
+	return err
+}

internal/stream/transport/sse_test.go

@@ -0,0 +1,35 @@
+package transport
+
+import (
+	"bytes"
+	"strings"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/stream"
+)
+
+func TestWriteNeutralFrame(t *testing.T) {
+	var buf bytes.Buffer
+	enc := Encoder{W: &buf}
+	if err := enc.Write(stream.StreamEvent{Type: stream.EventDone, Timestamp: time.Unix(1, 0)}); err != nil {
+		t.Fatal(err)
+	}
+	got := buf.String()
+	if !strings.Contains(got, "event: done") || !strings.Contains(got, "data: {") {
+		t.Fatalf("%q", got)
+	}
+	if strings.Contains(got, "retry:") {
+		t.Fatal("transport must not own retry")
+	}
+}
+
+func TestHeartbeat(t *testing.T) {
+	var buf bytes.Buffer
+	if err := (Encoder{W: &buf}).Heartbeat(time.Unix(10, 0)); err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(buf.String(), ": heartbeat 10") {
+		t.Fatalf("%q", buf.String())
+	}
+}

scripts/arch-check.sh

@@ -31,7 +31,7 @@ fi
 
 if [[ "${MODE}" == "block" ]]; then
   if ! ARCHTEST_STRICT=1 go test ./internal/archtest -count=1 -run TestArchitectureBoundaries; then
-    echo "arch-check block: baseline exceptions remain; only cleared rules may use --mode block" >&2
+    echo "arch-check block: a cleared kind in block_kinds still has findings" >&2
     exit 1
   fi
 fi