merge: feature/track-i1-query-owner-2026-09-22
Land QueryOwner code; FEATURE-MATRIX deferred to docs apply.
This commit is contained in:
@@ -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: "aa78d03044851a6db792908cc011dae04280e703"
|
||||
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。
|
||||
@@ -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
|
||||
+8
-1
@@ -280,10 +280,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,
|
||||
|
||||
@@ -189,8 +189,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" {
|
||||
|
||||
Reference in New Issue
Block a user