feat(intentflow): persist plans with CAS lease and fencing

Own plan/step/decision writes behind UoW so two workers cannot both execute the same step.

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

internal/app/intentflow/doc.go

@@ -9,5 +9,6 @@
 //   - Confirmation is not authorization; each step re-checks the current principal.
 //   - PostgreSQL plan/step/decision rows are authoritative; cards and SSE are projections.
 //   - internal/intent stays a pure rule module (intent -> maturity).
-//   - This package must not import docgov, taskrun, coordinator, server, or pgx.
+//   - This package must not import docgov, taskrun, coordinator, or server.
+//   - Only the owner repository unwraps pgx via app/uow (see unwrap_owners).
 package intentflow

internal/app/intentflow/errors.go

@@ -14,6 +14,8 @@ var (
 	ErrUnauthenticated           = errors.New("ERR_UNAUTHENTICATED")
 )
 
+var errTurnExists = errors.New("intentflow: turn exists")
+
 // ErrorCode returns the stable machine code for a sentinel, or empty.
 func ErrorCode(err error) string {
 	switch {

internal/app/intentflow/memory.go

@@ -0,0 +1,322 @@
+package intentflow
+
+import (
+	"context"
+	"fmt"
+	"slices"
+	"sync"
+	"time"
+
+	"chat-agent-system/internal/app/uow"
+)
+
+type serialScope struct{ ctx context.Context }
+
+func (s serialScope) Context() context.Context { return s.ctx }
+
+// SerialUoW serializes commands. Memory tests use it so CAS races have one winner.
+type SerialUoW struct{ mu sync.Mutex }
+
+func (s *SerialUoW) Within(ctx context.Context, fn func(uow.TxScope) error) error {
+	if s == nil {
+		return fmt.Errorf("intentflow: serial uow required")
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return fn(serialScope{ctx: ctx})
+}
+
+type MemoryStore struct {
+	mu        sync.Mutex
+	plans     map[string]*Plan
+	turn      map[string]string
+	decisions map[string]*Decision
+}
+
+func NewMemoryStore() *MemoryStore {
+	return &MemoryStore{
+		plans:     map[string]*Plan{},
+		turn:      map[string]string{},
+		decisions: map[string]*Decision{},
+	}
+}
+
+func turnKey(env, session, turn string) string {
+	return env + "\x1f" + session + "\x1f" + turn
+}
+
+func nonceKey(planID, nonce string) string {
+	return planID + "\x1f" + nonce
+}
+
+func (m *MemoryStore) InsertPlan(ctx context.Context, scope uow.TxScope, plan *Plan) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if plan == nil || plan.PlanID == "" {
+		return fmt.Errorf("intentflow: plan required")
+	}
+	tk := turnKey(plan.Principal.Environment, plan.Principal.SessionID, plan.ClientTurnID)
+	if existingID, ok := m.turn[tk]; ok {
+		_ = existingID
+		return errTurnExists
+	}
+	if plan.WritePlan && isActiveStatus(plan.Status) {
+		for _, p := range m.plans {
+			if sameScope(p.Principal, plan.Principal) && p.WritePlan && isActiveStatus(p.Status) {
+				return uow.ErrBusy
+			}
+		}
+	}
+	cloned := clonePlan(plan)
+	m.plans[plan.PlanID] = cloned
+	m.turn[tk] = plan.PlanID
+	return nil
+}
+
+func (m *MemoryStore) GetPlan(ctx context.Context, scope uow.TxScope, planID string) (*Plan, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	p := m.plans[planID]
+	if p == nil {
+		return nil, nil
+	}
+	return clonePlan(p), nil
+}
+
+func (m *MemoryStore) GetPlanForUpdate(ctx context.Context, scope uow.TxScope, planID string) (*Plan, error) {
+	return m.GetPlan(ctx, scope, planID)
+}
+
+func (m *MemoryStore) GetByTurn(ctx context.Context, scope uow.TxScope, env, sessionID, clientTurnID string) (*Plan, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	id := m.turn[turnKey(env, sessionID, clientTurnID)]
+	if id == "" {
+		return nil, nil
+	}
+	return clonePlan(m.plans[id]), nil
+}
+
+func (m *MemoryStore) GetActiveWrite(ctx context.Context, scope uow.TxScope, p Principal) (*Plan, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	for _, plan := range m.plans {
+		if sameScope(plan.Principal, p) && plan.WritePlan && isActiveStatus(plan.Status) {
+			return clonePlan(plan), nil
+		}
+	}
+	return nil, nil
+}
+
+func (m *MemoryStore) GetActive(ctx context.Context, scope uow.TxScope, p Principal) (*Plan, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	var best *Plan
+	for _, plan := range m.plans {
+		if !sameScope(plan.Principal, p) || !isActiveStatus(plan.Status) {
+			continue
+		}
+		if best == nil || plan.UpdatedAt.After(best.UpdatedAt) {
+			cp := clonePlan(plan)
+			best = cp
+		}
+	}
+	return best, nil
+}
+
+func (m *MemoryStore) CASPlan(ctx context.Context, scope uow.TxScope, planID string, expectedRev int64, from []PlanStatus, to PlanStatus, now time.Time) (int64, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	p := m.plans[planID]
+	if p == nil || p.Revision != expectedRev || !slices.Contains(from, p.Status) {
+		return 0, nil
+	}
+	p.Status = to
+	p.Revision++
+	p.UpdatedAt = now
+	return p.Revision, nil
+}
+
+func (m *MemoryStore) InsertDecision(ctx context.Context, scope uow.TxScope, d *Decision) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	k := nonceKey(d.PlanID, d.Nonce)
+	if _, ok := m.decisions[k]; ok {
+		return uow.ErrDigestConflict
+	}
+	cp := *d
+	cp.StepIDs = slices.Clone(d.StepIDs)
+	m.decisions[k] = &cp
+	return nil
+}
+
+func (m *MemoryStore) GetDecisionByNonce(ctx context.Context, scope uow.TxScope, planID, nonce string) (*Decision, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	d := m.decisions[nonceKey(planID, nonce)]
+	if d == nil {
+		return nil, nil
+	}
+	cp := *d
+	cp.StepIDs = slices.Clone(d.StepIDs)
+	return &cp, nil
+}
+
+func (m *MemoryStore) ReplaceSteps(ctx context.Context, scope uow.TxScope, planID string, steps []Step) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	p := m.plans[planID]
+	if p == nil {
+		return fmt.Errorf("intentflow: plan %s not found", planID)
+	}
+	p.Steps = cloneSteps(steps)
+	p.UpdatedAt = time.Now().UTC()
+	return nil
+}
+
+func (m *MemoryStore) ClaimStep(ctx context.Context, scope uow.TxScope, planID, stepID, worker string, until, now time.Time, idempotencyKey string) (*Step, int64, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	p := m.plans[planID]
+	if p == nil {
+		return nil, 0, nil
+	}
+	for i := range p.Steps {
+		s := &p.Steps[i]
+		if s.StepID != stepID {
+			continue
+		}
+		claimable := s.Status == StepEligible ||
+			(s.Status == StepRunning && (s.LeaseUntil == nil || s.LeaseUntil.Before(now)))
+		if !claimable {
+			return nil, 0, nil
+		}
+		p.LastFencingToken++
+		s.Status = StepRunning
+		s.LeaseOwner = worker
+		untilCopy := until
+		s.LeaseUntil = &untilCopy
+		s.Attempt++
+		s.Version++
+		s.FencingToken = p.LastFencingToken
+		if s.IdempotencyKey == "" {
+			s.IdempotencyKey = idempotencyKey
+		}
+		s.UpdatedAt = now
+		p.UpdatedAt = now
+		return cloneStep(s), s.FencingToken, nil
+	}
+	return nil, 0, nil
+}
+
+func (m *MemoryStore) WriteStepReceipt(ctx context.Context, scope uow.TxScope, worker string, token int64, step Step, now time.Time) (bool, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	p := m.plans[step.PlanID]
+	if p == nil {
+		return false, nil
+	}
+	for i := range p.Steps {
+		s := &p.Steps[i]
+		if s.StepID != step.StepID {
+			continue
+		}
+		if s.Status == StepSucceeded && s.ResultDigest == step.ResultDigest {
+			return true, nil
+		}
+		if s.Status != StepRunning || s.LeaseOwner != worker || s.FencingToken != token {
+			return false, nil
+		}
+		s.Status = StepSucceeded
+		s.ResultRef = step.ResultRef
+		s.ResultDigest = step.ResultDigest
+		s.SafeSummary = step.SafeSummary
+		s.LeaseOwner = ""
+		s.LeaseUntil = nil
+		s.Version++
+		s.UpdatedAt = now
+		p.UpdatedAt = now
+		return true, nil
+	}
+	return false, nil
+}
+
+func (m *MemoryStore) ExpireWaiting(ctx context.Context, scope uow.TxScope, now time.Time) (int, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	n := 0
+	for _, p := range m.plans {
+		if !isWaitingStatus(p.Status) || !now.After(p.ExpiresAt) {
+			continue
+		}
+		p.Status = PlanExpired
+		p.Revision++
+		p.UpdatedAt = now
+		for i := range p.Steps {
+			if p.Steps[i].Status == StepPending || p.Steps[i].Status == StepEligible {
+				p.Steps[i].Status = StepCancelled
+				p.Steps[i].UpdatedAt = now
+			}
+		}
+		n++
+	}
+	return n, nil
+}
+
+func (m *MemoryStore) PurgeClosed(ctx context.Context, scope uow.TxScope, before time.Time) (int, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	n := 0
+	for id, p := range m.plans {
+		switch p.Status {
+		case PlanCompleted, PlanFailed, PlanCancelled, PlanExpired, PlanSuperseded:
+		default:
+			continue
+		}
+		if !p.UpdatedAt.Before(before) {
+			continue
+		}
+		delete(m.turn, turnKey(p.Principal.Environment, p.Principal.SessionID, p.ClientTurnID))
+		for k, d := range m.decisions {
+			if d.PlanID == id {
+				delete(m.decisions, k)
+			}
+		}
+		delete(m.plans, id)
+		n++
+	}
+	return n, nil
+}
+
+func sameScope(a, b Principal) bool {
+	return a.Environment == b.Environment && a.ProjectID == b.ProjectID && a.SessionID == b.SessionID && a.UserID == b.UserID
+}
+
+func clonePlan(p *Plan) *Plan {
+	if p == nil {
+		return nil
+	}
+	cp := *p
+	cp.DisplaySnapshot = slices.Clone(p.DisplaySnapshot)
+	cp.Steps = cloneSteps(p.Steps)
+	return &cp
+}
+
+func cloneSteps(in []Step) []Step {
+	out := make([]Step, len(in))
+	for i := range in {
+		out[i] = *cloneStep(&in[i])
+	}
+	return out
+}
+
+func cloneStep(s *Step) *Step {
+	cp := *s
+	cp.Slots = slices.Clone(s.Slots)
+	cp.DependsOn = slices.Clone(s.DependsOn)
+	if s.LeaseUntil != nil {
+		t := *s.LeaseUntil
+		cp.LeaseUntil = &t
+	}
+	return &cp
+}

