diff --git a/internal/app/app.go b/internal/app/app.go index 58b0f081..015827d8 100644 --- a/internal/app/app.go +++ b/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{}, }) } diff --git a/internal/app/intentflow/mux.go b/internal/app/intentflow/mux.go new file mode 100644 index 00000000..584c6572 --- /dev/null +++ b/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) + } +} diff --git a/internal/app/intentflow/mux_test.go b/internal/app/intentflow/mux_test.go new file mode 100644 index 00000000..ebba5f8d --- /dev/null +++ b/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) + } +} diff --git a/internal/app/intentflow/plans.go b/internal/app/intentflow/plans.go index 7b671590..2a865fd6 100644 --- a/internal/app/intentflow/plans.go +++ b/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 diff --git a/internal/app/intentflow/query.go b/internal/app/intentflow/query.go index 403de9ad..728d6200 100644 --- a/internal/app/intentflow/query.go +++ b/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) + } +} diff --git a/internal/app/intentflow/turn.go b/internal/app/intentflow/turn.go index a995e928..9963bb58 100644 --- a/internal/app/intentflow/turn.go +++ b/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, diff --git a/internal/app/intentflow/turn_test.go b/internal/app/intentflow/turn_test.go index ffe19422..c46c61a6 100644 --- a/internal/app/intentflow/turn_test.go +++ b/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) + } +} diff --git a/internal/app/plan_owner.go b/internal/app/plan_owner.go new file mode 100644 index 00000000..321fb825 --- /dev/null +++ b/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 +} diff --git a/internal/app/plan_owner_test.go b/internal/app/plan_owner_test.go new file mode 100644 index 00000000..9dafe4cd --- /dev/null +++ b/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") +}