feat(intentflow): wire U9–U12 ports and gated OwnerMux

Register dry-run adapters via BuildOwnerMux when independent flags are on;
U9 stub fail-closes on malformed/timeout/out-of-bounds without raising PlanTier.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-16 15:42:49 +00:00
co-authored by Cursor
parent 1cd1cddf64
commit 0611b1f24f
6 changed files with 436 additions and 102 deletions
+6
View File
@@ -94,4 +94,10 @@ INTENT_V2_DECISION=false
INTENT_V2_DECISION_TIER=none
WORKFLOW_INTENTS_ENABLED=true
MULTI_INTENT_ENABLED=true
# MULTI_INTENT_PLAN_TIER defaults to legacy; do not raise for U9–U12 scaffolding.
# MULTI_INTENT_PLAN_TIER=legacy
MULTI_INTENT_U9_SEGMENT=false
MULTI_INTENT_U10_MUTATE=false
MULTI_INTENT_U11_DOCGOV=false
MULTI_INTENT_U12_RUN_CONTROL=false
SEMANTIC_FALLBACK_ENABLED=false
+5 -4
View File
@@ -196,11 +196,12 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) {
livePI := cfg != nil && strings.TrimSpace(cfg.PiAgentRunnerURL) != ""
taskOwner := NewTaskRunAdapter()
taskOwner.Starter = LiveTaskStarter(taskMgr, livePI)
// BuildOwnerMux registers U10–U12 dry-run stubs only when independent flags are on.
rt.IntentPlans = intentflow.NewPlans(rt.Docflow.UoW, intentflow.NewRepository(), intentflow.Options{
Owner: intentflow.OwnerMux{
Query: intentflow.QueryGuard{Inner: intentflow.NewFakeActionOwner()},
Task: taskOwner,
},
Owner: intentflow.BuildOwnerMux(
intentflow.QueryGuard{Inner: intentflow.NewFakeActionOwner()},
taskOwner,
),
Events: events.PGSink{},
})
}
+8
View File
@@ -35,6 +35,14 @@ func ErrorCode(err error) string {
return "PLAN_ACTION_VERSION_CONFLICT"
case errors.Is(err, ErrUnauthenticated):
return "ERR_UNAUTHENTICATED"
case errors.Is(err, ErrSegmentProposalDisabled):
return "SEGMENT_PROPOSAL_DISABLED"
case errors.Is(err, ErrSegmentProposalMalformed):
return "SEGMENT_PROPOSAL_MALFORMED"
case errors.Is(err, ErrSegmentProposalTimeout):
return "SEGMENT_PROPOSAL_TIMEOUT"
case errors.Is(err, ErrSegmentProposalOutOfBounds):
return "SEGMENT_PROPOSAL_OUT_OF_BOUNDS"
default:
return ""
}
+224 -48
View File
@@ -2,84 +2,260 @@ package intentflow
import (
"context"
"errors"
"fmt"
"sync"
"time"
"unicode/utf8"
)
// Expansion adapters for PU8. All gated by independent flags (default OFF).
const SegmentProposalSchemaV1 = "intentflow.segment_proposal.v1"
// SegmentProposal is a fail-closed LLM segment proposal when U9 is on.
type SegmentProposal struct {
SchemaVersion string `json:"schema_version"`
Spans []string `json:"spans"`
Digest string `json:"digest"`
var (
ErrSegmentProposalDisabled = errors.New("SEGMENT_PROPOSAL_DISABLED")
ErrSegmentProposalMalformed = errors.New("SEGMENT_PROPOSAL_MALFORMED")
ErrSegmentProposalTimeout = errors.New("SEGMENT_PROPOSAL_TIMEOUT")
ErrSegmentProposalOutOfBounds = errors.New("SEGMENT_PROPOSAL_OUT_OF_BOUNDS")
)
// SegmentSpan is a half-open byte range into the original utterance.
type SegmentSpan struct {
Start int `json:"start"`
End int `json:"end"`
Text string `json:"text,omitempty"`
}
// ProposeSegments returns ErrExpansionDisabled when flag off; when on, rejects
// empty/malformed input fail-closed (no silent legacy bypass).
func ProposeSegments(ctx context.Context, text string) (*SegmentProposal, error) {
_ = ctx
// SegmentProposalRequest is the minimal U9 input.
type SegmentProposalRequest struct {
Text string
RequestDigest string
MaxSegments int
SchemaVersion string
}
// SegmentProposal is a versioned, digest-bound suggestion.
type SegmentProposal struct {
SchemaVersion string `json:"schema_version"`
Digest string `json:"digest"`
Spans []SegmentSpan `json:"spans"`
}
// SegmentProposer proposes spans only when rules cannot decide (U9).
type SegmentProposer interface {
Propose(ctx context.Context, req SegmentProposalRequest) (*SegmentProposal, error)
}
type disabledSegmentProposer struct{}
func (disabledSegmentProposer) Propose(context.Context, SegmentProposalRequest) (*SegmentProposal, error) {
return nil, ErrSegmentProposalDisabled
}
// ActiveSegmentProposer returns disabled when MULTI_INTENT_U9_SEGMENT is off.
func ActiveSegmentProposer() SegmentProposer {
if !CurrentFeatureFlags().U9Segment {
return nil, ErrExpansionDisabled
return disabledSegmentProposer{}
}
if len(text) == 0 {
return nil, fmt.Errorf("u9 segment: empty input (fail closed)")
return NewStubSegmentProposer(0)
}
// StubSegmentProposer is scaffolding (no LLM): schema/span validation + timeout fail-closed.
type StubSegmentProposer struct {
ArtificialDelay time.Duration
}
func NewStubSegmentProposer(delay time.Duration) *StubSegmentProposer {
return &StubSegmentProposer{ArtificialDelay: delay}
}
func (s *StubSegmentProposer) Propose(ctx context.Context, req SegmentProposalRequest) (*SegmentProposal, error) {
if err := ctx.Err(); err != nil {
return nil, ErrSegmentProposalTimeout
}
if s != nil && s.ArtificialDelay > 0 {
timer := time.NewTimer(s.ArtificialDelay)
defer timer.Stop()
select {
case <-ctx.Done():
return nil, fmt.Errorf("u9 segment: timeout/cancel (fail closed): %w", ctx.Err())
default:
return nil, ErrSegmentProposalTimeout
case <-timer.C:
}
}
if err := ctx.Err(); err != nil {
return nil, ErrSegmentProposalTimeout
}
schema := req.SchemaVersion
if schema == "" {
schema = SegmentProposalSchemaV1
}
if schema != SegmentProposalSchemaV1 {
return nil, ErrSegmentProposalMalformed
}
text := req.Text
if text == "" || !utf8.ValidString(text) {
return nil, ErrSegmentProposalMalformed
}
maxSeg := req.MaxSegments
if maxSeg <= 0 {
maxSeg = 4
}
if maxSeg > 4 || req.RequestDigest == "" {
return nil, ErrSegmentProposalMalformed
}
span := SegmentSpan{Start: 0, End: len(text), Text: text}
if err := ValidateProposedSpans(text, []SegmentSpan{span}); err != nil {
return nil, err
}
// Deterministic stub proposal (no LLM) for gate wiring; real model behind same schema.
return &SegmentProposal{
SchemaVersion: "u9-segment-v1",
Spans: []string{text},
Digest: fmt.Sprintf("sha-stub-%d", len(text)),
SchemaVersion: schema,
Digest: fmt.Sprintf("stub:%s", req.RequestDigest),
Spans: []SegmentSpan{span},
}, nil
}
// MutateDryRun validates mutate adapters when U10 on; never applies side effects here.
func MutateDryRun(ctx context.Context, action string, expectedRev int64) error {
_ = ctx
if !CurrentFeatureFlags().U10Mutate {
return ErrExpansionDisabled
// ValidateProposedSpans rejects out-of-bounds or overlapping spans.
func ValidateProposedSpans(text string, spans []SegmentSpan) error {
if len(spans) == 0 {
return ErrSegmentProposalMalformed
}
if action == "" || expectedRev <= 0 {
return fmt.Errorf("u10 mutate: missing action or expected revision (fail closed)")
n := len(text)
prevEnd := 0
for _, sp := range spans {
if sp.Start < 0 || sp.End < sp.Start || sp.End > n {
return ErrSegmentProposalOutOfBounds
}
if sp.Start < prevEnd {
return ErrSegmentProposalOutOfBounds
}
if sp.Text != "" && sp.Text != text[sp.Start:sp.End] {
return ErrSegmentProposalMalformed
}
prevEnd = sp.End
}
return nil
}
// DocGovDryRun validates docgov action when U11 on.
func DocGovDryRun(ctx context.Context, action string, expectedDocVersion int64) error {
_ = ctx
if !CurrentFeatureFlags().U11DocGov {
return ErrExpansionDisabled
var mutateActionKeys = map[string]struct{}{
"project.fact.supplement": {},
}
var docGovActionKeys = map[string]struct{}{
"document.claim": {},
"document.review.submit": {},
"document.review.approve": {},
"document.review.reject": {},
"document.final.approve": {},
"document.final.reject": {},
"document.proposal.decide": {},
}
var runControlActionKeys = map[string]struct{}{
"workflow.switch": {},
"workflow.run": {},
"run.resume": {},
"run.restart": {},
}
// DryRunAdapter is a fail-closed / dry-run ActionOwner for U10–U12 scaffolding.
type DryRunAdapter struct {
mu sync.Mutex
allowed map[string]struct{}
receipts map[string]Receipt
Denied bool
Label string
RequireRev bool
}
func NewMutateDryRunAdapter() *DryRunAdapter {
return newDryRunAdapter("mutate", mutateActionKeys, true)
}
func NewDocGovDryRunAdapter() *DryRunAdapter {
return newDryRunAdapter("docgov", docGovActionKeys, true)
}
func NewRunControlDryRunAdapter() *DryRunAdapter {
return newDryRunAdapter("run_control", runControlActionKeys, false)
}
func newDryRunAdapter(label string, keys map[string]struct{}, requireRev bool) *DryRunAdapter {
cp := make(map[string]struct{}, len(keys))
for k := range keys {
cp[k] = struct{}{}
}
if action == "" || expectedDocVersion <= 0 {
return fmt.Errorf("u11 docgov: missing action or document version (fail closed)")
return &DryRunAdapter{
allowed: cp,
receipts: map[string]Receipt{},
Label: label,
RequireRev: requireRev,
}
}
func (a *DryRunAdapter) allows(key string) bool {
if a == nil {
return false
}
_, ok := a.allowed[key]
return ok
}
func (a *DryRunAdapter) Authorize(_ context.Context, exec StepExec) error {
if a == nil || a.Denied || !a.allows(exec.ActionKey) {
return ErrPlanActionForbidden
}
if a.RequireRev && exec.ExpectedResourceVersion == "" {
return ErrPlanActionVersionConflict
}
return nil
}
// RunControlDryRun validates workflow/run-control when U12 on.
func RunControlDryRun(ctx context.Context, kind string) error {
_ = ctx
if !CurrentFeatureFlags().U12RunControl {
return ErrExpansionDisabled
func (a *DryRunAdapter) Execute(ctx context.Context, exec StepExec) (Receipt, error) {
if err := a.Authorize(ctx, exec); err != nil {
return Receipt{}, err
}
switch kind {
case "workflow_switch", "run_resume", "run_restart", "run_cancel":
return nil
default:
return fmt.Errorf("u12 run-control: unknown kind %q (fail closed)", kind)
a.mu.Lock()
defer a.mu.Unlock()
if existing, ok := a.receipts[exec.IdempotencyKey]; ok {
if existing.ActionDigest != exec.ActionDigest {
return Receipt{}, ErrPlanDecisionConflict
}
return existing, nil
}
rec := Receipt{
IdempotencyKey: exec.IdempotencyKey,
ActionDigest: exec.ActionDigest,
ResultRef: "dryrun:" + a.Label + ":" + exec.StepID,
ResultDigest: "validated_only:" + exec.ActionDigest,
SafeSummary: a.Label + " dry-run accepted",
}
a.receipts[exec.IdempotencyKey] = rec
return rec, nil
}
// ErrExpansionDisabled means the independent PU8 flag is off (legacy path).
var ErrExpansionDisabled = fmt.Errorf("intentflow expansion disabled")
func (a *DryRunAdapter) LookupReceipt(_ context.Context, idempotencyKey string) (Receipt, bool, error) {
if a == nil {
return Receipt{}, false, nil
}
a.mu.Lock()
defer a.mu.Unlock()
rec, ok := a.receipts[idempotencyKey]
return rec, ok, nil
}
// WithTimeout is a small helper for U9 fail-closed timeouts in tests.
func WithTimeout(parent context.Context, d time.Duration) (context.Context, context.CancelFunc) {
return context.WithTimeout(parent, d)
// BuildOwnerMux registers U10–U12 dry-run owners only when their flags are on.
func BuildOwnerMux(query, task ActionOwner) OwnerMux {
m := OwnerMux{Query: query, Task: task}
f := CurrentFeatureFlags()
if f.U10Mutate {
m.Mutate = NewMutateDryRunAdapter()
}
if f.U11DocGov {
m.DocGov = NewDocGovDryRunAdapter()
}
if f.U12RunControl {
m.RunControl = NewRunControlDryRunAdapter()
}
return m
}
+155 -34
View File
@@ -2,11 +2,11 @@ package intentflow
import (
"context"
"errors"
"testing"
"time"
)
func TestPU8FlagsDefaultOff(t *testing.T) {
func TestExpansionFlagsDefaultOff(t *testing.T) {
t.Setenv("MULTI_INTENT_U9_SEGMENT", "")
t.Setenv("MULTI_INTENT_U10_MUTATE", "")
t.Setenv("MULTI_INTENT_U11_DOCGOV", "")
@@ -14,50 +14,171 @@ func TestPU8FlagsDefaultOff(t *testing.T) {
t.Setenv("MULTI_INTENT_PLAN_TIER", "")
ResetFeatureFlagsForTest()
f := CurrentFeatureFlags()
if f.PlanTier != TierLegacy {
t.Fatalf("PlanTier = %s, want legacy", f.PlanTier)
}
if f.U9Segment || f.U10Mutate || f.U11DocGov || f.U12RunControl {
t.Fatalf("PU8 flags must default off: %+v", f)
t.Fatalf("expansion flags must default off: %+v", f)
}
}
func TestExpansionFlagsEnvOn(t *testing.T) {
t.Setenv("MULTI_INTENT_U9_SEGMENT", "true")
t.Setenv("MULTI_INTENT_U10_MUTATE", "1")
t.Setenv("MULTI_INTENT_U11_DOCGOV", "yes")
t.Setenv("MULTI_INTENT_U12_RUN_CONTROL", "TRUE")
t.Setenv("MULTI_INTENT_PLAN_TIER", "legacy")
ResetFeatureFlagsForTest()
f := CurrentFeatureFlags()
if !f.U9Segment || !f.U10Mutate || !f.U11DocGov || !f.U12RunControl {
t.Fatalf("expected all expansion flags on: %+v", f)
}
if f.PlanTier != TierLegacy {
t.Fatalf("tier=%s want legacy", f.PlanTier)
t.Fatal("enabling U9–U12 must not raise MULTI_INTENT_PLAN_TIER")
}
}
func TestPU8ExpansionDisabledWhenOff(t *testing.T) {
func TestActiveSegmentProposerDisabledByDefault(t *testing.T) {
t.Setenv("MULTI_INTENT_U9_SEGMENT", "")
ResetFeatureFlagsForTest()
SetFeatureFlagsForTest(FeatureFlags{PlanTier: TierLegacy})
if _, err := ProposeSegments(context.Background(), "x"); !errors.Is(err, ErrExpansionDisabled) {
t.Fatalf("u9: %v", err)
}
if err := MutateDryRun(context.Background(), "edit", 1); !errors.Is(err, ErrExpansionDisabled) {
t.Fatalf("u10: %v", err)
}
if err := DocGovDryRun(context.Background(), "claim", 1); !errors.Is(err, ErrExpansionDisabled) {
t.Fatalf("u11: %v", err)
}
if err := RunControlDryRun(context.Background(), "run_cancel"); !errors.Is(err, ErrExpansionDisabled) {
t.Fatalf("u12: %v", err)
_, err := ActiveSegmentProposer().Propose(context.Background(), SegmentProposalRequest{
Text: "查法规然后生成报告", RequestDigest: "d1",
})
if err != ErrSegmentProposalDisabled {
t.Fatalf("want DISABLED, got %v", err)
}
}
func TestPU8ExpansionEnablePaths(t *testing.T) {
SetFeatureFlagsForTest(FeatureFlags{
PlanTier: TierLegacy, U9Segment: true, U10Mutate: true, U11DocGov: true, U12RunControl: true,
func TestStubSegmentProposerFailClosed(t *testing.T) {
t.Setenv("MULTI_INTENT_U9_SEGMENT", "true")
ResetFeatureFlagsForTest()
p := ActiveSegmentProposer()
if _, err := p.Propose(context.Background(), SegmentProposalRequest{}); err != ErrSegmentProposalMalformed {
t.Fatalf("empty request: %v", err)
}
if _, err := p.Propose(context.Background(), SegmentProposalRequest{
Text: "ok", RequestDigest: "d", SchemaVersion: "wrong.v0",
}); err != ErrSegmentProposalMalformed {
t.Fatalf("bad schema: %v", err)
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Millisecond)
defer cancel()
slow := NewStubSegmentProposer(50 * time.Millisecond)
if _, err := slow.Propose(ctx, SegmentProposalRequest{Text: "ok", RequestDigest: "d"}); err != ErrSegmentProposalTimeout {
t.Fatalf("timeout: %v", err)
}
if err := ValidateProposedSpans("ab", []SegmentSpan{{Start: 0, End: 5}}); err != ErrSegmentProposalOutOfBounds {
t.Fatalf("oob: %v", err)
}
got, err := p.Propose(context.Background(), SegmentProposalRequest{
Text: "查法规", RequestDigest: "digest-1",
})
defer ResetFeatureFlagsForTest()
p, err := ProposeSegments(context.Background(), "hello")
if err != nil || p == nil || p.SchemaVersion != "u9-segment-v1" {
t.Fatalf("u9 on: %v %+v", err, p)
}
if _, err := ProposeSegments(context.Background(), ""); err == nil {
t.Fatal("u9 empty must fail closed")
}
if err := MutateDryRun(context.Background(), "edit", 2); err != nil {
if err != nil {
t.Fatal(err)
}
if err := DocGovDryRun(context.Background(), "submit", 3); err != nil {
t.Fatal(err)
}
if err := RunControlDryRun(context.Background(), "run_resume"); err != nil {
t.Fatal(err)
if got.SchemaVersion != SegmentProposalSchemaV1 || len(got.Spans) != 1 {
t.Fatalf("proposal = %+v", got)
}
}
func TestBuildOwnerMuxFlagOffNoExpansion(t *testing.T) {
t.Setenv("MULTI_INTENT_U10_MUTATE", "")
t.Setenv("MULTI_INTENT_U11_DOCGOV", "")
t.Setenv("MULTI_INTENT_U12_RUN_CONTROL", "")
ResetFeatureFlagsForTest()
mux := BuildOwnerMux(NewFakeActionOwner(), NewFakeActionOwner())
if mux.Mutate != nil || mux.DocGov != nil || mux.RunControl != nil {
t.Fatal("flag-off must not register expansion adapters")
}
q := &routeCountOwner{}
mux = BuildOwnerMux(q, NewFakeActionOwner())
_, err := mux.Execute(context.Background(), StepExec{
ActionKey: "project.fact.supplement", IdempotencyKey: "k", ActionDigest: "d",
ExpectedResourceVersion: "1",
})
if err != nil {
t.Fatal(err)
}
if q.n != 1 {
t.Fatalf("flag-off mutate must hit Query, got n=%d", q.n)
}
}
func TestBuildOwnerMuxEnablePathsNoPanic(t *testing.T) {
t.Setenv("MULTI_INTENT_U10_MUTATE", "true")
t.Setenv("MULTI_INTENT_U11_DOCGOV", "true")
t.Setenv("MULTI_INTENT_U12_RUN_CONTROL", "true")
ResetFeatureFlagsForTest()
mux := BuildOwnerMux(NewFakeActionOwner(), NewFakeActionOwner())
if mux.Mutate == nil || mux.DocGov == nil || mux.RunControl == nil {
t.Fatal("flag-on must register dry-run adapters")
}
ctx := context.Background()
cases := []struct {
key string
rev string
}{
{"project.fact.supplement", "rev-1"},
{"document.claim", "doc-v1"},
{"workflow.run", ""},
{"run.resume", ""},
}
for i, tc := range cases {
_, err := mux.Execute(ctx, StepExec{
ActionKey: tc.key, StepID: "s", IdempotencyKey: "id-" + tc.key,
ActionDigest: "d", ExpectedResourceVersion: tc.rev,
})
if err != nil {
t.Fatalf("case %d %s: %v", i, tc.key, err)
}
}
_, err := mux.Execute(ctx, StepExec{
ActionKey: "project.fact.supplement", IdempotencyKey: "no-rev", ActionDigest: "d",
})
if err != ErrPlanActionVersionConflict {
t.Fatalf("mutate without revision: %v", err)
}
bare := OwnerMux{Query: NewFakeActionOwner(), Task: NewFakeActionOwner()}
SetFeatureFlagsForTest(FeatureFlags{PlanTier: TierLegacy, U10Mutate: true})
defer ResetFeatureFlagsForTest()
_, err = bare.Execute(ctx, StepExec{
ActionKey: "project.fact.supplement", IdempotencyKey: "nil", ActionDigest: "d",
ExpectedResourceVersion: "1",
})
if err != ErrPlanActionForbidden {
t.Fatalf("nil Mutate must fail closed, got %v", err)
}
}
func TestDryRunAdapterRejectsForeignKeys(t *testing.T) {
a := NewMutateDryRunAdapter()
_, err := a.Execute(context.Background(), StepExec{
ActionKey: "document.claim", IdempotencyKey: "x", ActionDigest: "d",
ExpectedResourceVersion: "1",
})
if err != ErrPlanActionForbidden {
t.Fatalf("mutate adapter must not accept docgov key: %v", err)
}
}
type routeCountOwner struct {
n int
}
func (r *routeCountOwner) Authorize(context.Context, StepExec) error { return nil }
func (r *routeCountOwner) Execute(_ context.Context, exec StepExec) (Receipt, error) {
r.n++
return Receipt{IdempotencyKey: exec.IdempotencyKey, ActionDigest: exec.ActionDigest}, nil
}
func (r *routeCountOwner) LookupReceipt(context.Context, string) (Receipt, bool, error) {
return Receipt{}, false, nil
}
+32 -10
View File
@@ -2,19 +2,41 @@ package intentflow
import "context"
// OwnerMux routes generate keys to Task and everything else to Query.
// OwnerMux routes generate keys to Task, optional U10–U12 owners when their
// flags are on, and everything else to Query.
type OwnerMux struct {
Query ActionOwner
Task ActionOwner
Mutate ActionOwner
DocGov ActionOwner
RunControl ActionOwner
}
func (m OwnerMux) pick(actionKey string) ActionOwner {
switch actionKey {
case "task.run", "task.run.adhoc":
return m.Task
default:
}
f := CurrentFeatureFlags()
if _, ok := mutateActionKeys[actionKey]; ok {
if f.U10Mutate {
return m.Mutate
}
return m.Query
}
if _, ok := docGovActionKeys[actionKey]; ok {
if f.U11DocGov {
return m.DocGov
}
return m.Query
}
if _, ok := runControlActionKeys[actionKey]; ok {
if f.U12RunControl {
return m.RunControl
}
return m.Query
}
return m.Query
}
func (m OwnerMux) Authorize(ctx context.Context, exec StepExec) error {
@@ -40,22 +62,22 @@ func (m OwnerMux) Execute(ctx context.Context, exec StepExec) (Receipt, error) {
}
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 {
owners := []ActionOwner{m.Task, m.Mutate, m.DocGov, m.RunControl, m.Query}
for _, o := range owners {
if o == nil {
continue
}
if rec, ok, err := o.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 {
for _, o := range []ActionOwner{m.Query, m.Task, m.Mutate, m.DocGov, m.RunControl} {
if obs, ok := o.(FenceObserver); ok {
obs.ObserveFence(stepID, token)
}
if obs, ok := m.Task.(FenceObserver); ok {
obs.ObserveFence(stepID, token)
}
}