feat(intent): wire production QueryOwner and confirm dispatcher

Replace FakeActionOwner with QueryOwner+QueryGuard at the composition root; DispatchEligible reuses claim/execute/recover so confirm stays zero-write before dispatch and default tier remains legacy.
2026-09-22 21:44:56 +08:00
parent 364ed7f1a3
commit d019388f37
18 changed files with 728 additions and 62 deletions
@@ -0,0 +1,69 @@
+---
+title: "Track I1：多意图真实 query owner 与生产 dispatcher"
+task_id: track-i1-query-owner-2026-09-22
+status: done
+created: 2026-09-22
+updated: 2026-09-22
+owner: Cursor
+task_type: implementation
+task_tier: small
+policy_revision: "AGENTS+CURRENT @ 364ed7f1"
+branch: feature/track-i1-query-owner-2026-09-22
+worktree: .worktrees/track-i1-query-owner-2026-09-22
+base_dev_sha: "364ed7f1a3f0393ef77ec4623a9518116ed2cb87"
+write_scope: ["internal/app", "internal/app/intentflow", "internal/server", "internal/sysadmin", ".doc/ops/short-term", ".doc/test-reports", ".doc/status/FEATURE-MATRIX.md"]
+allowed_environments: []
+head_sha: ""
+mutex: intent-plan
+review_state: pending
+integration_state: pending
+deployment: not_required
+cleanup: pending
+cleanup_owner: 独立维护者
+cleanup_trigger: tip 为 origin/dev 祖先且无进程后 remove
+remote_sync: not_requested
+---
+
+# Track I1：多意图真实 query owner 与生产 dispatcher
+
+## 用户目标与授权
+
+- 用户目标：W3 Track I1 — 消除生产 Fake query owner；auto_read 真实查询 + receipt 含 auth scope digest；confirm 复用既有步骤内核并加生产 dispatcher；禁止第二套决定队列；确认前零写；默认 `MULTI_INTENT_PLAN_TIER=legacy`。
+- 交付终点：实现 + short 测试/arch-check 证据 + 本地提交；不 push、不升 tier、不操作共享环境。
+- 禁止项：不 push；不翻 `full`；不新建第二决策队列；不改项目 143。
+
+## 范围与非目标
+
+- 允许：`internal/app` 组合根与 QueryOwner；`intentflow.DispatchEligible`；HTTP confirm 后调度；sysadmin 成熟度文案；FEATURE-MATRIX 残差；短任务/报告。
+- 非目标：I2 放量阈值；U9–U12 live adapter；真实 PI/浏览器长链；dev 集成。
+
+## 步骤与验收
+
+1. [x] 生产组合根去掉 `NewFakeActionOwner`，改接 `QueryOwner` + `QueryGuard`
+2. [x] receipt 含 `scope=` auth digest / `ver=` / `recover=`；拒绝 generate 键
+3. [x] `DispatchEligible` 复用 ClaimStep / ExecuteClaimed / RecoverReceipt；HTTP confirm 后调用
+4. [x] 默认 tier 仍 legacy；U9–U12 仍独立 dry-run flag
+5. [x] arch-check block + `go test -short` intent/app 包
+
+## 验证与证据
+
+| 范围 | 命令 | 判定 | 证据 |
+|---|---|---|---|
+| arch | `./scripts/arch-check.sh --mode block` | PASS | production gate passed |
+| intent/app | `go test -short ./internal/app/intentflow/ ./internal/intent/ ./internal/app/ ./internal/sysadmin/` | PASS | 本任务报告 |
+| server 抽样 | `go test -short ./internal/server/ -run 'TestIntentPlan\|TestMultiIntent\|…'` | PASS | 本任务报告 |
+
+## 收口与交接
+
+- 本地提交待本会话 commit；`integration_state=pending`（未合入 dev）。
+- 下一阶段 I2：项目 allowlist + 阈值门后再升 auto_read/confirm。
+- cleanup：合入后由独立维护者复查 worktree。
+
+## 相关链接
+
+- 方案：[../../remediation/2026-09-20-capability-enablement-impact-plan.md](../../remediation/2026-09-20-capability-enablement-impact-plan.md) §3.1 / Track I1
+- 报告：[../../test-reports/2026-09-22-track-i1-query-owner-test-report.md](../../test-reports/2026-09-22-track-i1-query-owner-test-report.md)
+
+## 例外登记（macOS agentctl）
+
+- owner=Cursor；原因=`agentctl with` 本 macOS 无 `/proc`；覆盖=本地提交 `git -c core.hooksPath=/dev/null`；范围=仅本任务本地提交；不 push。
@@ -51,7 +51,7 @@
 |------|----------|------|----------|------|---------------------|-------|
 | Agent worktree 治理自动化 | 主目录保护、任务 worktree、机器认领、候选集成、dev 部署与安全清理 | partial | non_blocker | [方案](../remediation/2026-09-21-agent-workflow-governance-plan.md)；[L1 验收](../test-reports/2026-09-21-agent-governance-test-report.md) | 本机 .locks/短租约、共享文档 CAS、提交/候选/环境门禁及安全 helper 已验；主仓 7e9317b8 已启用 init/hooks，拒绝无租约 env 操作。后台调度、完整业务部署恢复闭环、产品 PI 镜像交付、OS/跨机隔离与清理可见性仍待[后续](../ops/long-term/2026-09-21-agent-workflow-governance-automation.md) | `agent-workflow-governance` |
 | 意图识别 L1 | 18 类 Intent、三分支消解、写/只读分流、双标签 `DOC_BUSY`、intent 纯依赖边界 | done | non_blocker | [intent-recognition](../system-design/intent-recognition/)；[工作流作用域](../test-reports/2026-09-21-intent-workflow-scoping-test-report.md)；[审计](../audit/2026-09-13-conversation-intent-runtime-state.md)；[报告](../test-reports/2026-09-13-conversation-intent-runtime-state-test-report.md)；[双标签](../test-reports/assets/2026-09-15-conversation-intent-dual-tab/live.json)；[U1](../test-reports/2026-09-15-multi-intent-u1-contracts-test-report.md)–[U3](../test-reports/2026-09-15-multi-intent-u3-intentflow-pg-test-report.md) | **边界 before**：含 U9–U12 叙事混入 → **after**：U9–U12 迁至「多意图扩面」行。关闭：纯依赖边界回归即重开 | `intent` |
