diff --git a/.env.example b/.env.example index d2d9a40d..ce2982c6 100644 --- a/.env.example +++ b/.env.example @@ -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 diff --git a/internal/app/app.go b/internal/app/app.go index 015827d8..15c84231 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -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{}, }) } diff --git a/internal/app/intentflow/errors.go b/internal/app/intentflow/errors.go index e33ebf7b..3737e226 100644 --- a/internal/app/intentflow/errors.go +++ b/internal/app/intentflow/errors.go @@ -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 "" } diff --git a/internal/app/intentflow/expansion.go b/internal/app/intentflow/expansion.go index 8a7cd9ee..b9d27f53 100644 --- a/internal/app/intentflow/expansion.go +++ b/internal/app/intentflow/expansion.go @@ -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 } - select { - case <-ctx.Done(): - return nil, fmt.Errorf("u9 segment: timeout/cancel (fail closed): %w", ctx.Err()) - default: + if s != nil && s.ArtificialDelay > 0 { + timer := time.NewTimer(s.ArtificialDelay) + defer timer.Stop() + select { + case <-ctx.Done(): + 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 } diff --git a/internal/app/intentflow/expansion_test.go b/internal/app/intentflow/expansion_test.go index ddad29f7..d2753e19 100644 --- a/internal/app/intentflow/expansion_test.go +++ b/internal/app/intentflow/expansion_test.go @@ -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 +} diff --git a/internal/app/intentflow/mux.go b/internal/app/intentflow/mux.go index 584c6572..589db2d2 100644 --- a/internal/app/intentflow/mux.go +++ b/internal/app/intentflow/mux.go @@ -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 + 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 { - obs.ObserveFence(stepID, token) - } - if obs, ok := m.Task.(FenceObserver); ok { - obs.ObserveFence(stepID, token) + for _, o := range []ActionOwner{m.Query, m.Task, m.Mutate, m.DocGov, m.RunControl} { + if obs, ok := o.(FenceObserver); ok { + obs.ObserveFence(stepID, token) + } } }