internal/app/intentflow/owner.go

@@ -0,0 +1,88 @@
+package intentflow
+
+import (
+	"context"
+	"sync"
+)
+
+// ActionOwner executes one step. Old fencing tokens must be rejected here,
+// not only in the plan store. The owner also stores receipts by idempotency key.
+type ActionOwner interface {
+	Execute(ctx context.Context, exec StepExec) (Receipt, error)
+	LookupReceipt(ctx context.Context, idempotencyKey string) (Receipt, bool, error)
+}
+
+// FenceObserver is optional; ClaimStep notifies it so fakes can reject stale tokens.
+type FenceObserver interface {
+	ObserveFence(stepID string, token int64)
+}
+
+// FakeActionOwner is the U3 test double. It records executions and receipts
+// without touching business tables.
+type FakeActionOwner struct {
+	mu        sync.Mutex
+	execCount map[string]int
+	receipts  map[string]Receipt
+	fences    map[string]int64
+	FailWrite bool
+}
+
+func NewFakeActionOwner() *FakeActionOwner {
+	return &FakeActionOwner{
+		execCount: map[string]int{},
+		receipts:  map[string]Receipt{},
+		fences:    map[string]int64{},
+	}
+}
+
+func (f *FakeActionOwner) ObserveFence(stepID string, token int64) {
+	if f == nil {
+		return
+	}
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.fences[stepID] = token
+}
+
+func (f *FakeActionOwner) Execute(ctx context.Context, exec StepExec) (Receipt, error) {
+	if f == nil {
+		return Receipt{}, ErrPlanActionForbidden
+	}
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if cur, ok := f.fences[exec.StepID]; ok && exec.FencingToken != cur {
+		return Receipt{}, ErrPlanActionVersionConflict
+	}
+	if existing, ok := f.receipts[exec.IdempotencyKey]; ok {
+		if existing.ActionDigest != exec.ActionDigest {
+			return Receipt{}, ErrPlanDecisionConflict
+		}
+		return existing, nil
+	}
+	f.execCount[exec.IdempotencyKey]++
+	rec := Receipt{
+		IdempotencyKey: exec.IdempotencyKey,
+		ActionDigest:   exec.ActionDigest,
+		ResultRef:      "fake:" + exec.StepID,
+		ResultDigest:   "digest:" + exec.ActionDigest,
+		SafeSummary:    "validated_only",
+	}
+	f.receipts[exec.IdempotencyKey] = rec
+	return rec, nil
+}
+
+func (f *FakeActionOwner) LookupReceipt(ctx context.Context, idempotencyKey string) (Receipt, bool, error) {
+	if f == nil {
+		return Receipt{}, false, nil
+	}
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	rec, ok := f.receipts[idempotencyKey]
+	return rec, ok, nil
+}
+
+func (f *FakeActionOwner) ExecCount(idempotencyKey string) int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.execCount[idempotencyKey]
+}

internal/app/intentflow/pg_test.go