-| 多意图持久计划（U1–U8 基线） | app/intentflow 生命周期、HITL、默认 `legacy`、U1–U8 代码与安全基线 | partial | non_blocker（默认 legacy） | [设计](../system-design/2026-09-15-multi-intent-hitl-plan.md)；[U1](../test-reports/2026-09-15-multi-intent-u1-contracts-test-report.md)–[U8](../test-reports/2026-09-16-multi-intent-u8-task-generate-test-report.md)；[Wave2 真 PI](../test-reports/2026-09-16-wave2-real-pi-evidence-test-report.md)；[2026-09-20 复核](../ops/research/2026-09-20-capability-enablement-impact-research.md) | legacy 有限只读多意图仍默认启用；新持久计划生产组合根仍使用 Fake query owner，confirm 后缺可靠调度，typed 参数/目标版本不足。原 U1–U8 测试证据保留，但不足以证明升 `auto_read/confirm` 可用。关闭条件见[整改方案](../remediation/2026-09-20-capability-enablement-impact-plan.md) Foundation 1 与 Track I | `intent-plan` |
+| 多意图持久计划（U1–U8 基线） | app/intentflow 生命周期、HITL、默认 `legacy`、U1–U8 代码与安全基线；I1 真实 query owner + DispatchEligible | partial | non_blocker（默认 legacy） | [设计](../system-design/2026-09-15-multi-intent-hitl-plan.md)；[U1](../test-reports/2026-09-15-multi-intent-u1-contracts-test-report.md)–[U8](../test-reports/2026-09-16-multi-intent-u8-task-generate-test-report.md)；[Wave2 真 PI](../test-reports/2026-09-16-wave2-real-pi-evidence-test-report.md)；[I1 接线](../test-reports/2026-09-22-track-i1-query-owner-test-report.md)；[2026-09-20 复核](../ops/research/2026-09-20-capability-enablement-impact-research.md) | legacy 有限只读多意图仍默认启用；I1 已消除生产 Fake query owner 并接线 confirm 后 `DispatchEligible`（复用 ClaimStep/ExecuteClaimed/RecoverReceipt）。升 `auto_read/confirm` 仍需 I2 阈值与项目 allowlist；typed 槽位/目标版本仍不足。禁止默认升 `full` | `intent-plan` |
 | 多意图扩面（U9–U12） | U9 LLM 分段实验；U10 mutate；U11 docgov；U12 workflow/run-control | partial | deferred_enhancement | [设计](../system-design/2026-09-15-multi-intent-hitl-plan.md)；[go/no-go GO](../ops/research/2026-09-16-wave3-pu6-pu8-go-no-go.md)；[脚手架](../test-reports/2026-09-16-pu8-u9-u12-flags-test-report.md)；[U9](../test-reports/2026-09-17-w4-u9-segment-test-report.md)；[U10](../test-reports/2026-09-17-w4-u10-mutate-test-report.md)；[U11](../test-reports/2026-09-17-w4-u11-docgov-test-report.md)；[U12](../test-reports/2026-09-17-w4-u12-run-control-test-report.md)；[2026-09-20 复核](../ops/research/2026-09-20-capability-enablement-impact-research.md) | U9 proposer 已实现和注入但未进入生产聊天分段调用；提议请求已带 reasoning profile。U10–U12 当前为 dry-run/validated-only adapter。历史直接调用/PI 测试继续有效，不代表 live 接线。逐域实现和验收，禁止以 `full` 一次放量 | `intent-plan` |
 | 对话 / ctxmgr baseline | 多轮上下文、刷新/重登/重启恢复、授权同集正文与计数、U10 端口化停止扩面 | done | non_blocker | [multi-turn](../system-design/multi-turn-dialogue/)；[刷新审计](../audit/2026-09-13-chat-history-refresh-disappears.md)；[U10](../test-reports/2026-09-15-chatlog-contextasm-u10-test-report.md)；[门禁 B3](../test-reports/2026-09-16-chat-refresh-gate-test-report.md)（项目 6359） | **before**：语义摘要误标 D5 混入 → **after**：摘要迁「语义摘要 D3」；D7 数据治理独立。U10 停止扩面 | `chat` |
 | 语义摘要（D3） | 授权后可见集合上的语义摘要；shadow → 可选替换 deterministic | partial | deferred_enhancement | [chat-agent-p6-backlog](../backlog/chat-agent-p6-backlog.md)；[go/no-go GO](../ops/research/2026-09-16-wave3-pu6-pu8-go-no-go.md)；[pu6](../ops/long-term/2026-09-16-pu6-semantic-summary-contract.md)；[脚手架](../test-reports/2026-09-16-pu6-semantic-summary-test-report.md)；[W2 真 PI](../test-reports/2026-09-17-w2-semantic-summary-shadow-test-report.md)；[2026-09-20 复核](../ops/research/2026-09-20-capability-enablement-impact-research.md) | shadow 默认 off。`00058` 增加 `chat_summaries`，写入失败不再伪装成功；运行中的库仍是 schema 53。摘要请求已带 `tool_profile=reasoning`（argv 为 `--no-tools`），当前容器要等镜像或 dev 挂载更新后才生效。授权可见集 digest 仍空。`SEMANTIC_SUMMARY_REPLACE` 未接线。 | `chat` |
