fix(intentflow): bind resource versions for U10/U11 at plan create

Mutate/docgov live Authorize requires ExpectedResourceVersion; populate it
(and frame slots) when HandleTurn drafts confirm-class steps so confirm
no longer fails closed with PLAN_ACTION_VERSION_CONFLICT.
This commit is contained in:
weichen
2026-09-23 15:44:48 +00:00
parent 2f7e4b8fc0
commit dbae19e80d
5 changed files with 132 additions and 23 deletions
+2 -1
View File
@@ -306,7 +306,8 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) {
taskOwner, taskOwner,
liveOwners, liveOwners,
), ),
Events: events.PGSink{}, Events: events.PGSink{},
ResourceVersionBinder: newResourceVersionBinder(projectBlackboardPort(pool, parts.Ctx), parts.Gov),
}) })
} }
svc.SetAgentMaterializer(rt.Docflow) svc.SetAgentMaterializer(rt.Docflow)
+3 -10
View File
@@ -2,10 +2,8 @@ package app
import ( import (
"context" "context"
"encoding/json"
"fmt" "fmt"
"strconv" "strconv"
"strings"
"sync" "sync"
"chat-agent-system/internal/app/intentflow" "chat-agent-system/internal/app/intentflow"
@@ -148,15 +146,10 @@ func (d *DocGovOwner) LookupReceipt(ctx context.Context, idempotencyKey string)
} }
func docIDFromExec(exec intentflow.StepExec) string { func docIDFromExec(exec intentflow.StepExec) string {
if len(exec.Slots) > 0 { if id := docIDFromSlotsOrSummary(exec.Slots, exec.SafeSummary); id != "" {
var slots map[string]string return id
if json.Unmarshal(exec.Slots, &slots) == nil {
if id := strings.TrimSpace(slots["doc_id"]); id != "" {
return id
}
}
} }
return strings.TrimSpace(exec.SafeSummary) return ""
} }
func docVersion(doc *docgov.DocumentRecord) string { func docVersion(doc *docgov.DocumentRecord) string {
+26 -8
View File
@@ -24,20 +24,35 @@ type Plans struct {
owner ActionOwner owner ActionOwner
authz StepAuthorizer authz StepAuthorizer
events EventSink events EventSink
bind ResourceVersionBinder
now func() time.Time now func() time.Time
newID func(prefix string) string newID func(prefix string) string
lease time.Duration lease time.Duration
ttl time.Duration ttl time.Duration
} }
// ResourceVersionBindInput carries enough context to resolve optimistic versions
// for mutate/docgov steps at plan-create time (U10/U11).
type ResourceVersionBindInput struct {
Principal Principal
ActionKey string
SafeSummary string
Slots []byte
}
// ResourceVersionBinder returns the expected resource version for a draft step.
// Empty string means the action key does not require a version binding.
type ResourceVersionBinder func(ctx context.Context, in ResourceVersionBindInput) (string, error)
type Options struct { type Options struct {
Owner ActionOwner Owner ActionOwner
Authorizer StepAuthorizer Authorizer StepAuthorizer
Events EventSink Events EventSink
Now func() time.Time ResourceVersionBinder ResourceVersionBinder
NewID func(prefix string) string Now func() time.Time
Lease time.Duration NewID func(prefix string) string
PlanTTL time.Duration Lease time.Duration
PlanTTL time.Duration
} }
func NewPlans(unit uow.UnitOfWork, store Store, opt Options) *Plans { func NewPlans(unit uow.UnitOfWork, store Store, opt Options) *Plans {
@@ -63,7 +78,10 @@ func NewPlans(unit uow.UnitOfWork, store Store, opt Options) *Plans {
authz = a authz = a
} }
} }
return &Plans{uow: unit, store: store, owner: opt.Owner, authz: authz, events: opt.Events, now: now, newID: newID, lease: lease, ttl: ttl} return &Plans{
uow: unit, store: store, owner: opt.Owner, authz: authz, events: opt.Events,
bind: opt.ResourceVersionBinder, now: now, newID: newID, lease: lease, ttl: ttl,
}
} }
func (p *Plans) stepAuthorizer() StepAuthorizer { func (p *Plans) stepAuthorizer() StepAuthorizer {
+42 -4
View File
@@ -4,6 +4,7 @@ import (
"context" "context"
"crypto/sha256" "crypto/sha256"
"encoding/hex" "encoding/hex"
"encoding/json"
"fmt" "fmt"
"strings" "strings"
@@ -118,11 +119,27 @@ func (p *Plans) handleTurnElevated(ctx context.Context, in TurnInput) (*TurnResu
if tier == TierShadow { if tier == TierShadow {
mode = ModeValidatedOnly mode = ModeValidatedOnly
} }
slots := slotsJSONFromSegment(seg)
expectedRev := ""
if p != nil && p.bind != nil && mode == ModeLive {
bound, bindErr := p.bind(ctx, ResourceVersionBindInput{
Principal: in.Principal,
ActionKey: key,
SafeSummary: summary,
Slots: slots,
})
if bindErr != nil {
return &TurnResult{}, nil
}
expectedRev = bound
}
drafts = append(drafts, StepDraft{ drafts = append(drafts, StepDraft{
ActionKey: key, ActionKey: key,
ActionDigest: fmt.Sprintf("%s:%d:%s", proposal.Digest, i, key), ActionDigest: fmt.Sprintf("%s:%d:%s", proposal.Digest, i, key),
SafeSummary: summary, SafeSummary: summary,
ExecutionMode: mode, ExecutionMode: mode,
Slots: slots,
ExpectedResourceVersion: expectedRev,
}) })
} }
turnID := in.ClientTurnID turnID := in.ClientTurnID
@@ -202,6 +219,27 @@ func counterfactualRoute(proposal *intent.PlanProposal) string {
} }
} }
func slotsJSONFromSegment(seg intent.ProposedSegment) []byte {
if seg.Frame == nil || len(seg.Frame.Slots) == 0 {
return nil
}
out := make(map[string]string, len(seg.Frame.Slots))
for k, sv := range seg.Frame.Slots {
v := strings.TrimSpace(sv.Value)
if v != "" {
out[k] = v
}
}
if len(out) == 0 {
return nil
}
b, err := json.Marshal(out)
if err != nil {
return nil
}
return b
}
func shortDigest(text string) string { func shortDigest(text string) string {
sum := sha256.Sum256([]byte(text)) sum := sha256.Sum256([]byte(text))
return hex.EncodeToString(sum[:8]) return hex.EncodeToString(sum[:8])
+59
View File
@@ -0,0 +1,59 @@
package app
import (
"context"
"encoding/json"
"strings"
"chat-agent-system/internal/app/intentflow"
"chat-agent-system/internal/docgov"
)
// newResourceVersionBinder binds optimistic versions for U10/U11 live keys at
// plan-create time so confirm-time Authorize sees ExpectedResourceVersion.
func newResourceVersionBinder(bb ProjectBlackboardPort, gov *docgov.GovernanceEngine) intentflow.ResourceVersionBinder {
return func(ctx context.Context, in intentflow.ResourceVersionBindInput) (string, error) {
switch in.ActionKey {
case mutateActionKey:
if bb.CurrentVersion == nil {
return "0", nil
}
return bb.CurrentVersion(ctx, in.Principal.ProjectID)
case "document.claim", "document.review.submit":
if gov == nil {
return "", intentflow.ErrPlanActionForbidden
}
docID := docIDFromSlotsOrSummary(in.Slots, in.SafeSummary)
if docID == "" {
return "", intentflow.ErrPlanActionForbidden
}
doc, err := gov.GetDocument(ctx, docID)
if err != nil || doc == nil {
return "", intentflow.ErrPlanActionForbidden
}
return docVersion(doc), nil
default:
return "", nil
}
}
}
func docIDFromSlotsOrSummary(slots []byte, summary string) string {
if len(slots) > 0 {
var m map[string]string
if json.Unmarshal(slots, &m) == nil {
if id := strings.TrimSpace(m["doc_id"]); id != "" {
return id
}
}
}
summary = strings.TrimSpace(summary)
// Prefer explicit doc_proj* tokens embedded in the utterance summary.
for _, tok := range strings.Fields(summary) {
tok = strings.Trim(tok, ",。,.;;::")
if strings.HasPrefix(tok, "doc_") {
return tok
}
}
return summary
}