@@ -0,0 +1,113 @@
+package intentflow_test
+
+import (
+	"context"
+	"errors"
+	"os"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/app/intentflow"
+	"chat-agent-system/internal/app/uow"
+	"chat-agent-system/internal/db"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+func init() {
+	if _, err := os.Stat("deploy/postgres/goose"); err == nil {
+		return
+	}
+	for _, cand := range []string{"../../..", "../../../.."} {
+		if fi, err := os.Stat(cand + "/deploy/postgres/goose"); err == nil && fi.IsDir() {
+			_ = os.Chdir(cand)
+			return
+		}
+	}
+}
+
+func TestPlansPG_CreateClaimAndFence(t *testing.T) {
+	if testing.Short() {
+		t.Skip("intentflow PG is not part of -short")
+	}
+	dsn := db.SkipIfNoTestDatabaseURL(t)
+	db.RefuseSharedEnvDSN(t, dsn)
+	ctx := context.Background()
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(pool.Close)
+	if err := db.RunMigrationsOnceForTest(ctx, pool, dsn); err != nil {
+		t.Fatalf("migrate: %v", err)
+	}
+
+	now := time.Date(2026, 9, 15, 15, 0, 0, 0, time.UTC)
+	clock := now
+	owner := intentflow.NewFakeActionOwner()
+	plans := intentflow.NewPlans(uow.NewPG(pool), intentflow.NewRepository(), intentflow.Options{
+		Owner: owner,
+		Now:   func() time.Time { return clock },
+		Lease: 5 * time.Second,
+	})
+	prin := intentflow.Principal{UserID: 9, Environment: "ci", ProjectID: 1001, SessionID: "sess-pg-u3"}
+	created, err := plans.CreatePlan(ctx, intentflow.CreateCommand{
+		Principal: prin, ClientTurnID: "pg-turn-1", RequestDigest: "pg-d1",
+		WritePlan: true,
+		Steps: []intentflow.StepDraft{
+			{ActionKey: "regulation.query", ActionDigest: "q"},
+			{ActionKey: "task.run", ActionDigest: "w"},
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `DELETE FROM intent_plans WHERE plan_id=$1`, created.PlanID)
+	})
+	again, err := plans.CreatePlan(ctx, intentflow.CreateCommand{
+		Principal: prin, ClientTurnID: "pg-turn-1", RequestDigest: "pg-d1",
+		WritePlan: true,
+		Steps: []intentflow.StepDraft{
+			{ActionKey: "regulation.query", ActionDigest: "q"},
+			{ActionKey: "task.run", ActionDigest: "w"},
+		},
+	})
+	if err != nil || again.PlanID != created.PlanID {
+		t.Fatalf("idempotent create: %v %+v", err, again)
+	}
+	confirmed, err := plans.DecidePlan(ctx, intentflow.DecisionCommand{
+		Principal: prin, PlanID: created.PlanID, ExpectedRevision: created.Revision,
+		DecisionNonce: "pg-n", DecisionDigest: "pg-dec", Kind: intentflow.DecisionConfirm,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	a, err := plans.ClaimStep(ctx, "pg-a", created.PlanID, confirmed.Steps[0].StepID)
+	if err != nil || a == nil {
+		t.Fatalf("claim a: %v", err)
+	}
+	b, err := plans.ClaimStep(ctx, "pg-b", created.PlanID, confirmed.Steps[0].StepID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if b != nil {
+		t.Fatal("second worker must not own a live lease")
+	}
+	clock = clock.Add(time.Minute)
+	c, err := plans.ClaimStep(ctx, "pg-c", created.PlanID, confirmed.Steps[0].StepID)
+	if err != nil || c == nil {
+		t.Fatalf("takeover: %v", err)
+	}
+	if _, err := plans.ExecuteClaimed(ctx, a); !errors.Is(err, intentflow.ErrPlanActionVersionConflict) {
+		t.Fatalf("old fence: %v", err)
+	}
+	if _, err := plans.ExecuteClaimed(ctx, c); err != nil {
+		t.Fatal(err)
+	}
+	other := prin
+	other.Environment = "dev"
+	if _, err := plans.GetPlan(ctx, other, created.PlanID); !errors.Is(err, intentflow.ErrPlanNotFoundOrHidden) {
+		t.Fatalf("env isolation: %v", err)
+	}
+}

internal/app/intentflow/plans.go

@@ -0,0 +1,563 @@
+package intentflow
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"slices"
+	"time"
+
+	"chat-agent-system/internal/app/uow"
+)
+
+const (
+	defaultLeaseTTL = 15 * time.Second
+	defaultPlanTTL  = 30 * time.Minute
+)
+
+// Plans is the owner implementation of Service plus worker/recovery methods.
+type Plans struct {
+	uow   uow.UnitOfWork
+	store Store
+	owner ActionOwner
+	now   func() time.Time
+	newID func(prefix string) string
+	lease time.Duration
+	ttl   time.Duration
+}
+
+type Options struct {
+	Owner   ActionOwner
+	Now     func() time.Time
+	NewID   func(prefix string) string
+	Lease   time.Duration
+	PlanTTL time.Duration
+}
+
+func NewPlans(unit uow.UnitOfWork, store Store, opt Options) *Plans {
+	now := opt.Now
+	if now == nil {
+		now = func() time.Time { return time.Now().UTC() }
+	}
+	newID := opt.NewID
+	if newID == nil {
+		newID = randomID
+	}
+	lease := opt.Lease
+	if lease <= 0 {
+		lease = defaultLeaseTTL
+	}
+	ttl := opt.PlanTTL
+	if ttl <= 0 {
+		ttl = defaultPlanTTL
+	}
+	return &Plans{uow: unit, store: store, owner: opt.Owner, now: now, newID: newID, lease: lease, ttl: ttl}
+}
+
+func randomID(prefix string) string {
+	var b [16]byte
+	if _, err := rand.Read(b[:]); err != nil {
+		return prefix + fmt.Sprintf("%d", time.Now().UnixNano())
+	}
+	return prefix + hex.EncodeToString(b[:])
+}
+
+func requirePrincipal(p Principal) error {
+	if p.UserID <= 0 || p.ProjectID <= 0 || p.SessionID == "" || p.Environment == "" {
+		return ErrUnauthenticated
+	}
+	return nil
+}
+
+func visibleTo(plan *Plan, p Principal) bool {
+	return plan != nil && sameScope(plan.Principal, p)
+}
+
+// HandleTurn stays a no-op while MULTI_INTENT_PLAN_TIER is legacy/shadow (default).
+// U7 will map utterances onto CreatePlan; chat must not grow a second owner.
+func (p *Plans) HandleTurn(ctx context.Context, in TurnInput) (*TurnResult, error) {
+	if err := requirePrincipal(in.Principal); err != nil {
+		return nil, err
+	}
+	return &TurnResult{}, nil
+}
+
+func (p *Plans) GetPlan(ctx context.Context, principal Principal, planID string) (*PlanView, error) {
+	if err := requirePrincipal(principal); err != nil {
+		return nil, err
+	}
+	var view *PlanView
+	err := p.uow.Within(ctx, func(scope uow.TxScope) error {
+		plan, err := p.store.GetPlan(ctx, scope, planID)
+		if err != nil {
+			return err
+		}
+		if !visibleTo(plan, principal) {
+			return ErrPlanNotFoundOrHidden
+		}
+		view = toView(plan)
+		return nil
+	})
+	return view, err
+}
+
+func (p *Plans) GetActivePlan(ctx context.Context, principal Principal) (*PlanView, error) {
+	if err := requirePrincipal(principal); err != nil {
+		return nil, err
+	}
+	var view *PlanView
+	err := p.uow.Within(ctx, func(scope uow.TxScope) error {
+		plan, err := p.store.GetActive(ctx, scope, principal)
+		if err != nil {
+			return err
+		}
+		if plan == nil {
+			return nil
+		}
+		view = toView(plan)
+		return nil
+	})
+	return view, err
+}
+
+func (p *Plans) CreatePlan(ctx context.Context, in CreateCommand) (*PlanView, error) {
+	if err := requirePrincipal(in.Principal); err != nil {
+		return nil, err
+	}
+	if in.ClientTurnID == "" || in.RequestDigest == "" {
+		return nil, ErrPlanDecisionConflict
+	}
+	if len(in.Steps) == 0 || len(in.Steps) > 4 {
+		return nil, ErrPlanDependencyInvalid
+	}
+	now := p.now()
+	status := in.Status
+	if status == "" {
+		status = PlanAwaitingConfirmation
+	}
+	mode := in.ExecutionMode
+	if mode == "" {
+		mode = ModeValidatedOnly
+	}
+	expires := in.ExpiresAt
+	if expires.IsZero() {
+		expires = now.Add(p.ttl)
+	}
+	seg := in.SegmenterVersion
+	if seg == "" {
+		seg = SegmenterVersionUnknown
+	}
+	pol := in.PolicyVersion
+	if pol == "" {
+		pol = PolicyVersion
+	}
+
+	var view *PlanView
+	err := p.uow.Within(ctx, func(scope uow.TxScope) error {
+		existing, err := p.store.GetByTurn(ctx, scope, in.Principal.Environment, in.Principal.SessionID, in.ClientTurnID)
+		if err != nil {
+			return err
+		}
+		if existing != nil {
+			if existing.RequestDigest != in.RequestDigest {
+				return ErrPlanDecisionConflict
+			}
+			view = toView(existing)
+			return nil
+		}
+		if in.WritePlan {
+			active, err := p.store.GetActiveWrite(ctx, scope, in.Principal)
+			if err != nil {
+				return err
+			}
+			if active != nil {
+				view = toView(active)
+				return ErrPlanDecisionConflict
+			}
+		}
+		plan := &Plan{
+			PlanID:           p.newID("pln_"),
+			Principal:        in.Principal,
+			ClientTurnID:     in.ClientTurnID,
+			RequestDigest:    in.RequestDigest,
+			SegmenterVersion: seg,
+			PolicyVersion:    pol,
+			Status:           status,
+			Revision:         1,
+			ExpiresAt:        expires,
+			WritePlan:        in.WritePlan,
+			ExecutionMode:    mode,
+			SafeSummary:      in.SafeSummary,
+			CreatedAt:        now,
+			UpdatedAt:        now,
+		}
+		for i, d := range in.Steps {
+			mode := d.ExecutionMode
+			if mode == "" {
+				mode = plan.ExecutionMode
+			}
+			slots := d.Slots
+			if len(slots) == 0 {
+				slots = []byte("{}")
+			}
+			plan.Steps = append(plan.Steps, Step{
+				PlanID:                  plan.PlanID,
+				StepID:                  p.newID("stp_"),
+				Ordinal:                 i + 1,
+				ActionKey:               d.ActionKey,
+				Slots:                   slots,
+				DependsOn:               slices.Clone(d.DependsOn),
+				Status:                  StepPending,
+				ExpectedResourceVersion: d.ExpectedResourceVersion,
+				ActionDigest:            d.ActionDigest,
+				SafeSummary:             d.SafeSummary,
+				Version:                 1,
+				ExecutionMode:           mode,
+				CreatedAt:               now,
+				UpdatedAt:               now,
+			})
+		}
+		if err := p.store.InsertPlan(ctx, scope, plan); err != nil {
+			if errors.Is(err, errTurnExists) {
+				again, loadErr := p.store.GetByTurn(ctx, scope, in.Principal.Environment, in.Principal.SessionID, in.ClientTurnID)
+				if loadErr != nil {
+					return loadErr
+				}
+				if again != nil && again.RequestDigest == in.RequestDigest {
+					view = toView(again)
+					return nil
+				}
+				return ErrPlanDecisionConflict
+			}
+			if errors.Is(err, uow.ErrBusy) {
+				active, loadErr := p.store.GetActiveWrite(ctx, scope, in.Principal)
+				if loadErr != nil {
+					return loadErr
+				}
+				if active != nil {
+					view = toView(active)
+				}
+				return ErrPlanDecisionConflict
+			}
+			return err
+		}
+		view = toView(plan)
+		return nil
+	})
+	return view, err
+}
+
+func (p *Plans) DecidePlan(ctx context.Context, in DecisionCommand) (*PlanView, error) {
+	if err := requirePrincipal(in.Principal); err != nil {
+		return nil, err
+	}
+	if in.PlanID == "" || in.DecisionNonce == "" || in.DecisionDigest == "" {
+		return nil, ErrPlanDecisionConflict
+	}
+	now := p.now()
+	var view *PlanView
+	err := p.uow.Within(ctx, func(scope uow.TxScope) error {
+		plan, err := p.store.GetPlanForUpdate(ctx, scope, in.PlanID)
+		if err != nil {
+			return err
+		}
+		if !visibleTo(plan, in.Principal) {
+			return ErrPlanNotFoundOrHidden
+		}
+		existing, err := p.store.GetDecisionByNonce(ctx, scope, in.PlanID, in.DecisionNonce)
+		if err != nil {
+			return err
+		}
+		if existing != nil {
+			if existing.DecisionDigest != in.DecisionDigest {
+				return ErrPlanDecisionConflict
+			}
+			fresh, err := p.store.GetPlan(ctx, scope, in.PlanID)
+			if err != nil {
+				return err
+			}
+			view = toView(fresh)
+			return nil
+		}
+		if isWaitingStatus(plan.Status) && !now.Before(plan.ExpiresAt) {
+			if _, err := p.store.CASPlan(ctx, scope, plan.PlanID, plan.Revision, waitingStatuses(), PlanExpired, now); err != nil {
+				return err
+			}
+			cancelUnstarted(plan, now)
+			if err := p.store.ReplaceSteps(ctx, scope, plan.PlanID, plan.Steps); err != nil {
+				return err
+			}
+			return ErrPlanExpired
+		}
+		if in.ExpectedRevision != plan.Revision {
+			return ErrPlanStaleRevision
+		}
+		dec := &Decision{
+			DecisionID:        p.newID("dec_"),
+			PlanID:            plan.PlanID,
+			Nonce:             in.DecisionNonce,
+			PlanRevision:      plan.Revision,
+			ActorUserID:       in.Principal.UserID,
+			Kind:              in.Kind,
+			StepIDs:           slices.Clone(in.StepIDs),
+			ExcludeUnselected: in.ExcludeUnselected,
+			DecisionDigest:    in.DecisionDigest,
+			CreatedAt:         now,
+		}
+		if err := p.store.InsertDecision(ctx, scope, dec); err != nil {
+			if errors.Is(err, uow.ErrDigestConflict) || errors.Is(err, errTurnExists) {
+				again, loadErr := p.store.GetDecisionByNonce(ctx, scope, in.PlanID, in.DecisionNonce)
+				if loadErr != nil {
+					return loadErr
+				}
+				if again != nil && again.DecisionDigest == in.DecisionDigest {
+					fresh, err := p.store.GetPlan(ctx, scope, in.PlanID)
+					if err != nil {
+						return err
+					}
+					view = toView(fresh)
+					return nil
+				}
+				return ErrPlanDecisionConflict
+			}
+			return err
+		}
+		from := []PlanStatus{plan.Status}
+		var to PlanStatus
+		switch in.Kind {
+		case DecisionConfirm:
+			if plan.Status != PlanAwaitingConfirmation && plan.Status != PlanAwaitingClarification && plan.Status != PlanProposed {
+				return ErrPlanStaleRevision
+			}
+			if err := applyConfirm(plan, in.StepIDs, in.ExcludeUnselected, now); err != nil {
+				return err
+			}
+			to = PlanExecuting
+		case DecisionCancel:
+			if !isWaitingStatus(plan.Status) && plan.Status != PlanExecuting && plan.Status != PlanPartiallyCompleted {
+				return ErrPlanStaleRevision
+			}
+			cancelUnstarted(plan, now)
+			to = PlanCancelled
+		default:
+			return ErrPlanDependencyInvalid
+		}
+		newRev, err := p.store.CASPlan(ctx, scope, plan.PlanID, plan.Revision, from, to, now)
+		if err != nil {
+			return err
+		}
+		if newRev == 0 {
+			return ErrPlanStaleRevision
+		}
+		if err := p.store.ReplaceSteps(ctx, scope, plan.PlanID, plan.Steps); err != nil {
+			return err
+		}
+		plan.Status = to
+		plan.Revision = newRev
+		view = toView(plan)
+		return nil
+	})
+	return view, err
+}
+
+func applyConfirm(plan *Plan, selected []string, excludeUnselected bool, now time.Time) error {
+	byID := map[string]*Step{}
+	for i := range plan.Steps {
+		byID[plan.Steps[i].StepID] = &plan.Steps[i]
+	}
+	chosen := map[string]bool{}
+	if len(selected) == 0 {
+		for i := range plan.Steps {
+			chosen[plan.Steps[i].StepID] = true
+		}
+	} else {
+		for _, id := range selected {
+			if _, ok := byID[id]; !ok {
+				return ErrPlanDependencyInvalid
+			}
+			chosen[id] = true
+		}
+		for id := range chosen {
+			for _, dep := range byID[id].DependsOn {
+				if !chosen[dep] && byID[dep].Status != StepSucceeded && byID[dep].Status != StepSkipped {
+					return ErrPlanDependencyInvalid
+				}
+			}
+		}
+	}
+	for i := range plan.Steps {
+		s := &plan.Steps[i]
+		if chosen[s.StepID] {
+			if s.Status == StepPending || s.Status == StepEligible {
+				s.Status = StepEligible
+				s.UpdatedAt = now
+			}
+			continue
+		}
+		if excludeUnselected && (s.Status == StepPending || s.Status == StepEligible) {
+			s.Status = StepCancelled
+			s.UpdatedAt = now
+		}
+	}
+	return nil
+}
+
+func cancelUnstarted(plan *Plan, now time.Time) {
+	for i := range plan.Steps {
+		s := &plan.Steps[i]
+		if s.Status == StepPending || s.Status == StepEligible {
+			s.Status = StepCancelled
+			s.UpdatedAt = now
+		}
+	}
+}
+
+func (p *Plans) ClaimStep(ctx context.Context, workerID, planID, stepID string) (*ClaimedStep, error) {
+	if workerID == "" || planID == "" || stepID == "" {
+		return nil, ErrPlanDependencyInvalid
+	}
+	now := p.now()
+	until := now.Add(p.lease)
+	var out *ClaimedStep
+	err := p.uow.Within(ctx, func(scope uow.TxScope) error {
+		plan, err := p.store.GetPlanForUpdate(ctx, scope, planID)
+		if err != nil {
+			return err
+		}
+		if plan == nil {
+			return ErrPlanNotFoundOrHidden
+		}
+		var digest string
+		for _, s := range plan.Steps {
+			if s.StepID == stepID {
+				digest = s.ActionDigest
+				break
+			}
+		}
+		key := fmt.Sprintf("%s:%d:%s:%s", planID, plan.Revision, stepID, digest)
+		step, token, err := p.store.ClaimStep(ctx, scope, planID, stepID, workerID, until, now, key)
+		if err != nil {
+			return err
+		}
+		if step == nil {
+			return nil
+		}
+		if obs, ok := p.owner.(FenceObserver); ok {
+			obs.ObserveFence(step.StepID, token)
+		}
+		out = &ClaimedStep{Plan: *plan, Step: *step, WorkerID: workerID, FencingToken: token, IdempotencyKey: step.IdempotencyKey}
+		return nil
+	})
+	return out, err
+}
+
+func (p *Plans) ExecuteClaimed(ctx context.Context, claimed *ClaimedStep) (*PlanView, error) {
+	if claimed == nil {
+		return nil, ErrPlanNotFoundOrHidden
+	}
+	if p.owner == nil {
+		return nil, ErrPlanActionForbidden
+	}
+	rec, err := p.owner.Execute(ctx, StepExec{
+		Principal:               claimed.Plan.Principal,
+		PlanID:                  claimed.Plan.PlanID,
+		StepID:                  claimed.Step.StepID,
+		ActionKey:               claimed.Step.ActionKey,
+		IdempotencyKey:          claimed.IdempotencyKey,
+		ActionDigest:            claimed.Step.ActionDigest,
+		ExpectedResourceVersion: claimed.Step.ExpectedResourceVersion,
+		FencingToken:            claimed.FencingToken,
+	})
+	if err != nil {
+		return nil, err
+	}
+	return p.writeReceipt(ctx, claimed, rec)
+}
+
+func (p *Plans) RecoverReceipt(ctx context.Context, claimed *ClaimedStep) (*PlanView, error) {
+	if claimed == nil || p.owner == nil {
+		return nil, ErrPlanNotFoundOrHidden
+	}
+	rec, ok, err := p.owner.LookupReceipt(ctx, claimed.IdempotencyKey)
+	if err != nil {
+		return nil, err
+	}
+	if !ok {
+		return nil, ErrPlanActionVersionConflict
+	}
+	return p.writeReceipt(ctx, claimed, rec)
+}
+
+func (p *Plans) writeReceipt(ctx context.Context, claimed *ClaimedStep, rec Receipt) (*PlanView, error) {
+	now := p.now()
+	var view *PlanView
+	err := p.uow.Within(ctx, func(scope uow.TxScope) error {
+		ok, err := p.store.WriteStepReceipt(ctx, scope, claimed.WorkerID, claimed.FencingToken, Step{
+			PlanID:       claimed.Plan.PlanID,
+			StepID:       claimed.Step.StepID,
+			ResultRef:    rec.ResultRef,
+			ResultDigest: rec.ResultDigest,
+			SafeSummary:  rec.SafeSummary,
+		}, now)
+		if err != nil {
+			return err
+		}
+		if !ok {
+			return ErrPlanActionVersionConflict
+		}
+		plan, err := p.store.GetPlan(ctx, scope, claimed.Plan.PlanID)
+		if err != nil {
+			return err
+		}
+		view = toView(plan)
+		return nil
+	})
+	return view, err
+}
+
+func (p *Plans) ExpireDue(ctx context.Context) (int, error) {
+	var n int
+	err := p.uow.Within(ctx, func(scope uow.TxScope) error {
+		var err error
+		n, err = p.store.ExpireWaiting(ctx, scope, p.now())
+		return err
+	})
+	return n, err
+}
+
+func (p *Plans) PurgeClosed(ctx context.Context, olderThan time.Time) (int, error) {
+	var n int
+	err := p.uow.Within(ctx, func(scope uow.TxScope) error {
+		var err error
+		n, err = p.store.PurgeClosed(ctx, scope, olderThan)
+		return err
+	})
+	return n, err
+}
+
+func toView(plan *Plan) *PlanView {
+	if plan == nil {
+		return nil
+	}
+	v := &PlanView{
+		PlanID:        plan.PlanID,
+		Revision:      plan.Revision,
+		Status:        plan.Status,
+		ExecutionMode: plan.ExecutionMode,
+		ExpiresAt:     plan.ExpiresAt,
+		SafeSummary:   plan.SafeSummary,
+	}
+	for _, s := range plan.Steps {
+		v.Steps = append(v.Steps, StepView{
+			StepID:        s.StepID,
+			ActionKey:     s.ActionKey,
+			Status:        s.Status,
+			ExecutionMode: s.ExecutionMode,
+			DependsOn:     slices.Clone(s.DependsOn),
+			SafeSummary:   s.SafeSummary,
+		})
+	}
+	return v
+}

internal/app/intentflow/plans_test.go

@@ -0,0 +1,321 @@
+package intentflow_test
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/app/intentflow"
+)
+
+func testClock() (now *time.Time, opts intentflow.Options, owner *intentflow.FakeActionOwner) {
+	t0 := time.Date(2026, 9, 15, 12, 0, 0, 0, time.UTC)
+	now = &t0
+	owner = intentflow.NewFakeActionOwner()
+	opts = intentflow.Options{
+		Owner:   owner,
+		Now:     func() time.Time { return now.UTC() },
+		Lease:   10 * time.Second,
+		PlanTTL: 30 * time.Minute,
+	}
+	return now, opts, owner
+}
+
+func testPlans() (*intentflow.Plans, *time.Time, *intentflow.FakeActionOwner) {
+	now, opts, owner := testClock()
+	return intentflow.NewPlans(&intentflow.SerialUoW{}, intentflow.NewMemoryStore(), opts), now, owner
+}
+
+func principal() intentflow.Principal {
+	return intentflow.Principal{UserID: 7, Environment: "dev", ProjectID: 42, SessionID: "sess-u3"}
+}
+
+func draft(digest string) intentflow.CreateCommand {
+	return intentflow.CreateCommand{
+		Principal:     principal(),
+		ClientTurnID:  "turn-1",
+		RequestDigest: digest,
+		WritePlan:     true,
+		SafeSummary:   "query then generate",
+		Steps: []intentflow.StepDraft{
+			{ActionKey: "regulation.query", ActionDigest: "q1"},
+			{ActionKey: "task.run", ActionDigest: "w1"},
+		},
+	}
+}
+
+func TestHandleTurnLegacyDoesNotPersist(t *testing.T) {
+	p, _, _ := testPlans()
+	ctx := context.Background()
+	res, err := p.HandleTurn(ctx, intentflow.TurnInput{Principal: principal(), Text: "生成报价单", ClientTurnID: "turn-1", RequestDigest: "d1"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res == nil || res.Plan != nil {
+		t.Fatalf("legacy HandleTurn must not create a plan: %+v", res)
+	}
+	active, err := p.GetActivePlan(ctx, principal())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if active != nil {
+		t.Fatal("no plan should be visible")
+	}
+}
+
+func TestCreateIdempotentSameDigest(t *testing.T) {
+	p, _, _ := testPlans()
+	ctx := context.Background()
+	a, err := p.CreatePlan(ctx, draft("digest-a"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	b, err := p.CreatePlan(ctx, draft("digest-a"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if a.PlanID != b.PlanID {
+		t.Fatalf("same turn/digest must reuse plan %s vs %s", a.PlanID, b.PlanID)
+	}
+	other := draft("digest-b")
+	if _, err := p.CreatePlan(ctx, other); !errors.Is(err, intentflow.ErrPlanDecisionConflict) {
+		t.Fatalf("different digest must conflict, got %v", err)
+	}
+}
+
+func TestCreateRejectsSecondActiveWritePlan(t *testing.T) {
+	p, _, _ := testPlans()
+	ctx := context.Background()
+	first, err := p.CreatePlan(ctx, draft("digest-a"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	second := draft("digest-x")
+	second.ClientTurnID = "turn-2"
+	got, err := p.CreatePlan(ctx, second)
+	if !errors.Is(err, intentflow.ErrPlanDecisionConflict) {
+		t.Fatalf("want conflict, got %v", err)
+	}
+	if got == nil || got.PlanID != first.PlanID {
+		t.Fatalf("must return the active write plan, got %+v", got)
+	}
+}
+
+func TestDecisionNonceSameAndDifferentDigest(t *testing.T) {
+	p, _, _ := testPlans()
+	ctx := context.Background()
+	created, err := p.CreatePlan(ctx, draft("digest-a"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	cmd := intentflow.DecisionCommand{
+		Principal:        principal(),
+		PlanID:           created.PlanID,
+		ExpectedRevision: created.Revision,
+		DecisionNonce:    "nonce-1",
+		DecisionDigest:   "dec-1",
+		Kind:             intentflow.DecisionConfirm,
+	}
+	first, err := p.DecidePlan(ctx, cmd)
+	if err != nil {
+		t.Fatal(err)
+	}
+	again, err := p.DecidePlan(ctx, cmd)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if first.Revision != again.Revision {
+		t.Fatal("same nonce/digest must return the same receipt")
+	}
+	cmd.DecisionDigest = "dec-other"
+	if _, err := p.DecidePlan(ctx, cmd); !errors.Is(err, intentflow.ErrPlanDecisionConflict) {
+		t.Fatalf("different digest must conflict, got %v", err)
+	}
+}
+
+func TestConfirmCancelRaceOneWinner(t *testing.T) {
+	p, _, _ := testPlans()
+	ctx := context.Background()
+	created, err := p.CreatePlan(ctx, draft("digest-a"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var okConfirm, okCancel atomic.Int32
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		_, err := p.DecidePlan(ctx, intentflow.DecisionCommand{
+			Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision,
+			DecisionNonce: "n-confirm", DecisionDigest: "d-confirm", Kind: intentflow.DecisionConfirm,
+		})
+		if err == nil {
+			okConfirm.Add(1)
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		_, err := p.DecidePlan(ctx, intentflow.DecisionCommand{
+			Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision,
+			DecisionNonce: "n-cancel", DecisionDigest: "d-cancel", Kind: intentflow.DecisionCancel,
+		})
+		if err == nil {
+			okCancel.Add(1)
+		}
+	}()
+	wg.Wait()
+	if okConfirm.Load()+okCancel.Load() != 1 {
+		t.Fatalf("exactly one CAS winner, confirm=%d cancel=%d", okConfirm.Load(), okCancel.Load())
+	}
+}
+
+func TestTwoWorkersOneOwnerAndLeaseTakeover(t *testing.T) {
+	p, clock, owner := testPlans()
+	ctx := context.Background()
+	created, err := p.CreatePlan(ctx, draft("digest-a"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	confirmed, err := p.DecidePlan(ctx, intentflow.DecisionCommand{
+		Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision,
+		DecisionNonce: "n-c", DecisionDigest: "d-c", Kind: intentflow.DecisionConfirm,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	stepID := confirmed.Steps[0].StepID
+	var gotA, gotB atomic.Int32
+	var first *intentflow.ClaimedStep
+	var mu sync.Mutex
+	var wg sync.WaitGroup
+	wg.Add(2)
+	claim := func(worker string, slot *atomic.Int32) {
+		defer wg.Done()
+		c, err := p.ClaimStep(ctx, worker, created.PlanID, stepID)
+		if err != nil {
+			t.Errorf("claim %s: %v", worker, err)
+			return
+		}
+		if c != nil {
+			slot.Add(1)
+			mu.Lock()
+			if first == nil {
+				first = c
+			}
+			mu.Unlock()
+		}
+	}
+	go claim("worker-a", &gotA)
+	go claim("worker-b", &gotB)
+	wg.Wait()
+	if gotA.Load()+gotB.Load() != 1 {
+		t.Fatalf("two workers must not both own, a=%d b=%d", gotA.Load(), gotB.Load())
+	}
+	stale := first
+	*clock = clock.Add(time.Minute)
+	takeover, err := p.ClaimStep(ctx, "worker-c", created.PlanID, stepID)
+	if err != nil || takeover == nil {
+		t.Fatalf("expired lease must be taken over: %v %+v", err, takeover)
+	}
+	if takeover.FencingToken <= stale.FencingToken {
+		t.Fatalf("fencing must be monotonic: old=%d new=%d", stale.FencingToken, takeover.FencingToken)
+	}
+	if _, err := p.ExecuteClaimed(ctx, stale); !errors.Is(err, intentflow.ErrPlanActionVersionConflict) {
+		t.Fatalf("old fence must be rejected by owner, got %v", err)
+	}
+	if _, err := p.ExecuteClaimed(ctx, takeover); err != nil {
+		t.Fatal(err)
+	}
+	if owner.ExecCount(takeover.IdempotencyKey) != 1 {
+		t.Fatalf("execute count = %d", owner.ExecCount(takeover.IdempotencyKey))
+	}
+}
+
+func TestReceiptWritebackFailureDoesNotRerun(t *testing.T) {
+	p, _, owner := testPlans()
+	ctx := context.Background()
+	created, err := p.CreatePlan(ctx, draft("digest-a"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	confirmed, err := p.DecidePlan(ctx, intentflow.DecisionCommand{
+		Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision,
+		DecisionNonce: "n-c", DecisionDigest: "d-c", Kind: intentflow.DecisionConfirm,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	claimed, err := p.ClaimStep(ctx, "w1", created.PlanID, confirmed.Steps[0].StepID)
+	if err != nil || claimed == nil {
+		t.Fatalf("claim: %v %+v", err, claimed)
+	}
+	if _, err := owner.Execute(ctx, intentflow.StepExec{
+		Principal: 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 _, err := p.RecoverReceipt(ctx, claimed); err != nil {
+		t.Fatal(err)
+	}
+	if owner.ExecCount(claimed.IdempotencyKey) != 1 {
+		t.Fatalf("recover must not re-execute, count=%d", owner.ExecCount(claimed.IdempotencyKey))
+	}
+	got, err := p.GetPlan(ctx, principal(), created.PlanID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.Steps[0].Status != intentflow.StepSucceeded {
+		t.Fatalf("step status %s", got.Steps[0].Status)
+	}
+}
+
+func TestExpireAndEnvironmentIsolationAndRetention(t *testing.T) {
+	p, clock, _ := testPlans()
+	ctx := context.Background()
+	created, err := p.CreatePlan(ctx, draft("digest-a"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	other := principal()
+	other.Environment = "test"
+	if _, err := p.GetPlan(ctx, other, created.PlanID); !errors.Is(err, intentflow.ErrPlanNotFoundOrHidden) {
+		t.Fatalf("cross-env must hide, got %v", err)
+	}
+	*clock = clock.Add(time.Hour)
+	n, err := p.ExpireDue(ctx)
+	if err != nil || n != 1 {
+		t.Fatalf("expire n=%d err=%v", n, err)
+	}
+	if _, err := p.DecidePlan(ctx, intentflow.DecisionCommand{
+		Principal: principal(), PlanID: created.PlanID, ExpectedRevision: 1,
+		DecisionNonce: "late", DecisionDigest: "late", Kind: intentflow.DecisionConfirm,
+	}); !errors.Is(err, intentflow.ErrPlanExpired) && !errors.Is(err, intentflow.ErrPlanStaleRevision) {
+		got, _ := p.GetPlan(ctx, principal(), created.PlanID)
+		if got == nil || got.Status != intentflow.PlanExpired {
+			t.Fatalf("expired plan confirm: %v view=%+v", err, got)
+		}
+	}
+	purged, err := p.PurgeClosed(ctx, clock.Add(time.Second))
+	if err != nil || purged != 1 {
+		t.Fatalf("purge n=%d err=%v", purged, err)
+	}
+	if _, err := p.GetPlan(ctx, principal(), created.PlanID); !errors.Is(err, intentflow.ErrPlanNotFoundOrHidden) {
+		t.Fatalf("purged plan must be hidden, got %v", err)
+	}
+}
+
+func TestUnauthenticatedRejected(t *testing.T) {
+	p, _, _ := testPlans()
+	_, err := p.CreatePlan(context.Background(), intentflow.CreateCommand{
+		Principal: intentflow.Principal{}, ClientTurnID: "t", RequestDigest: "d",
+		Steps: []intentflow.StepDraft{{ActionKey: "regulation.query"}},
+	})
+	if !errors.Is(err, intentflow.ErrUnauthenticated) {
+		t.Fatalf("got %v", err)
+	}
+}

internal/app/intentflow/repository.go

@@ -0,0 +1,486 @@
+package intentflow
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"time"
+
+	"chat-agent-system/internal/app/uow"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+)
+
+// Repository is the PostgreSQL owner of intent_plans*.
+type Repository struct{}
+
+func NewRepository() Repository { return Repository{} }
+
+func txFrom(scope uow.TxScope) (pgx.Tx, error) {
+	return uow.AsPG(scope)
+}
+
+func isUniqueViolation(err error) bool {
+	var pgErr *pgconn.PgError
+	return errors.As(err, &pgErr) && pgErr.Code == "23505"
+}
+
+func uniqueConstraint(err error) string {
+	var pgErr *pgconn.PgError
+	if errors.As(err, &pgErr) {
+		return pgErr.ConstraintName
+	}
+	return ""
+}
+
+func (Repository) InsertPlan(ctx context.Context, scope uow.TxScope, plan *Plan) error {
+	tx, err := txFrom(scope)
+	if err != nil {
+		return err
+	}
+	snap := plan.DisplaySnapshot
+	if len(snap) == 0 {
+		snap = nil
+	}
+	_, err = tx.Exec(ctx, `
+INSERT INTO intent_plans (
+  plan_id, environment, project_id, session_id, user_id, client_turn_id, request_digest,
+  segmenter_version, policy_version, status, revision, expires_at, last_fencing_token,
+  write_plan, execution_mode, safe_summary, display_snapshot, created_at, updated_at
+) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19)`,
+		plan.PlanID, plan.Principal.Environment, plan.Principal.ProjectID, plan.Principal.SessionID,
+		plan.Principal.UserID, plan.ClientTurnID, plan.RequestDigest, plan.SegmenterVersion,
+		plan.PolicyVersion, string(plan.Status), plan.Revision, plan.ExpiresAt, plan.LastFencingToken,
+		plan.WritePlan, string(plan.ExecutionMode), plan.SafeSummary, jsonbOrNil(snap),
+		plan.CreatedAt, plan.UpdatedAt,
+	)
+	if err != nil {
+		if isUniqueViolation(err) {
+			if uniqueConstraint(err) == "uq_intent_plans_one_active_write" {
+				return uow.ErrBusy
+			}
+			return errTurnExists
+		}
+		return err
+	}
+	for _, s := range plan.Steps {
+		if err := insertStep(ctx, tx, s); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func insertStep(ctx context.Context, tx pgx.Tx, s Step) error {
+	slots := s.Slots
+	if len(slots) == 0 {
+		slots = []byte("{}")
+	}
+	_, err := tx.Exec(ctx, `
+INSERT INTO intent_plan_steps (
+  step_id, plan_id, ordinal, action_key, slots, depends_on, status, expected_resource_version,
+  action_digest, idempotency_key, attempt, result_ref, result_digest, safe_summary,
+  lease_owner, lease_until, fencing_token, version, execution_mode, created_at, updated_at
+) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21)`,
+		s.StepID, s.PlanID, s.Ordinal, s.ActionKey, slots, s.DependsOn, string(s.Status),
+		s.ExpectedResourceVersion, s.ActionDigest, s.IdempotencyKey, s.Attempt, s.ResultRef,
+		s.ResultDigest, s.SafeSummary, s.LeaseOwner, s.LeaseUntil, s.FencingToken, s.Version,
+		string(s.ExecutionMode), s.CreatedAt, s.UpdatedAt,
+	)
+	return err
+}
+
+func (r Repository) GetPlan(ctx context.Context, scope uow.TxScope, planID string) (*Plan, error) {
+	return r.loadPlan(ctx, scope, planID, false)
+}
+
+func (r Repository) GetPlanForUpdate(ctx context.Context, scope uow.TxScope, planID string) (*Plan, error) {
+	return r.loadPlan(ctx, scope, planID, true)
+}
+
+func (Repository) loadPlan(ctx context.Context, scope uow.TxScope, planID string, forUpdate bool) (*Plan, error) {
+	tx, err := txFrom(scope)
+	if err != nil {
+		return nil, err
+	}
+	q := planSelect + " WHERE plan_id=$1"
+	if forUpdate {
+		q += " FOR UPDATE"
+	}
+	plan, err := scanPlan(tx.QueryRow(ctx, q, planID))
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	steps, err := loadSteps(ctx, tx, plan.PlanID)
+	if err != nil {
+		return nil, err
+	}
+	plan.Steps = steps
+	return plan, nil
+}
+
+func (Repository) GetByTurn(ctx context.Context, scope uow.TxScope, env, sessionID, clientTurnID string) (*Plan, error) {
+	tx, err := txFrom(scope)
+	if err != nil {
+		return nil, err
+	}
+	plan, err := scanPlan(tx.QueryRow(ctx, planSelect+`
+WHERE environment=$1 AND session_id=$2 AND client_turn_id=$3`, env, sessionID, clientTurnID))
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	steps, err := loadSteps(ctx, tx, plan.PlanID)
+	if err != nil {
+		return nil, err
+	}
+	plan.Steps = steps
+	return plan, nil
+}
+
+func (Repository) GetActiveWrite(ctx context.Context, scope uow.TxScope, p Principal) (*Plan, error) {
+	return loadActive(ctx, scope, p, true)
+}
+
+func (Repository) GetActive(ctx context.Context, scope uow.TxScope, p Principal) (*Plan, error) {
+	return loadActive(ctx, scope, p, false)
+}
+
+func loadActive(ctx context.Context, scope uow.TxScope, p Principal, writeOnly bool) (*Plan, error) {
+	tx, err := txFrom(scope)
+	if err != nil {
+		return nil, err
+	}
+	q := planSelect + `
+WHERE environment=$1 AND project_id=$2 AND session_id=$3 AND user_id=$4
+  AND status = ANY($5)`
+	args := []any{p.Environment, p.ProjectID, p.SessionID, p.UserID, activeStatusStrings()}
+	if writeOnly {
+		q += ` AND write_plan`
+	}
+	q += ` ORDER BY updated_at DESC LIMIT 1`
+	plan, err := scanPlan(tx.QueryRow(ctx, q, args...))
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	steps, err := loadSteps(ctx, tx, plan.PlanID)
+	if err != nil {
+		return nil, err
+	}
+	plan.Steps = steps
+	return plan, nil
+}
+
+func (Repository) CASPlan(ctx context.Context, scope uow.TxScope, planID string, expectedRev int64, from []PlanStatus, to PlanStatus, now time.Time) (int64, error) {
+	tx, err := txFrom(scope)
+	if err != nil {
+		return 0, err
+	}
+	var newRev int64
+	err = tx.QueryRow(ctx, `
+UPDATE intent_plans
+SET status=$1, revision=revision+1, updated_at=$2
+WHERE plan_id=$3 AND revision=$4 AND status = ANY($5)
+RETURNING revision`, string(to), now, planID, expectedRev, statusStrings(from)).Scan(&newRev)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return 0, nil
+	}
+	return newRev, err
+}
+
+func (Repository) InsertDecision(ctx context.Context, scope uow.TxScope, d *Decision) error {
+	tx, err := txFrom(scope)
+	if err != nil {
+		return err
+	}
+	_, err = tx.Exec(ctx, `
+INSERT INTO intent_plan_decisions (
+  decision_id, plan_id, nonce, plan_revision, actor_user_id, kind, step_ids,
+  exclude_unselected, decision_digest, outcome, created_at
+) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)`,
+		d.DecisionID, d.PlanID, d.Nonce, d.PlanRevision, d.ActorUserID, string(d.Kind),
+		d.StepIDs, d.ExcludeUnselected, d.DecisionDigest, d.Outcome, d.CreatedAt,
+	)
+	if isUniqueViolation(err) {
+		return uow.ErrDigestConflict
+	}
+	return err
+}
+
+func (Repository) GetDecisionByNonce(ctx context.Context, scope uow.TxScope, planID, nonce string) (*Decision, error) {
+	tx, err := txFrom(scope)
+	if err != nil {
+		return nil, err
+	}
+	var d Decision
+	var kind string
+	err = tx.QueryRow(ctx, `
+SELECT decision_id, plan_id, nonce, plan_revision, actor_user_id, kind, step_ids,
+       exclude_unselected, decision_digest, outcome, created_at
+FROM intent_plan_decisions WHERE plan_id=$1 AND nonce=$2`, planID, nonce).Scan(
+		&d.DecisionID, &d.PlanID, &d.Nonce, &d.PlanRevision, &d.ActorUserID, &kind,
+		&d.StepIDs, &d.ExcludeUnselected, &d.DecisionDigest, &d.Outcome, &d.CreatedAt,
+	)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	d.Kind = DecisionKind(kind)
+	return &d, nil
+}
+
+func (Repository) ReplaceSteps(ctx context.Context, scope uow.TxScope, planID string, steps []Step) error {
+	tx, err := txFrom(scope)
+	if err != nil {
+		return err
+	}
+	for _, s := range steps {
+		if s.PlanID == "" {
+			s.PlanID = planID
+		}
+		_, err := tx.Exec(ctx, `
+UPDATE intent_plan_steps
+SET status=$1, depends_on=$2, expected_resource_version=$3, action_digest=$4,
+    idempotency_key=$5, attempt=$6, result_ref=$7, result_digest=$8, safe_summary=$9,
+    lease_owner=$10, lease_until=$11, fencing_token=$12, version=$13, execution_mode=$14,
+    updated_at=$15, slots=$16
+WHERE step_id=$17 AND plan_id=$18`,
+			string(s.Status), s.DependsOn, s.ExpectedResourceVersion, s.ActionDigest,
+			s.IdempotencyKey, s.Attempt, s.ResultRef, s.ResultDigest, s.SafeSummary,
+			s.LeaseOwner, s.LeaseUntil, s.FencingToken, s.Version, string(s.ExecutionMode),
+			s.UpdatedAt, jsonbOrNil(s.Slots), s.StepID, planID,
+		)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (Repository) ClaimStep(ctx context.Context, scope uow.TxScope, planID, stepID, worker string, until, now time.Time, idempotencyKey string) (*Step, int64, error) {
+	tx, err := txFrom(scope)
+	if err != nil {
+		return nil, 0, err
+	}
+	var token int64
+	err = tx.QueryRow(ctx, `
+UPDATE intent_plans
+SET last_fencing_token = last_fencing_token + 1, updated_at = $2
+WHERE plan_id = $1
+  AND EXISTS (
+    SELECT 1 FROM intent_plan_steps s
+    WHERE s.plan_id = $1 AND s.step_id = $3
+      AND (
+        s.status = 'eligible'
+        OR (s.status = 'running' AND (s.lease_until IS NULL OR s.lease_until < $2))
+      )
+  )
+RETURNING last_fencing_token`, planID, now, stepID).Scan(&token)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, 0, nil
+	}
+	if err != nil {
+		return nil, 0, err
+	}
+	row := tx.QueryRow(ctx, `
+UPDATE intent_plan_steps
+SET status='running',
+    lease_owner=$1,
+    lease_until=$2,
+    attempt=attempt+1,
+    version=version+1,
+    fencing_token=$3,
+    idempotency_key=CASE WHEN idempotency_key='' THEN $4 ELSE idempotency_key END,
+    updated_at=$5
+WHERE step_id=$6 AND plan_id=$7
+  AND (
+    status='eligible'
+    OR (status='running' AND (lease_until IS NULL OR lease_until < $5))
+  )
+RETURNING `+stepColumns, worker, until, token, idempotencyKey, now, stepID, planID)
+	step, err := scanStep(row)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, 0, nil
+	}
+	return step, token, err
+}
+
+func (Repository) WriteStepReceipt(ctx context.Context, scope uow.TxScope, worker string, token int64, step Step, now time.Time) (bool, error) {
+	tx, err := txFrom(scope)
+	if err != nil {
+		return false, err
+	}
+	tag, err := tx.Exec(ctx, `
+UPDATE intent_plan_steps
+SET status='succeeded', result_ref=$1, result_digest=$2, safe_summary=$3,
+    lease_owner='', lease_until=NULL, version=version+1, updated_at=$4
+WHERE step_id=$5 AND plan_id=$6 AND fencing_token=$7 AND lease_owner=$8 AND status='running'`,
+		step.ResultRef, step.ResultDigest, step.SafeSummary, now, step.StepID, step.PlanID, token, worker)
+	if err != nil {
+		return false, err
+	}
+	if tag.RowsAffected() == 1 {
+		return true, nil
+	}
+	var status, digest string
+	err = tx.QueryRow(ctx, `SELECT status, result_digest FROM intent_plan_steps WHERE step_id=$1 AND plan_id=$2`,
+		step.StepID, step.PlanID).Scan(&status, &digest)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return false, nil
+	}
+	if err != nil {
+		return false, err
+	}
+	return status == string(StepSucceeded) && digest == step.ResultDigest, nil
+}
+
+func (Repository) ExpireWaiting(ctx context.Context, scope uow.TxScope, now time.Time) (int, error) {
+	tx, err := txFrom(scope)
+	if err != nil {
+		return 0, err
+	}
+	rows, err := tx.Query(ctx, `
+UPDATE intent_plans
+SET status='expired', revision=revision+1, updated_at=$1
+WHERE status = ANY($2) AND expires_at <= $1
+RETURNING plan_id`, now, waitingStatusStrings())
+	if err != nil {
+		return 0, err
+	}
+	defer rows.Close()
+	var ids []string
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			return 0, err
+		}
+		ids = append(ids, id)
+	}
+	if err := rows.Err(); err != nil {
+		return 0, err
+	}
+	if len(ids) == 0 {
+		return 0, nil
+	}
+	_, err = tx.Exec(ctx, `
+UPDATE intent_plan_steps
+SET status='cancelled', updated_at=$1
+WHERE plan_id = ANY($2) AND status IN ('pending','eligible')`, now, ids)
+	return len(ids), err
+}
+
+func (Repository) PurgeClosed(ctx context.Context, scope uow.TxScope, before time.Time) (int, error) {
+	tx, err := txFrom(scope)
+	if err != nil {
+		return 0, err
+	}
+	tag, err := tx.Exec(ctx, `
+DELETE FROM intent_plans
+WHERE updated_at < $1
+  AND status IN ('completed','failed','cancelled','expired','superseded')`, before)
+	if err != nil {
+		return 0, err
+	}
+	return int(tag.RowsAffected()), nil
+}
+
+const planSelect = `
+SELECT plan_id, environment, project_id, session_id, user_id, client_turn_id, request_digest,
+       segmenter_version, policy_version, status, revision, expires_at, last_fencing_token,
+       write_plan, execution_mode, safe_summary, display_snapshot, created_at, updated_at
+FROM intent_plans`
+
+const stepColumns = `
+step_id, plan_id, ordinal, action_key, slots, depends_on, status, expected_resource_version,
+action_digest, idempotency_key, attempt, result_ref, result_digest, safe_summary,
+lease_owner, lease_until, fencing_token, version, execution_mode, created_at, updated_at`
+
+func loadSteps(ctx context.Context, tx pgx.Tx, planID string) ([]Step, error) {
+	rows, err := tx.Query(ctx, `SELECT `+stepColumns+` FROM intent_plan_steps WHERE plan_id=$1 ORDER BY ordinal`, planID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []Step
+	for rows.Next() {
+		s, err := scanStep(rows)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, *s)
+	}
+	return out, rows.Err()
+}
+
+type scannable interface {
+	Scan(dest ...any) error
+}
+
+func scanPlan(row scannable) (*Plan, error) {
+	var p Plan
+	var status, mode string
+	var snap []byte
+	err := row.Scan(
+		&p.PlanID, &p.Principal.Environment, &p.Principal.ProjectID, &p.Principal.SessionID,
+		&p.Principal.UserID, &p.ClientTurnID, &p.RequestDigest, &p.SegmenterVersion,
+		&p.PolicyVersion, &status, &p.Revision, &p.ExpiresAt, &p.LastFencingToken,
+		&p.WritePlan, &mode, &p.SafeSummary, &snap, &p.CreatedAt, &p.UpdatedAt,
+	)
+	if err != nil {
+		return nil, err
+	}
+	p.Status = PlanStatus(status)
+	p.ExecutionMode = ExecutionMode(mode)
+	p.DisplaySnapshot = snap
+	return &p, nil
+}
+
+func scanStep(row scannable) (*Step, error) {
+	var s Step
+	var status, mode string
+	err := row.Scan(
+		&s.StepID, &s.PlanID, &s.Ordinal, &s.ActionKey, &s.Slots, &s.DependsOn, &status,
+		&s.ExpectedResourceVersion, &s.ActionDigest, &s.IdempotencyKey, &s.Attempt,
+		&s.ResultRef, &s.ResultDigest, &s.SafeSummary, &s.LeaseOwner, &s.LeaseUntil,
+		&s.FencingToken, &s.Version, &mode, &s.CreatedAt, &s.UpdatedAt,
+	)
+	if err != nil {
+		return nil, err
+	}
+	s.Status = StepStatus(status)
+	s.ExecutionMode = ExecutionMode(mode)
+	return &s, nil
+}
+
+func jsonbOrNil(raw []byte) []byte {
+	if len(raw) == 0 {
+		return []byte("{}")
+	}
+	if !json.Valid(raw) {
+		return []byte("{}")
+	}
+	return raw
+}
+
+func statusStrings(in []PlanStatus) []string {
+	out := make([]string, len(in))
+	for i, s := range in {
+		out[i] = string(s)
+	}
+	return out
+}
+
+func activeStatusStrings() []string { return statusStrings(activeStatuses()) }
+
+func waitingStatusStrings() []string { return statusStrings(waitingStatuses()) }

internal/app/intentflow/store.go

@@ -0,0 +1,55 @@
+package intentflow
+
+import (
+	"context"
+	"time"
+
+	"chat-agent-system/internal/app/uow"
+)
+
+// Store is the intent_plans* owner port. Callers pass the command TxScope.
+type Store interface {
+	InsertPlan(ctx context.Context, scope uow.TxScope, plan *Plan) error
+	GetPlan(ctx context.Context, scope uow.TxScope, planID string) (*Plan, error)
+	GetPlanForUpdate(ctx context.Context, scope uow.TxScope, planID string) (*Plan, error)
+	GetByTurn(ctx context.Context, scope uow.TxScope, env, sessionID, clientTurnID string) (*Plan, error)
+	GetActiveWrite(ctx context.Context, scope uow.TxScope, p Principal) (*Plan, error)
+	GetActive(ctx context.Context, scope uow.TxScope, p Principal) (*Plan, error)
+	CASPlan(ctx context.Context, scope uow.TxScope, planID string, expectedRev int64, from []PlanStatus, to PlanStatus, now time.Time) (newRevision int64, err error)
+	InsertDecision(ctx context.Context, scope uow.TxScope, d *Decision) error
+	GetDecisionByNonce(ctx context.Context, scope uow.TxScope, planID, nonce string) (*Decision, error)
+	ReplaceSteps(ctx context.Context, scope uow.TxScope, planID string, steps []Step) error
+	ClaimStep(ctx context.Context, scope uow.TxScope, planID, stepID, worker string, until, now time.Time, idempotencyKey string) (*Step, int64, error)
+	WriteStepReceipt(ctx context.Context, scope uow.TxScope, worker string, token int64, step Step, now time.Time) (bool, error)
+	ExpireWaiting(ctx context.Context, scope uow.TxScope, now time.Time) (int, error)
+	PurgeClosed(ctx context.Context, scope uow.TxScope, before time.Time) (int, error)
+}
+
+func waitingStatuses() []PlanStatus {
+	return []PlanStatus{PlanProposed, PlanAwaitingClarification, PlanAwaitingConfirmation}
+}
+
+func activeStatuses() []PlanStatus {
+	return []PlanStatus{
+		PlanProposed, PlanAwaitingClarification, PlanAwaitingConfirmation,
+		PlanExecuting, PlanPartiallyCompleted,
+	}
+}
+
+func isActiveStatus(s PlanStatus) bool {
+	for _, x := range activeStatuses() {
+		if x == s {
+			return true
+		}
+	}
+	return false
+}
+
+func isWaitingStatus(s PlanStatus) bool {
+	for _, x := range waitingStatuses() {
+		if x == s {
+			return true
+		}
+	}
+	return false
+}

internal/app/intentflow/types.go

@@ -119,3 +119,123 @@ const (
 	PathPlanByID      = "/api/v1/intent-plans/{plan_id}"
 	PathPlanDecisions = "/api/v1/intent-plans/{plan_id}/decisions"
 )
+
+const SegmenterVersionUnknown = "u3-explicit"
+
+// CreateCommand persists a plan without going through chat. U7 will map HandleTurn
+// onto this command; U3 keeps HandleTurn a no-op so production routing stays legacy.
+type CreateCommand struct {
+	Principal        Principal
+	ClientTurnID     string
+	RequestDigest    string
+	SegmenterVersion string
+	PolicyVersion    string
+	Status           PlanStatus
+	ExecutionMode    ExecutionMode
+	WritePlan        bool
+	ExpiresAt        time.Time
+	SafeSummary      string
+	Steps            []StepDraft
+}
+
+// StepDraft is one proposed step at create time.
+type StepDraft struct {
+	ActionKey               string
+	DependsOn               []string
+	Slots                   []byte
+	ExpectedResourceVersion string
+	ActionDigest            string
+	SafeSummary             string
+	ExecutionMode           ExecutionMode
+}
+
+// Plan is the owner-side aggregate. It is not returned over HTTP.
+type Plan struct {
+	PlanID           string
+	Principal        Principal
+	ClientTurnID     string
+	RequestDigest    string
+	SegmenterVersion string
+	PolicyVersion    string
+	Status           PlanStatus
+	Revision         int64
+	ExpiresAt        time.Time
+	LastFencingToken int64
+	WritePlan        bool
+	ExecutionMode    ExecutionMode
+	SafeSummary      string
+	DisplaySnapshot  []byte
+	CreatedAt        time.Time
+	UpdatedAt        time.Time
+	Steps            []Step
+}
+
+// Step is one row of the plan ledger.
+type Step struct {
+	PlanID                  string
+	StepID                  string
+	Ordinal                 int
+	ActionKey               string
+	Slots                   []byte
+	DependsOn               []string
+	Status                  StepStatus
+	ExpectedResourceVersion string
+	ActionDigest            string
+	IdempotencyKey          string
+	Attempt                 int
+	ResultRef               string
+	ResultDigest            string
+	SafeSummary             string
+	LeaseOwner              string
+	LeaseUntil              *time.Time
+	FencingToken            int64
+	Version                 int64
+	ExecutionMode           ExecutionMode
+	CreatedAt               time.Time
+	UpdatedAt               time.Time
+}
+
+// Decision is an append-only command receipt.
+type Decision struct {
+	DecisionID        string
+	PlanID            string
+	Nonce             string
+	PlanRevision      int64
+	ActorUserID       int64
+	Kind              DecisionKind
+	StepIDs           []string
+	ExcludeUnselected bool
+	DecisionDigest    string
+	Outcome           string
+	CreatedAt         time.Time
+}
+
+// ClaimedStep is a worker lease plus the fence the action owner must see.
+type ClaimedStep struct {
+	Plan           Plan
+	Step           Step
+	WorkerID       string
+	FencingToken   int64
+	IdempotencyKey string
+}
+
+// Receipt is the action-owner result. Intermediate payloads stay behind ResultRef.
+type Receipt struct {
+	IdempotencyKey string
+	ActionDigest   string
+	ResultRef      string
+	ResultDigest   string
+	SafeSummary    string
+}
+
+// StepExec is the port payload for an action owner. Confirmation is not authorization.
+type StepExec struct {
+	Principal               Principal
+	PlanID                  string
+	StepID                  string
+	ActionKey               string
+	IdempotencyKey          string
+	ActionDigest            string
+	ExpectedResourceVersion string
+	FencingToken            int64
+}