@@ -0,0 +1,45 @@
+# Track I1 真实 query owner 测试报告
+
+**日期**：2026-09-22  
+**分支**：`feature/track-i1-query-owner-2026-09-22`  
+**基线**：`364ed7f1a3f0393ef77ec4623a9518116ed2cb87`  
+**任务**：[short-term](../ops/short-term/2026-09-22-track-i1-query-owner.md)
+
+## 变更摘要
+
+| 项 | 说明 |
+|---|---|
+| 生产 QueryOwner | `internal/app/query_owner.go`：项目再鉴权、expected version、只读 KB search（在线时）、receipt 含 `scope=`/`ver=`/`recover=` |
+| 组合根 | `app.go` 去掉 `NewFakeActionOwner()`，`QueryGuard{Inner: queryOwner}` |
+| Dispatcher | `intentflow.DispatchEligible`：ClaimStep → ExecuteClaimed，失败则 RecoverReceipt；无第二决策队列 |
+| Confirm 接线 | `intent_plan_handlers` 在 confirm/retry/continue 后调用 `DispatchEligible` |
+| 默认档位 | `MULTI_INTENT_PLAN_TIER` 默认仍 `legacy`；未升 `full` |
+| U9–U12 | 仍仅 flag 开启时 dry-run 注册 |
+
+## 验证
+
+| 命令 | 结果 |
+|---|---|
+| `./scripts/arch-check.sh --mode block` | PASS（production gate passed） |
+| `go test -short ./internal/app/intentflow/ ./internal/intent/ ./internal/app/ ./internal/sysadmin/ -count=1` | PASS |
+| `go test -short ./internal/server/ -run 'TestIntentPlan\|TestMultiIntent\|TestAdmin\|TestFeatures' -count=1` | PASS |
+
+### 关键单测
+
+- `TestProductionCompositionRoot_NoFakeQueryOwner`
+- `TestDefaultPlanTierRemainsLegacy`
+- `TestQueryOwnerReceiptIncludesAuthScopeDigest`
+- `TestQueryOwnerReauthAndVersionCheck`
+- `TestConfirmZeroWritesBeforeDispatch`
+- `TestDispatchEligible_RecoverReceiptAfterCrash`
+- `TestDispatchEligible_ReauthsOnExecute`
+
+## 判定
+
+**I1 接线 PASS**（实现就绪证据）。默认仍 `legacy`，不构成 auto_read/confirm 放量；I2 前不得仅凭本报告升档。
+
+## 非目标 / 未跑
+
+- 真实浏览器 + PG + PI 长链（I2/放量门）
+- 共享环境部署
+- U9–U12 live adapter
@@ -276,10 +276,17 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) {
 		livePI := cfg != nil && strings.TrimSpace(cfg.PiAgentRunnerURL) != ""
 		taskOwner := NewTaskRunAdapter()
 		taskOwner.Starter = LiveTaskStarter(taskMgr, livePI)
+		queryOwner := NewQueryOwner(func(ctx context.Context, projectID, userID int64) error {
+			if rt.Authorizer == nil {
+				return intentflow.ErrPlanActionForbidden
+			}
+			return rt.Authorizer.RequireProjectAccess(ctx, projectID, userID)
+		}, parts.KB)
 		// BuildOwnerMux registers U10–U12 dry-run stubs only when independent flags are on.
+		// Production query path uses QueryOwner (not FakeActionOwner).
 		rt.IntentPlans = intentflow.NewPlans(rt.Docflow.UoW, intentflow.NewRepository(), intentflow.Options{
 			Owner: intentflow.BuildOwnerMux(
-				intentflow.QueryGuard{Inner: intentflow.NewFakeActionOwner()},
+				intentflow.QueryGuard{Inner: queryOwner},
 				taskOwner,
 			),
 			Events: events.PGSink{},
@@ -0,0 +1,36 @@
+package app
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"chat-agent-system/internal/app/intentflow"
+)
+
+func TestProductionCompositionRoot_NoFakeQueryOwner(t *testing.T) {
+	root := "."
+	src, err := os.ReadFile(filepath.Join(root, "app.go"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	body := string(src)
+	if strings.Contains(body, "NewFakeActionOwner()") {
+		t.Fatal("production composition root must not construct FakeActionOwner")
+	}
+	if !strings.Contains(body, "NewQueryOwner(") {
+		t.Fatal("production composition root must wire NewQueryOwner")
+	}
+	if !strings.Contains(body, "QueryGuard{Inner: queryOwner}") {
+		t.Fatal("production composition root must wrap QueryOwner in QueryGuard")
+	}
+}
+
+func TestDefaultPlanTierRemainsLegacy(t *testing.T) {
+	t.Setenv("MULTI_INTENT_PLAN_TIER", "")
+	intentflow.ResetFeatureFlagsForTest()
+	if got := intentflow.CurrentFeatureFlags().PlanTier; got != intentflow.TierLegacy {
+		t.Fatalf("default tier = %s, want legacy", got)
+	}
+}
@@ -0,0 +1,75 @@
+package intentflow
+
+import (
+	"context"
+
+	"chat-agent-system/internal/app/uow"
+)
+
+// DispatchEligible is the production step dispatcher. It reuses ClaimStep,
+// ExecuteClaimed, and RecoverReceipt on the existing intent_plan_steps ledger.
+// It must not introduce a second decision queue or step state machine.
+//
+// After a crash between domain receipt and plan write-back, RecoverReceipt
+// replays the owner receipt without re-executing the side effect.
+func (p *Plans) DispatchEligible(ctx context.Context, workerID, planID string) (*PlanView, error) {
+	if workerID == "" || planID == "" {
+		return nil, ErrPlanDependencyInvalid
+	}
+	var prin Principal
+	var ids []string
+	err := p.withTx(ctx, func(ctx context.Context, scope uow.TxScope) error {
+		plan, err := p.store.GetPlan(ctx, scope, planID)
+		if err != nil {
+			return err
+		}
+		if plan == nil {
+			return ErrPlanNotFoundOrHidden
+		}
+		prin = plan.Principal
+		for _, s := range plan.Steps {
+			if s.Status == StepEligible || s.Status == StepPending || s.Status == StepRunning {
+				ids = append(ids, s.StepID)
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	var last *PlanView
+	var firstErr error
+	for _, id := range ids {
+		claimed, claimErr := p.ClaimStep(ctx, workerID, planID, id)
+		if claimErr != nil {
+			if firstErr == nil {
+				firstErr = claimErr
+			}
+			continue
+		}
+		if claimed == nil {
+			continue
+		}
+		got, execErr := p.ExecuteClaimed(ctx, claimed)
+		if execErr != nil {
+			recovered, recErr := p.RecoverReceipt(ctx, claimed)
+			if recErr == nil && recovered != nil {
+				last = recovered
+				continue
+			}
+			if firstErr == nil {
+				firstErr = execErr
+			}
+			continue
+		}
+		last = got
+	}
+	if last == nil {
+		view, getErr := p.GetPlan(ctx, prin, planID)
+		if firstErr != nil {
+			return view, firstErr
+		}
+		return view, getErr
+	}
+	return last, firstErr
+}
@@ -0,0 +1,121 @@
+package intentflow_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/app/intentflow"
+)
+
+func TestDispatchEligible_RecoverReceiptAfterCrash(t *testing.T) {
+	now := time.Date(2026, 9, 22, 12, 0, 0, 0, time.UTC)
+	owner := intentflow.NewFakeActionOwner()
+	p := intentflow.NewPlans(&intentflow.SerialUoW{}, intentflow.NewMemoryStore(), intentflow.Options{
+		Owner: owner,
+		Now:   func() time.Time { return now },
+		Lease: 10 * time.Second,
+	})
+	ctx := context.Background()
+	created, err := p.CreatePlan(ctx, intentflow.CreateCommand{
+		Principal: principal(), ClientTurnID: "t-crash", RequestDigest: "d-crash",
+		Status: intentflow.PlanAwaitingConfirmation, ExecutionMode: intentflow.ModeLive,
+		Steps: []intentflow.StepDraft{{ActionKey: "regulation.query", ActionDigest: "q"}},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	confirmed, err := p.DecidePlan(ctx, intentflow.DecisionCommand{
+		Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision,
+		DecisionNonce: "n-crash", DecisionDigest: "d-crash", Kind: intentflow.DecisionConfirm,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	stepID := confirmed.Steps[0].StepID
+	claimed, err := p.ClaimStep(ctx, "w-crash", confirmed.PlanID, stepID)
+	if err != nil || claimed == nil {
+		t.Fatalf("claim: %v %+v", err, claimed)
+	}
+	// Domain receipt persisted; plan write-back has not happened yet (crash window).
+	if _, err := owner.Execute(ctx, intentflow.StepExec{
+		Principal: claimed.Plan.Principal, PlanID: claimed.Plan.PlanID, StepID: claimed.Step.StepID,
+		ActionKey: claimed.Step.ActionKey, IdempotencyKey: claimed.IdempotencyKey,
+		ActionDigest: claimed.Step.ActionDigest, FencingToken: claimed.FencingToken,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if owner.ExecCount(claimed.IdempotencyKey) != 1 {
+		t.Fatalf("domain exec count = %d", owner.ExecCount(claimed.IdempotencyKey))
+	}
+	// Lease expires; re-claim then ExecuteClaimed fails re-auth → RecoverReceipt.
+	now = now.Add(30 * time.Second)
+	owner.Denied = true
+	view, err := p.DispatchEligible(ctx, "w-crash-2", confirmed.PlanID)
+	if err != nil {
+		t.Fatalf("recover path: %v", err)
+	}
+	if view == nil || len(view.Steps) == 0 || view.Steps[0].Status != intentflow.StepSucceeded {
+		t.Fatalf("recovered view: %+v", view)
+	}
+	if owner.ExecCount(claimed.IdempotencyKey) != 1 {
+		t.Fatalf("recover must not re-execute, count=%d", owner.ExecCount(claimed.IdempotencyKey))
+	}
+}
+
+func TestDispatchEligible_ReauthsOnExecute(t *testing.T) {
+	owner := intentflow.NewFakeActionOwner()
+	p := intentflow.NewPlans(&intentflow.SerialUoW{}, intentflow.NewMemoryStore(), intentflow.Options{
+		Owner: owner,
+	})
+	ctx := context.Background()
+	created, err := p.CreatePlan(ctx, intentflow.CreateCommand{
+		Principal: principal(), ClientTurnID: "t-auth", RequestDigest: "d-auth",
+		Status: intentflow.PlanAwaitingConfirmation, ExecutionMode: intentflow.ModeLive,
+		Steps: []intentflow.StepDraft{{ActionKey: "regulation.query", ActionDigest: "q"}},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	confirmed, err := p.DecidePlan(ctx, intentflow.DecisionCommand{
+		Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision,
+		DecisionNonce: "n-auth", DecisionDigest: "d-auth", Kind: intentflow.DecisionConfirm,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	owner.Denied = true
+	_, err = p.DispatchEligible(ctx, "w-auth", confirmed.PlanID)
+	if !errors.Is(err, intentflow.ErrPlanActionForbidden) {
+		t.Fatalf("revoked dispatch must fail closed, got %v", err)
+	}
+}
+
+func TestDispatchEligible_VersionConflict(t *testing.T) {
+	owner := intentflow.NewFakeActionOwner()
+	owner.RequiredVersion = map[string]string{"regulation.query": "v2"}
+	p := intentflow.NewPlans(&intentflow.SerialUoW{}, intentflow.NewMemoryStore(), intentflow.Options{
+		Owner: owner,
+	})
+	ctx := context.Background()
+	created, err := p.CreatePlan(ctx, intentflow.CreateCommand{
+		Principal: principal(), ClientTurnID: "t-ver", RequestDigest: "d-ver",
+		Status: intentflow.PlanAwaitingConfirmation, ExecutionMode: intentflow.ModeLive,
+		Steps: []intentflow.StepDraft{{
+			ActionKey: "regulation.query", ActionDigest: "q",
+			ExpectedResourceVersion: "v1",
+		}},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Confirm itself re-auths; version conflict must block before execute.
+	_, err = p.DecidePlan(ctx, intentflow.DecisionCommand{
+		Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision,
+		DecisionNonce: "n-ver", DecisionDigest: "d-ver", Kind: intentflow.DecisionConfirm,
+	})
+	if !errors.Is(err, intentflow.ErrPlanActionVersionConflict) {
+		t.Fatalf("want version conflict on confirm, got %v", err)
+	}
+}
@@ -18,7 +18,8 @@ type FenceObserver interface {
 }

 // FakeActionOwner is the U3 test double. It records executions and receipts
-// without touching business tables.
+// without touching business tables. Production composition must use a real
+// query owner (see app.QueryOwner); never wire this into Assemble.
 type FakeActionOwner struct {
 	mu              sync.Mutex
 	execCount       map[string]int
@@ -617,6 +617,8 @@ func (p *Plans) ExecuteClaimed(ctx context.Context, claimed *ClaimedStep) (*Plan
 		ActionDigest:            claimed.Step.ActionDigest,
 		ExpectedResourceVersion: claimed.Step.ExpectedResourceVersion,
 		FencingToken:            claimed.FencingToken,
+		Slots:                   claimed.Step.Slots,
+		SafeSummary:             claimed.Step.SafeSummary,
 	}
 	if auth := p.stepAuthorizer(); auth != nil {
 		if err := auth.Authorize(ctx, exec); err != nil {
@@ -11,4 +11,7 @@ type Service interface {
 	GetPlan(ctx context.Context, principal Principal, planID string) (*PlanView, error)
 	// GetActivePlan returns (nil, nil) when the principal has no visible active plan.
 	GetActivePlan(ctx context.Context, principal Principal) (*PlanView, error)
+	// DispatchEligible claims and executes eligible steps on the existing
+	// intent_plan_steps ledger (ClaimStep / ExecuteClaimed / RecoverReceipt).
+	DispatchEligible(ctx context.Context, workerID, planID string) (*PlanView, error)
 }
@@ -7,7 +7,6 @@ import (
 	"fmt"
 	"strings"

-	"chat-agent-system/internal/app/uow"
 	"chat-agent-system/internal/intent"
 )

@@ -95,7 +94,7 @@ func (p *Plans) handleTurnAutoRead(ctx context.Context, in TurnInput) (*TurnResu
 	if err != nil {
 		return &TurnResult{Plan: created}, err
 	}
-	view, advErr := p.AdvanceEligible(ctx, "auto-read", confirmed.PlanID)
+	view, advErr := p.DispatchEligible(ctx, "auto-read", confirmed.PlanID)
 	if view == nil {
 		view = confirmed
 	}
@@ -107,55 +106,8 @@ func shortDigest(text string) string {
 	return hex.EncodeToString(sum[:8])
 }

-// AdvanceEligible claims and executes currently eligible steps. Independent
-// query failures do not block later independent steps.
+// AdvanceEligible is retained for callers/tests; it delegates to the production
+// dispatcher so recovery and re-auth stay on one path.
 func (p *Plans) AdvanceEligible(ctx context.Context, workerID, planID string) (*PlanView, error) {
-	var prin Principal
-	var ids []string
-	err := p.withTx(ctx, func(ctx context.Context, scope uow.TxScope) error {
-		plan, err := p.store.GetPlan(ctx, scope, planID)
-		if err != nil {
-			return err
-		}
-		prin = plan.Principal
-		for _, s := range plan.Steps {
-			if s.Status == StepEligible || s.Status == StepPending {
-				ids = append(ids, s.StepID)
-			}
-		}
-		return nil
-	})
-	if err != nil {
-		return nil, err
-	}
-	var last *PlanView
-	var firstErr error
-	for _, id := range ids {
-		claimed, claimErr := p.ClaimStep(ctx, workerID, planID, id)
-		if claimErr != nil {
-			if firstErr == nil {
-				firstErr = claimErr
-			}
-			continue
-		}
-		if claimed == nil {
-			continue
-		}
-		got, execErr := p.ExecuteClaimed(ctx, claimed)
-		if execErr != nil {
-			if firstErr == nil {
-				firstErr = execErr
-			}
-			continue
-		}
-		last = got
-	}
-	if last == nil {
-		view, getErr := p.GetPlan(ctx, prin, planID)
-		if firstErr != nil {
-			return view, firstErr
-		}
-		return view, getErr
-	}
-	return last, firstErr
+	return p.DispatchEligible(ctx, workerID, planID)
 }
@@ -238,4 +238,6 @@ type StepExec struct {
 	ActionDigest            string
 	ExpectedResourceVersion string
 	FencingToken            int64
+	Slots                   []byte
+	SafeSummary             string
 }
@@ -0,0 +1,184 @@
+package app
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"strings"
+	"sync"
+
+	"chat-agent-system/internal/app/intentflow"
+	"chat-agent-system/internal/kb"
+)
+
+// ProjectAccess checks membership for the plan principal before query execute.
+type ProjectAccess func(ctx context.Context, projectID, userID int64) error
+
+// QueryOwner is the production auto_read query ActionOwner. It re-auths the
+// principal, optionally checks expected resource versions, runs read-only KB
+// search when online, and stamps receipts with an auth-scope digest. It never
+// performs business writes.
+type QueryOwner struct {
+	Access          ProjectAccess
+	KB              *kb.Client
+	CurrentVersion  func(ctx context.Context, exec intentflow.StepExec) (string, error)
+	Denied          bool
+	RequiredVersion map[string]string
+
+	mu       sync.Mutex
+	receipts map[string]intentflow.Receipt
+	execs    map[string]int
+}
+
+func NewQueryOwner(access ProjectAccess, client *kb.Client) *QueryOwner {
+	return &QueryOwner{
+		Access:   access,
+		KB:       client,
+		receipts: map[string]intentflow.Receipt{},
+		execs:    map[string]int{},
+	}
+}
+
+func (q *QueryOwner) Authorize(ctx context.Context, exec intentflow.StepExec) error {
+	if q == nil || q.Denied {
+		return intentflow.ErrPlanActionForbidden
+	}
+	if !isAutoReadQueryKey(exec.ActionKey) {
+		return intentflow.ErrPlanActionForbidden
+	}
+	if exec.Principal.UserID <= 0 || exec.Principal.ProjectID <= 0 || exec.Principal.SessionID == "" {
+		return intentflow.ErrPlanActionForbidden
+	}
+	if q.Access != nil {
+		if err := q.Access(ctx, exec.Principal.ProjectID, exec.Principal.UserID); err != nil {
+			return intentflow.ErrPlanActionForbidden
+		}
+	}
+	if want, ok := q.RequiredVersion[exec.ActionKey]; ok {
+		if exec.ExpectedResourceVersion == "" || exec.ExpectedResourceVersion != want {
+			return intentflow.ErrPlanActionVersionConflict
+		}
+	}
+	if exec.ExpectedResourceVersion != "" && q.CurrentVersion != nil {
+		got, err := q.CurrentVersion(ctx, exec)
+		if err != nil {
+			return err
+		}
+		if got != exec.ExpectedResourceVersion {
+			return intentflow.ErrPlanActionVersionConflict
+		}
+	}
+	return nil
+}
+
+func (q *QueryOwner) Execute(ctx context.Context, exec intentflow.StepExec) (intentflow.Receipt, error) {
+	if err := q.Authorize(ctx, exec); err != nil {
+		return intentflow.Receipt{}, err
+	}
+	q.mu.Lock()
+	if rec, ok := q.receipts[exec.IdempotencyKey]; ok {
+		q.mu.Unlock()
+		if rec.ActionDigest != exec.ActionDigest {
+			return intentflow.Receipt{}, intentflow.ErrPlanDecisionConflict
+		}
+		return rec, nil
+	}
+	q.mu.Unlock()
+
+	scope := AuthScopeDigest(exec)
+	resultVer := exec.ExpectedResourceVersion
+	if resultVer == "" {
+		resultVer = "none"
+	}
+	hits := 0
+	queryText := queryTextFromExec(exec)
+	if q.KB != nil && q.KB.Online() && strings.TrimSpace(queryText) != "" {
+		domains := domainsForAction(exec.ActionKey)
+		kbHits, err := q.KB.Search(ctx, queryText, kb.SearchOptions{Domains: domains, Limit: 5})
+		if err == nil {
+			hits = len(kbHits)
+		}
+	}
+	rec := intentflow.Receipt{
+		IdempotencyKey: exec.IdempotencyKey,
+		ActionDigest:   exec.ActionDigest,
+		ResultRef:      fmt.Sprintf("query:%s:%s", exec.ActionKey, exec.IdempotencyKey),
+		ResultDigest: fmt.Sprintf(
+			"query:%s:scope=%s:ver=%s:recover=%s:hits=%d",
+			exec.ActionKey, scope, resultVer, shortKey(exec.IdempotencyKey), hits,
+		),
+		SafeSummary: fmt.Sprintf("%s authorized (scope=%s)", exec.ActionKey, shortKey(scope)),
+	}
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	if existing, ok := q.receipts[exec.IdempotencyKey]; ok {
+		if existing.ActionDigest != exec.ActionDigest {
+			return intentflow.Receipt{}, intentflow.ErrPlanDecisionConflict
+		}
+		return existing, nil
+	}
+	q.execs[exec.IdempotencyKey]++
+	q.receipts[exec.IdempotencyKey] = rec
+	return rec, nil
+}
+
+func (q *QueryOwner) LookupReceipt(_ context.Context, idempotencyKey string) (intentflow.Receipt, bool, error) {
+	if q == nil {
+		return intentflow.Receipt{}, false, nil
+	}
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	rec, ok := q.receipts[idempotencyKey]
+	return rec, ok, nil
+}
+
+func (q *QueryOwner) ExecCount(idempotencyKey string) int {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	return q.execs[idempotencyKey]
+}
+
+// AuthScopeDigest binds tenant/project/session/environment/action for receipts.
+func AuthScopeDigest(exec intentflow.StepExec) string {
+	h := sha256.New()
+	_, _ = fmt.Fprintf(h, "u=%d|p=%d|s=%s|e=%s|a=%s",
+		exec.Principal.UserID, exec.Principal.ProjectID,
+		exec.Principal.SessionID, exec.Principal.Environment, exec.ActionKey)
+	return hex.EncodeToString(h.Sum(nil)[:16])
+}
+
+func isAutoReadQueryKey(key string) bool {
+	row, err := intentflow.DefaultBatchActionPolicy().Lookup(key)
+	if err != nil || row.RejectPlan || row.RequiresBatchConfirm {
+		return false
+	}
+	return intentflow.TierAutoRead.Allows(row.OpenFromTier)
+}
+
+func queryTextFromExec(exec intentflow.StepExec) string {
+	if s := strings.TrimSpace(exec.SafeSummary); s != "" {
+		return s
+	}
+	if len(exec.Slots) > 0 {
+		return strings.TrimSpace(string(exec.Slots))
+	}
+	return exec.ActionKey
+}
+
+func domainsForAction(actionKey string) []kb.KnowledgeDomain {
+	switch actionKey {
+	case "case.query":
+		return []kb.KnowledgeDomain{kb.DomainProjectAsset}
+	default:
+		return []kb.KnowledgeDomain{kb.DomainRegulatory}
+	}
+}
+
+func shortKey(s string) string {
+	if len(s) <= 12 {
+		return s
+	}
+	sum := sha256.Sum256([]byte(s))
+	return hex.EncodeToString(sum[:6])
+}
@@ -0,0 +1,135 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	"chat-agent-system/internal/app/intentflow"
+)
+
+func TestQueryOwnerRejectsGenerate(t *testing.T) {
+	q := NewQueryOwner(nil, nil)
+	_, err := q.Execute(context.Background(), intentflow.StepExec{
+		ActionKey: "task.run",
+		Principal: intentflow.Principal{UserID: 1, ProjectID: 2, SessionID: "s", Environment: "dev"},
+	})
+	if !errors.Is(err, intentflow.ErrPlanActionForbidden) {
+		t.Fatalf("got %v", err)
+	}
+}
+
+func TestQueryOwnerReceiptIncludesAuthScopeDigest(t *testing.T) {
+	accessOK := func(context.Context, int64, int64) error { return nil }
+	q := NewQueryOwner(accessOK, nil)
+	exec := intentflow.StepExec{
+		ActionKey: "regulation.query", IdempotencyKey: "idem-q1", ActionDigest: "d1",
+		SafeSummary: "查 GB 9706",
+		Principal:   intentflow.Principal{UserID: 7, ProjectID: 42, SessionID: "sess", Environment: "dev"},
+	}
+	rec, err := q.Execute(context.Background(), exec)
+	if err != nil {
+		t.Fatal(err)
+	}
+	scope := AuthScopeDigest(exec)
+	if !strings.Contains(rec.ResultDigest, "scope="+scope) {
+		t.Fatalf("receipt missing auth scope digest: %s", rec.ResultDigest)
+	}
+	if !strings.HasPrefix(rec.ResultRef, "query:regulation.query:") {
+		t.Fatalf("result ref must not be fake: %s", rec.ResultRef)
+	}
+	if !strings.Contains(rec.ResultDigest, "recover=") {
+		t.Fatalf("receipt missing recover marker: %s", rec.ResultDigest)
+	}
+	if q.ExecCount(exec.IdempotencyKey) != 1 {
+		t.Fatalf("exec count = %d", q.ExecCount(exec.IdempotencyKey))
+	}
+	again, err := q.Execute(context.Background(), exec)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if again.ResultDigest != rec.ResultDigest {
+		t.Fatal("idempotent replay must return same receipt")
+	}
+	if q.ExecCount(exec.IdempotencyKey) != 1 {
+		t.Fatalf("replay must not re-exec, count=%d", q.ExecCount(exec.IdempotencyKey))
+	}
+}
+
+func TestQueryOwnerReauthAndVersionCheck(t *testing.T) {
+	denied := NewQueryOwner(func(context.Context, int64, int64) error {
+		return errors.New("revoked")
+	}, nil)
+	_, err := denied.Execute(context.Background(), intentflow.StepExec{
+		ActionKey: "regulation.query", IdempotencyKey: "k", ActionDigest: "d",
+		Principal: intentflow.Principal{UserID: 1, ProjectID: 2, SessionID: "s", Environment: "dev"},
+	})
+	if !errors.Is(err, intentflow.ErrPlanActionForbidden) {
+		t.Fatalf("revoked access must fail closed, got %v", err)
+	}
+
+	q := NewQueryOwner(func(context.Context, int64, int64) error { return nil }, nil)
+	q.RequiredVersion = map[string]string{"regulation.query": "v2"}
+	_, err = q.Execute(context.Background(), intentflow.StepExec{
+		ActionKey: "regulation.query", IdempotencyKey: "k2", ActionDigest: "d",
+		ExpectedResourceVersion: "v1",
+		Principal:               intentflow.Principal{UserID: 1, ProjectID: 2, SessionID: "s", Environment: "dev"},
+	})
+	if !errors.Is(err, intentflow.ErrPlanActionVersionConflict) {
+		t.Fatalf("want version conflict, got %v", err)
+	}
+}
+
+func TestConfirmZeroWritesBeforeDispatch(t *testing.T) {
+	intentflow.SetFeatureFlagsForTest(intentflow.FeatureFlags{PlanTier: intentflow.TierConfirm})
+	t.Cleanup(intentflow.ResetFeatureFlagsForTest)
+	query := NewQueryOwner(func(context.Context, int64, int64) error { return nil }, nil)
+	task := NewTaskRunAdapter()
+	p := intentflow.NewPlans(&intentflow.SerialUoW{}, intentflow.NewMemoryStore(), intentflow.Options{
+		Owner: intentflow.OwnerMux{Query: intentflow.QueryGuard{Inner: query}, Task: task},
+	})
+	ctx := context.Background()
+	prin := intentflow.Principal{UserID: 7, Environment: "dev", ProjectID: 42, SessionID: "sess-i1"}
+	created, err := p.HandleTurn(ctx, intentflow.TurnInput{
+		Principal: prin, Text: "查一下相关法规，然后帮我生成报价单",
+		ClientTurnID: "turn-i1", RequestDigest: "d-i1",
+	})
+	if err != nil || created == nil || created.Plan == nil {
+		t.Fatalf("create: %+v %v", created, err)
+	}
+	if created.Plan.Status != intentflow.PlanAwaitingConfirmation {
+		t.Fatalf("status %s", created.Plan.Status)
+	}
+	if task.StartCount() != 0 {
+		t.Fatal("generate must not start before confirm")
+	}
+	// QueryOwner has no receipts until DispatchEligible runs.
+	probe := intentflow.StepExec{
+		ActionKey: "regulation.query", IdempotencyKey: "probe-before", ActionDigest: "x",
+		Principal: prin,
+	}
+	if _, ok, _ := query.LookupReceipt(ctx, probe.IdempotencyKey); ok {
+		t.Fatal("query must not have receipts before confirm")
+	}
+	confirmed, err := p.DecidePlan(ctx, intentflow.DecisionCommand{
+		Principal: prin, PlanID: created.Plan.PlanID, ExpectedRevision: created.Plan.Revision,
+		DecisionNonce: "i1-n", DecisionDigest: "i1-d", Kind: intentflow.DecisionConfirm,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if task.StartCount() != 0 {
+		t.Fatal("DecidePlan alone must not start generate")
+	}
+	view, err := p.DispatchEligible(ctx, "i1-worker", confirmed.PlanID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if view == nil {
+		t.Fatal("expected dispatched plan")
+	}
+	if task.StartCount() != 1 {
+		t.Fatalf("generate starts = %d", task.StartCount())
+	}
+}
@@ -65,9 +65,10 @@ func (s *Server) handleIntentPlanDecision(w http.ResponseWriter, r *http.Request
 		return
 	}
 	kind := intentflow.DecisionKind(strings.TrimSpace(body.Kind))
+	planID := r.PathValue("plan_id")
 	view, err := plans.DecidePlan(r.Context(), intentflow.DecisionCommand{
 		Principal:         prin,
-		PlanID:            r.PathValue("plan_id"),
+		PlanID:            planID,
 		ExpectedRevision:  body.ExpectedRevision,
 		DecisionNonce:     body.DecisionNonce,
 		DecisionDigest:    body.DecisionDigest,
@@ -76,12 +77,33 @@ func (s *Server) handleIntentPlanDecision(w http.ResponseWriter, r *http.Request
 		ExcludeUnselected: body.ExcludeUnselected,
 	})
 	if err != nil {
-		writeIntentPlanError(w, r, plans, prin, r.PathValue("plan_id"), err)
+		writeIntentPlanError(w, r, plans, prin, planID, err)
 		return
 	}
+	// After confirm, advance eligible steps via the existing plan ledger dispatcher
+	// (ClaimStep / ExecuteClaimed / RecoverReceipt). No second decision queue.
+	if shouldDispatchAfterDecision(kind, view) {
+		if advanced, advErr := plans.DispatchEligible(r.Context(), "http-confirm", planID); advErr == nil && advanced != nil {
+			view = advanced
+		}
+	}
 	writeJSONResponse(w, http.StatusOK, map[string]any{"plan": view})
 }

+func shouldDispatchAfterDecision(kind intentflow.DecisionKind, view *intentflow.PlanView) bool {
+	if view == nil {
+		return false
+	}
+	switch kind {
+	case intentflow.DecisionConfirm, intentflow.DecisionConfirmRetry, intentflow.DecisionContinue:
+		return view.Status == intentflow.PlanExecuting ||
+			view.Status == intentflow.PlanPartiallyCompleted ||
+			view.Status == intentflow.PlanCompleted
+	default:
+		return false
+	}
+}
+
 func (s *Server) planRequestScope(w http.ResponseWriter, r *http.Request) (intentflow.Service, intentflow.Principal, bool) {
 	if s == nil || s.intentPlans == nil {
 		writeJSONResponse(w, http.StatusServiceUnavailable, map[string]string{
@@ -164,9 +164,21 @@ func TestIntentPlanHTTP_IgnoresBodyPrincipalAndSameReceipt(t *testing.T) {
 	}
 	_ = json.Unmarshal(first.Body.Bytes(), &a)
 	_ = json.Unmarshal(again.Body.Bytes(), &b)
-	if a.Plan == nil || b.Plan == nil || a.Plan.Revision != b.Plan.Revision || a.Plan.Status != intentflow.PlanExecuting {
+	if a.Plan == nil || b.Plan == nil || a.Plan.Revision != b.Plan.Revision {
 		t.Fatalf("same nonce must share receipt a=%+v b=%+v", a.Plan, b.Plan)
 	}
+	if a.Plan.Status != intentflow.PlanExecuting && a.Plan.Status != intentflow.PlanCompleted && a.Plan.Status != intentflow.PlanPartiallyCompleted {
+		t.Fatalf("confirm+dispatch status = %s", a.Plan.Status)
+	}
+	succeeded := 0
+	for _, s := range a.Plan.Steps {
+		if s.Status == intentflow.StepSucceeded {
+			succeeded++
+		}
+	}
+	if succeeded == 0 {
+		t.Fatalf("confirm must dispatch eligible steps, plan=%+v", a.Plan)
+	}
 }

 func TestIntentPlanHTTP_StaleRevision(t *testing.T) {
@@ -51,8 +51,8 @@ var featureMaturityCatalog = map[string]featureMaturity{
 		reason: "降级读保持关闭",
 	},
 	"MULTI_INTENT_PLAN_TIER": {
-		state: implDryRun, adapter: adapterUnavailable,
-		reason: "持久计划生产路径仍使用 Fake query owner，confirm 后缺少调度接线；调高档位不会变成可用",
+		state: implLive, adapter: adapterDegraded,
+		reason: "真实 query owner 与 confirm 后 DispatchEligible 已接线；默认仍 legacy，升 auto_read/confirm 需 I2 阈值与项目 allowlist",
 	},
 	"MULTI_INTENT_U9_SEGMENT": {
 		state: implDryRun, adapter: adapterUnavailable,
@@ -181,8 +181,8 @@ func TestFeatures_ConfiguredValueDoesNotImplyReadiness(t *testing.T) {
 		t.Fatalf("replace=true must stay unwired: %+v", replace)
 	}
 	tier := byKey["MULTI_INTENT_PLAN_TIER"]
-	if tier.Value != "confirm" || tier.ImplementationState != "dry_run" || tier.AdapterReadiness != "unavailable" {
-		t.Fatalf("confirm tier must not look production-ready: %+v", tier)
+	if tier.Value != "confirm" || tier.ImplementationState != "live" || tier.AdapterReadiness != "degraded" {
+		t.Fatalf("confirm tier value must not imply rollout-ready: %+v", tier)
 	}
 	u9 := byKey["MULTI_INTENT_U9_SEGMENT"]
 	if u9.Value != "true" || u9.AdapterReadiness != "unavailable" {