diff --git a/.doc/ops/short-term/2026-09-22-track-i1-query-owner.md b/.doc/ops/short-term/2026-09-22-track-i1-query-owner.md new file mode 100644 index 00000000..97a00d6b --- /dev/null +++ b/.doc/ops/short-term/2026-09-22-track-i1-query-owner.md @@ -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。 diff --git a/.doc/status/FEATURE-MATRIX.md b/.doc/status/FEATURE-MATRIX.md index c6112963..3d64d9e4 100644 --- a/.doc/status/FEATURE-MATRIX.md +++ b/.doc/status/FEATURE-MATRIX.md @@ -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` | diff --git a/.doc/test-reports/2026-09-22-track-i1-query-owner-test-report.md b/.doc/test-reports/2026-09-22-track-i1-query-owner-test-report.md new file mode 100644 index 00000000..de482919 --- /dev/null +++ b/.doc/test-reports/2026-09-22-track-i1-query-owner-test-report.md @@ -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 diff --git a/internal/app/app.go b/internal/app/app.go index 820a3eaa..c17f4152 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -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{}, diff --git a/internal/app/composition_query_owner_test.go b/internal/app/composition_query_owner_test.go new file mode 100644 index 00000000..10040151 --- /dev/null +++ b/internal/app/composition_query_owner_test.go @@ -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) + } +} diff --git a/internal/app/intentflow/dispatch.go b/internal/app/intentflow/dispatch.go new file mode 100644 index 00000000..709a3d23 --- /dev/null +++ b/internal/app/intentflow/dispatch.go @@ -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 +} diff --git a/internal/app/intentflow/dispatch_test.go b/internal/app/intentflow/dispatch_test.go new file mode 100644 index 00000000..e7e40fd6 --- /dev/null +++ b/internal/app/intentflow/dispatch_test.go @@ -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) + } +} diff --git a/internal/app/intentflow/owner.go b/internal/app/intentflow/owner.go index d1790251..35d724e3 100644 --- a/internal/app/intentflow/owner.go +++ b/internal/app/intentflow/owner.go @@ -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 diff --git a/internal/app/intentflow/plans.go b/internal/app/intentflow/plans.go index 2a865fd6..8bece61b 100644 --- a/internal/app/intentflow/plans.go +++ b/internal/app/intentflow/plans.go @@ -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 { diff --git a/internal/app/intentflow/service.go b/internal/app/intentflow/service.go index 9e595839..7bebf745 100644 --- a/internal/app/intentflow/service.go +++ b/internal/app/intentflow/service.go @@ -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) } diff --git a/internal/app/intentflow/turn.go b/internal/app/intentflow/turn.go index e050674d..7c63951d 100644 --- a/internal/app/intentflow/turn.go +++ b/internal/app/intentflow/turn.go @@ -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) } diff --git a/internal/app/intentflow/types.go b/internal/app/intentflow/types.go index 2373aec7..b70f6bd3 100644 --- a/internal/app/intentflow/types.go +++ b/internal/app/intentflow/types.go @@ -238,4 +238,6 @@ type StepExec struct { ActionDigest string ExpectedResourceVersion string FencingToken int64 + Slots []byte + SafeSummary string } diff --git a/internal/app/query_owner.go b/internal/app/query_owner.go new file mode 100644 index 00000000..751b2579 --- /dev/null +++ b/internal/app/query_owner.go @@ -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]) +} diff --git a/internal/app/query_owner_test.go b/internal/app/query_owner_test.go new file mode 100644 index 00000000..f42c6e4a --- /dev/null +++ b/internal/app/query_owner_test.go @@ -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()) + } +} diff --git a/internal/server/intent_plan_handlers.go b/internal/server/intent_plan_handlers.go index 70ced247..e78917f4 100644 --- a/internal/server/intent_plan_handlers.go +++ b/internal/server/intent_plan_handlers.go @@ -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{ diff --git a/internal/server/intent_plan_handlers_test.go b/internal/server/intent_plan_handlers_test.go index 82d8ad61..8172703e 100644 --- a/internal/server/intent_plan_handlers_test.go +++ b/internal/server/intent_plan_handlers_test.go @@ -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) { diff --git a/internal/sysadmin/features.go b/internal/sysadmin/features.go index 21444826..e3d90d48 100644 --- a/internal/sysadmin/features.go +++ b/internal/sysadmin/features.go @@ -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, diff --git a/internal/sysadmin/service_test.go b/internal/sysadmin/service_test.go index 95f8087a..c01e02cd 100644 --- a/internal/sysadmin/service_test.go +++ b/internal/sysadmin/service_test.go @@ -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" {