feat(intentflow): route confirmed generate steps through a task adapter

Persist ComboNeedsConfirm plans at confirm tier without executing writes, then run task.run via OwnerMux and a stable idempotency key.

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

internal/app/app.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"strings"
 	"sync"
 	"time"
 
@@ -192,8 +193,14 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) {
 	if pool != nil {
 		rt.Docflow.UoW = uow.NewPG(pool)
 		rt.Docflow.Events = events.PGSink{}
+		livePI := cfg != nil && strings.TrimSpace(cfg.PiAgentRunnerURL) != ""
+		taskOwner := NewTaskRunAdapter()
+		taskOwner.Starter = LiveTaskStarter(taskMgr, livePI)
 		rt.IntentPlans = intentflow.NewPlans(rt.Docflow.UoW, intentflow.NewRepository(), intentflow.Options{
-			Owner:  intentflow.QueryGuard{Inner: intentflow.NewFakeActionOwner()},
+			Owner: intentflow.OwnerMux{
+				Query: intentflow.QueryGuard{Inner: intentflow.NewFakeActionOwner()},
+				Task:  taskOwner,
+			},
 			Events: events.PGSink{},
 		})
 	}

internal/app/intentflow/mux.go

@@ -0,0 +1,61 @@
+package intentflow
+
+import "context"
+
+// OwnerMux routes generate keys to Task and everything else to Query.
+type OwnerMux struct {
+	Query ActionOwner
+	Task  ActionOwner
+}
+
+func (m OwnerMux) pick(actionKey string) ActionOwner {
+	switch actionKey {
+	case "task.run", "task.run.adhoc":
+		return m.Task
+	default:
+		return m.Query
+	}
+}
+
+func (m OwnerMux) Authorize(ctx context.Context, exec StepExec) error {
+	owner := m.pick(exec.ActionKey)
+	if owner == nil {
+		return ErrPlanActionForbidden
+	}
+	if a, ok := owner.(StepAuthorizer); ok {
+		return a.Authorize(ctx, exec)
+	}
+	return nil
+}
+
+func (m OwnerMux) Execute(ctx context.Context, exec StepExec) (Receipt, error) {
+	if err := m.Authorize(ctx, exec); err != nil {
+		return Receipt{}, err
+	}
+	owner := m.pick(exec.ActionKey)
+	if owner == nil {
+		return Receipt{}, ErrPlanActionForbidden
+	}
+	return owner.Execute(ctx, exec)
+}
+
+func (m OwnerMux) LookupReceipt(ctx context.Context, idempotencyKey string) (Receipt, bool, error) {
+	if m.Task != nil {
+		if rec, ok, err := m.Task.LookupReceipt(ctx, idempotencyKey); err != nil || ok {
+			return rec, ok, err
+		}
+	}
+	if m.Query != nil {
+		return m.Query.LookupReceipt(ctx, idempotencyKey)
+	}
+	return Receipt{}, false, nil
+}
+
+func (m OwnerMux) ObserveFence(stepID string, token int64) {
+	if obs, ok := m.Query.(FenceObserver); ok {
+		obs.ObserveFence(stepID, token)
+	}
+	if obs, ok := m.Task.(FenceObserver); ok {
+		obs.ObserveFence(stepID, token)
+	}
+}

internal/app/intentflow/mux_test.go

@@ -0,0 +1,58 @@
+package intentflow_test
+
+import (
+	"context"
+	"testing"
+
+	"chat-agent-system/internal/app/intentflow"
+)
+
+type routeProbe struct {
+	keys []string
+}
+
+func (p *routeProbe) Authorize(_ context.Context, exec intentflow.StepExec) error {
+	p.keys = append(p.keys, "auth:"+exec.ActionKey)
+	return nil
+}
+
+func (p *routeProbe) Execute(ctx context.Context, exec intentflow.StepExec) (intentflow.Receipt, error) {
+	p.keys = append(p.keys, "exec:"+exec.ActionKey)
+	return intentflow.Receipt{IdempotencyKey: exec.IdempotencyKey, ActionDigest: exec.ActionDigest, ResultRef: "probe:" + exec.ActionKey}, nil
+}
+
+func (p *routeProbe) LookupReceipt(context.Context, string) (intentflow.Receipt, bool, error) {
+	return intentflow.Receipt{}, false, nil
+}
+
+func TestOwnerMuxRoutesGenerateToTask(t *testing.T) {
+	query := &routeProbe{}
+	task := &routeProbe{}
+	mux := intentflow.OwnerMux{Query: query, Task: task}
+	ctx := context.Background()
+	if _, err := mux.Execute(ctx, intentflow.StepExec{ActionKey: "regulation.query", IdempotencyKey: "k-q", ActionDigest: "dq"}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := mux.Execute(ctx, intentflow.StepExec{ActionKey: "task.run", IdempotencyKey: "k-t", ActionDigest: "dt"}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := mux.Execute(ctx, intentflow.StepExec{ActionKey: "task.run.adhoc", IdempotencyKey: "k-a", ActionDigest: "da"}); err != nil {
+		t.Fatal(err)
+	}
+	if len(query.keys) == 0 || query.keys[len(query.keys)-1] != "exec:regulation.query" {
+		t.Fatalf("query probe = %v", query.keys)
+	}
+	if len(task.keys) < 2 {
+		t.Fatalf("task probe = %v", task.keys)
+	}
+}
+
+func TestOwnerMuxNilTaskFailClosed(t *testing.T) {
+	mux := intentflow.OwnerMux{Query: intentflow.NewFakeActionOwner(), Task: nil}
+	_, err := mux.Execute(context.Background(), intentflow.StepExec{
+		ActionKey: "task.run", Principal: principal(), IdempotencyKey: "k", ActionDigest: "d",
+	})
+	if err != intentflow.ErrPlanActionForbidden {
+		t.Fatalf("nil Task must fail closed, got %v", err)
+	}
+}

internal/app/intentflow/plans.go

@@ -99,7 +99,8 @@ func visibleTo(plan *Plan, p Principal) bool {
 }
 
 // HandleTurn stays a no-op while MULTI_INTENT_PLAN_TIER is legacy/shadow (default).
-// U7 maps read-only utterances onto CreatePlan when auto_read is enabled.
+// auto_read executes independent queries; confirm persists write/generate plans
+// awaiting confirmation and does not execute them.
 func (p *Plans) HandleTurn(ctx context.Context, in TurnInput) (*TurnResult, error) {
 	if err := requirePrincipal(in.Principal); err != nil {
 		return nil, err

internal/app/intentflow/query.go

@@ -3,7 +3,7 @@ package intentflow
 import "context"
 
 // QueryGuard only authorizes low-risk query actions opened from auto_read.
-// Write/generate keys fail closed until a later adapter is wired.
+// Write/generate keys fail closed; OwnerMux routes those to Task.
 type QueryGuard struct {
 	Inner  ActionOwner
 	Policy *BatchActionPolicy
@@ -43,3 +43,9 @@ func (g QueryGuard) LookupReceipt(ctx context.Context, idempotencyKey string) (R
 	}
 	return g.Inner.LookupReceipt(ctx, idempotencyKey)
 }
+
+func (g QueryGuard) ObserveFence(stepID string, token int64) {
+	if obs, ok := g.Inner.(FenceObserver); ok {
+		obs.ObserveFence(stepID, token)
+	}
+}

internal/app/intentflow/turn.go

@@ -26,7 +26,9 @@ func (p *Plans) handleTurnAutoRead(ctx context.Context, in TurnInput) (*TurnResu
 		return &TurnResult{}, nil
 	}
 	proposal := res.PlanProposal
-	if proposal.ComboHint != intent.ComboReadOnly || len(proposal.Segments) < 2 || len(proposal.Segments) > 4 {
+	confirmWrites := proposal.ComboHint == intent.ComboNeedsConfirm && tier.Allows(TierConfirm)
+	readOnly := proposal.ComboHint == intent.ComboReadOnly && tier.Allows(TierAutoRead)
+	if (!confirmWrites && !readOnly) || len(proposal.Segments) < 2 || len(proposal.Segments) > 4 {
 		return &TurnResult{}, nil
 	}
 	policy := DefaultBatchActionPolicy()
@@ -40,7 +42,10 @@ func (p *Plans) handleTurnAutoRead(ctx context.Context, in TurnInput) (*TurnResu
 			return &TurnResult{}, nil
 		}
 		row, err := policy.Lookup(key)
-		if err != nil || row.RejectPlan || row.RequiresBatchConfirm || !tier.Allows(row.OpenFromTier) {
+		if err != nil || row.RejectPlan || !tier.Allows(row.OpenFromTier) {
+			return &TurnResult{}, nil
+		}
+		if row.RequiresBatchConfirm && !confirmWrites {
 			return &TurnResult{}, nil
 		}
 		summary := strings.TrimSpace(seg.Normalized)
@@ -69,13 +74,16 @@ func (p *Plans) handleTurnAutoRead(ctx context.Context, in TurnInput) (*TurnResu
 		SegmenterVersion: proposal.SegmenterVersion,
 		Status:           PlanAwaitingConfirmation,
 		ExecutionMode:    ModeLive,
-		WritePlan:        false,
+		WritePlan:        confirmWrites,
 		SafeSummary:      text,
 		Steps:            drafts,
 	})
 	if err != nil {
 		return nil, err
 	}
+	if confirmWrites {
+		return &TurnResult{Plan: created}, nil
+	}
 	confirmed, err := p.DecidePlan(ctx, DecisionCommand{
 		Principal:        in.Principal,
 		PlanID:           created.PlanID,

internal/app/intentflow/turn_test.go

@@ -78,3 +78,80 @@ func TestQueryGuardRejectsGenerate(t *testing.T) {
 		t.Fatal("task.run must be forbidden by QueryGuard")
 	}
 }
+
+func TestHandleTurnConfirmPersistsAwaitingWithoutExecute(t *testing.T) {
+	intentflow.SetFeatureFlagsForTest(intentflow.FeatureFlags{PlanTier: intentflow.TierConfirm})
+	t.Cleanup(intentflow.ResetFeatureFlagsForTest)
+	p, _, owner := testPlans()
+	ctx := context.Background()
+	res, err := p.HandleTurn(ctx, intentflow.TurnInput{
+		Principal:     principal(),
+		Text:          "查一下相关法规，然后帮我生成报价单",
+		ClientTurnID:  "turn-cf-1",
+		RequestDigest: "d-cf-1",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res == nil || res.Plan == nil {
+		t.Fatal("confirm tier must persist a write plan")
+	}
+	if res.Plan.Status != intentflow.PlanAwaitingConfirmation {
+		t.Fatalf("status %s", res.Plan.Status)
+	}
+	if len(res.Plan.Steps) < 2 {
+		t.Fatalf("expected query+generate steps, got %+v", res.Plan.Steps)
+	}
+	hasGenerate := false
+	for _, s := range res.Plan.Steps {
+		if s.ActionKey == "task.run" || s.ActionKey == "task.run.adhoc" {
+			hasGenerate = true
+		}
+		if s.Status != intentflow.StepPending {
+			t.Fatalf("confirm must not execute before decision, step=%+v", s)
+		}
+	}
+	if !hasGenerate {
+		t.Fatalf("expected a generate step, plan=%+v", res.Plan)
+	}
+	if owner.ExecCount("") != 0 {
+		t.Fatal("no owner exec expected before confirm")
+	}
+	again, err := p.HandleTurn(ctx, intentflow.TurnInput{
+		Principal:     principal(),
+		Text:          "查一下相关法规，然后帮我生成报价单",
+		ClientTurnID:  "turn-cf-1",
+		RequestDigest: "d-cf-1",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if again.Plan.PlanID != res.Plan.PlanID {
+		t.Fatal("same turn must reuse the awaiting plan")
+	}
+}
+
+func TestHandleTurnConfirmRejectsSecondActiveWritePlan(t *testing.T) {
+	intentflow.SetFeatureFlagsForTest(intentflow.FeatureFlags{PlanTier: intentflow.TierConfirm})
+	t.Cleanup(intentflow.ResetFeatureFlagsForTest)
+	p, _, _ := testPlans()
+	ctx := context.Background()
+	first, err := p.HandleTurn(ctx, intentflow.TurnInput{
+		Principal:     principal(),
+		Text:          "查一下相关法规，然后帮我生成报价单",
+		ClientTurnID:  "turn-cf-a",
+		RequestDigest: "d-cf-a",
+	})
+	if err != nil || first == nil || first.Plan == nil {
+		t.Fatalf("first plan: %+v %v", first, err)
+	}
+	_, err = p.HandleTurn(ctx, intentflow.TurnInput{
+		Principal:     principal(),
+		Text:          "查一下相关法规，然后帮我生成报价单",
+		ClientTurnID:  "turn-cf-b",
+		RequestDigest: "d-cf-b",
+	})
+	if err != intentflow.ErrPlanDecisionConflict {
+		t.Fatalf("second write plan must conflict, got %v", err)
+	}
+}

internal/app/plan_owner.go

@@ -0,0 +1,146 @@
+package app
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"sync"
+
+	"chat-agent-system/internal/app/intentflow"
+	"chat-agent-system/internal/bizdev"
+	"chat-agent-system/internal/taskrun"
+)
+
+// TaskRunAdapter is the U8 generate adapter. It records a stable idempotency
+// key and result reference without changing single-intent taskrun HTTP.
+type TaskRunAdapter struct {
+	mu       sync.Mutex
+	receipts map[string]intentflow.Receipt
+	starts   int
+	Denied   bool
+	Starter  func(ctx context.Context, exec intentflow.StepExec) (resultRef string, err error)
+}
+
+func NewTaskRunAdapter() *TaskRunAdapter {
+	return &TaskRunAdapter{receipts: map[string]intentflow.Receipt{}}
+}
+
+func (a *TaskRunAdapter) Authorize(_ context.Context, exec intentflow.StepExec) error {
+	if a != nil && a.Denied {
+		return intentflow.ErrPlanActionForbidden
+	}
+	if exec.ActionKey != "task.run" && exec.ActionKey != "task.run.adhoc" {
+		return intentflow.ErrPlanActionForbidden
+	}
+	return nil
+}
+
+func (a *TaskRunAdapter) Execute(ctx context.Context, exec intentflow.StepExec) (intentflow.Receipt, error) {
+	if err := a.Authorize(ctx, exec); err != nil {
+		return intentflow.Receipt{}, err
+	}
+	a.mu.Lock()
+	if rec, ok := a.receipts[exec.IdempotencyKey]; ok {
+		a.mu.Unlock()
+		if rec.ActionDigest != exec.ActionDigest {
+			return intentflow.Receipt{}, intentflow.ErrPlanDecisionConflict
+		}
+		return rec, nil
+	}
+	a.mu.Unlock()
+	ref := "taskrun:" + exec.IdempotencyKey
+	if a.Starter != nil {
+		got, err := a.Starter(ctx, exec)
+		if err != nil {
+			return intentflow.Receipt{}, err
+		}
+		if got != "" {
+			ref = got
+		}
+	}
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	if rec, ok := a.receipts[exec.IdempotencyKey]; ok {
+		if rec.ActionDigest != exec.ActionDigest {
+			return intentflow.Receipt{}, intentflow.ErrPlanDecisionConflict
+		}
+		return rec, nil
+	}
+	a.starts++
+	rec := intentflow.Receipt{
+		IdempotencyKey: exec.IdempotencyKey,
+		ActionDigest:   exec.ActionDigest,
+		ResultRef:      ref,
+		ResultDigest:   "task:" + exec.ActionDigest,
+		SafeSummary:    "task.run accepted",
+	}
+	if a.receipts == nil {
+		a.receipts = map[string]intentflow.Receipt{}
+	}
+	a.receipts[exec.IdempotencyKey] = rec
+	return rec, nil
+}
+
+// LiveTaskStarter claims a taskrun with the plan step idempotency key.
+// When live is false it only occupies the run row and does not start PI.
+func LiveTaskStarter(mgr *taskrun.Manager, live bool) func(context.Context, intentflow.StepExec) (string, error) {
+	return func(ctx context.Context, exec intentflow.StepExec) (string, error) {
+		if mgr == nil {
+			return "", fmt.Errorf("task generate adapter: taskrun manager missing")
+		}
+		if existing, err := mgr.FindByIdempotencyKey(ctx, exec.IdempotencyKey); err != nil {
+			return "", err
+		} else if existing != nil && existing.RunID != "" {
+			return existing.RunID, nil
+		}
+		taskKey := "intake"
+		if exec.ActionKey == "task.run.adhoc" {
+			taskKey = "adhoc"
+		}
+		rec := &taskrun.RunRecord{
+			RunID:          planRunID(exec.IdempotencyKey),
+			SessionID:      exec.Principal.SessionID,
+			ProjectID:      exec.Principal.ProjectID,
+			UserID:         exec.Principal.UserID,
+			IdempotencyKey: exec.IdempotencyKey,
+			TaskKey:        taskKey,
+			Status:         "pending",
+		}
+		if !live {
+			claimed, _, err := mgr.ClaimRun(ctx, rec)
+			if err != nil {
+				return "", err
+			}
+			return claimed.RunID, nil
+		}
+		created, err := mgr.CreateAndStart(ctx, rec.RunID, rec, bizdev.TaskRunRequest{
+			TaskKey:   taskKey,
+			UserID:    exec.Principal.UserID,
+			ProjectID: exec.Principal.ProjectID,
+			SessionID: exec.Principal.SessionID,
+		}, bizdev.StageRunRequest{}, "")
+		if err != nil {
+			return "", err
+		}
+		return created.RunID, nil
+	}
+}
+
+func planRunID(idempotencyKey string) string {
+	sum := sha256.Sum256([]byte(idempotencyKey))
+	return "run_plan_" + hex.EncodeToString(sum[:8])
+}
+
+func (a *TaskRunAdapter) LookupReceipt(_ context.Context, idempotencyKey string) (intentflow.Receipt, bool, error) {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	rec, ok := a.receipts[idempotencyKey]
+	return rec, ok, nil
+}
+
+func (a *TaskRunAdapter) StartCount() int {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	return a.starts
+}

internal/app/plan_owner_test.go

@@ -0,0 +1,128 @@
+package app
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/app/intentflow"
+	"chat-agent-system/internal/taskrun"
+)
+
+func TestTaskRunAdapterIdempotentExecute(t *testing.T) {
+	a := NewTaskRunAdapter()
+	exec := intentflow.StepExec{
+		ActionKey: "task.run", IdempotencyKey: "idem-1", ActionDigest: "d1",
+		Principal: intentflow.Principal{UserID: 1, ProjectID: 2, SessionID: "s", Environment: "dev"},
+	}
+	first, err := a.Execute(context.Background(), exec)
+	if err != nil {
+		t.Fatal(err)
+	}
+	second, err := a.Execute(context.Background(), exec)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if first.ResultRef != second.ResultRef {
+		t.Fatalf("replay ref %s vs %s", first.ResultRef, second.ResultRef)
+	}
+	if a.StartCount() != 1 {
+		t.Fatalf("starts = %d", a.StartCount())
+	}
+}
+
+func TestTaskRunAdapterRejectsQueryKey(t *testing.T) {
+	a := NewTaskRunAdapter()
+	_, err := a.Execute(context.Background(), intentflow.StepExec{ActionKey: "regulation.query", Principal: intentflow.Principal{UserID: 1}})
+	if err != intentflow.ErrPlanActionForbidden {
+		t.Fatalf("got %v", err)
+	}
+}
+
+func TestTaskRunAdapterDeniedAfterRevoke(t *testing.T) {
+	a := NewTaskRunAdapter()
+	a.Denied = true
+	_, err := a.Execute(context.Background(), intentflow.StepExec{ActionKey: "task.run", Principal: intentflow.Principal{UserID: 1}})
+	if err != intentflow.ErrPlanActionForbidden {
+		t.Fatalf("got %v", err)
+	}
+}
+
+func TestConfirmThenMuxExecutesGenerateOnce(t *testing.T) {
+	intentflow.SetFeatureFlagsForTest(intentflow.FeatureFlags{PlanTier: intentflow.TierConfirm})
+	t.Cleanup(intentflow.ResetFeatureFlagsForTest)
+	query := intentflow.NewFakeActionOwner()
+	task := NewTaskRunAdapter()
+	t0 := time.Date(2026, 9, 16, 12, 0, 0, 0, time.UTC)
+	p := intentflow.NewPlans(&intentflow.SerialUoW{}, intentflow.NewMemoryStore(), intentflow.Options{
+		Owner: intentflow.OwnerMux{Query: intentflow.QueryGuard{Inner: query}, Task: task},
+		Now:   func() time.Time { return t0 },
+	})
+	ctx := context.Background()
+	prin := intentflow.Principal{UserID: 7, Environment: "dev", ProjectID: 42, SessionID: "sess-u8"}
+	created, err := p.HandleTurn(ctx, intentflow.TurnInput{
+		Principal: prin, Text: "查一下相关法规，然后帮我生成报价单",
+		ClientTurnID: "turn-u8", RequestDigest: "d-u8",
+	})
+	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")
+	}
+	confirmed, err := p.DecidePlan(ctx, intentflow.DecisionCommand{
+		Principal: prin, PlanID: created.Plan.PlanID, ExpectedRevision: created.Plan.Revision,
+		DecisionNonce: "u8-n", DecisionDigest: "u8-d", Kind: intentflow.DecisionConfirm,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	view, err := p.AdvanceEligible(ctx, "u8-worker", confirmed.PlanID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if view == nil {
+		t.Fatal("expected advanced plan")
+	}
+	if task.StartCount() != 1 {
+		t.Fatalf("generate starts = %d, plan=%+v", task.StartCount(), view)
+	}
+	_, err = p.AdvanceEligible(ctx, "u8-worker", confirmed.PlanID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if task.StartCount() != 1 {
+		t.Fatalf("replay must not restart generate, starts=%d", task.StartCount())
+	}
+}
+
+func TestLiveTaskStarterReplaysIdempotencyKey(t *testing.T) {
+	mgr := taskrun.NewManager(nil, nil)
+	start := LiveTaskStarter(mgr, false)
+	exec := intentflow.StepExec{
+		ActionKey: "task.run", IdempotencyKey: "plan-key-1", ActionDigest: "d",
+		Principal: intentflow.Principal{UserID: 3, ProjectID: 9, SessionID: "sess-live"},
+	}
+	a, err := start(context.Background(), exec)
+	if err != nil {
+		t.Fatal(err)
+	}
+	b, err := start(context.Background(), exec)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if a == "" || a != b {
+		t.Fatalf("replay run id %q vs %q", a, b)
+	}
+}
+
+func TestTaskGenerateRealPIStopGate(t *testing.T) {
+	if os.Getenv("PI_AGENT_RUNNER_URL") == "" {
+		t.Skip("stop-gate: PI_AGENT_RUNNER_URL not set; real PI generate is not claimed")
+	}
+	t.Skip("live PI generate path is gated; unit evidence is TaskRunAdapter + ClaimRun idempotency")
+}