Plan 3 — Workflow Engine Implementation Plan
For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (
- [ ]) syntax for tracking.
Goal: A per-organization, per-kind, versioned workflow engine that any module can pin a document to: publish-validated definitions (states, edges, gates, grants with amount floors), grantee resolution against the org structure including superior, delegation and escalation, a transition function that posts through posting.Dispatch inside its own transaction and writes an immutable decision log naming the resolved approver set, and the korre expense-claim flow seeded as v1 for every organization.
Architecture: internal/platform/workflow owns everything with a rule in it: the graph types, a pure publish-time validator (§5.7, all seven rules), the definition store (mutable drafts, insert-only published versions), grant resolution over orgstruct, the threshold ladder, delegation, lazy escalation, and the instance engine. internal/modules/workflow is a thin HTTP shell under /v1/workflow/… for the definition editor (Plan 10), delegations and the decision log. No module is imported by the engine; the ledger is reached only through internal/platform/posting.
Tech Stack: Go 1.26 · Gin · pgx v5 (raw SQL, the house pattern in internal/platform/*) · goose · testcontainers-go · testify · oapi-codegen (spec only; handlers are hand-written Gin, as in modules/orgstructure)
Specs (read before any task):
docs/superpowers/specs/2026-09-21-phase1-wave1-contracts.md— BINDING. Migration range, step orders, package paths, permission prefix, path prefix, shared-file rule, the posting seam.docs/superpowers/specs/2026-09-20-erp-architecture-design.md— §5 (whole), §6.3, §4.1–4.3, §11A, §12.2, §17.4, §20.1.1 item 5, §20.2.docs/superpowers/specs/2026-09-21-org-structure-and-onboarding-design.md— §5 (resolution chain,SuperiorOf,NextFilledAbove, §5.5 “as of a date”).
Global Constraints
Section titled “Global Constraints”Every task’s requirements include this section.
- Go floor is 1.26. CI derives it from
go.mod; never restate a version. - Ownership (contract §2). Go code only under
internal/platform/workflow/…andinternal/modules/workflow/…. Migrations only00030–00049, and this plan uses exactly00030–00033. Provisioning step orders only60–69; this plan uses60. Permission keys onlyworkflow.*. HTTP paths only/v1/workflow/…. Never editinternal/platform/posting. Never editinternal/platform/provisioning/steps.go(register the step from the workflow package instead). - Shared files are append-only (contract §3):
constants.yaml(permissions list),api/openapi.yaml,internal/app/app.go, the module imports incmd/api,cmd/migrate,cmd/orgadmin, the lists ininternal/kernel/db/retention_test.go,docs/ROADMAP.md(Plan 3’s lines only). Append your entries; never reorder or rewrite another plan’s. Generated files (internal/kernel/vocab/vocab.gen.go,packages/contracts/vocab.ts,internal/gen/api/api.gen.go) are regenerated, never hand-edited or hand-merged. - Tenant isolation, two layers. Every business table:
org_id UUID NOT NULL,ENABLEandFORCE ROW LEVEL SECURITY, policy namedorg_isolationwithUSINGandWITH CHECKoncurrent_setting('app.current_org', true)::uuid. Every SQL statement naming an org-scoped table carries an explicitorg_id = $npredicate on each org-scoped table it names (directly or througha.org_id = b.org_idjoin equalities); everyINSERTnamesorg_id.internal/kernel/layering/tenantpredicate_test.gofails the build otherwise. - Composite tenant FKs. Every FK between two org-scoped tables is
(org_id, …) REFERENCES t (org_id, …),org_idat the same position on both sides.internal/kernel/db/tenantfk_test.goenforces it. - Retention. Every new table appears in exactly one list in
internal/kernel/db/retention_test.go. Published definition tables and the decision log areappendOnly(noUPDATE, noDELETEgranted toerp_app). Instances and delegations areamendable(noDELETE). Drafts areoperational. - Every state change writes
audit.Writeon the samepgx.Txas the change. - Lock order. This plan adds no advisory lock class. Serialization is by
SELECT … FOR UPDATEon the row being changed and by unique constraints. InsideTransition,posting.Dispatchruns before the engine’saudit.Write, so any advisory lock aPostertakes is acquired before class 1 (the rule onlockClassAuditChainininternal/kernel/audit/audit.go). - No nested
db.InTx. Platform functions take(ctx, tx pgx.Tx, …)and never open a transaction. Only handlers calldb.InTx, once per request, reads included. - Money is
money.Amountend to end,NUMERIC(19,4)in Postgres, a string on the wire. Read a numeric column ascol::textinto a*stringandmoney.Parseit; never scan into a float. - Dates come from
kernel/clock. The business date isclock.TodayUTC(c); nevertime.Now()in code under test. Every function that stamps a time or decides “today” takes aclock.Clock. - Errors are RFC 7807. Handlers
c.Error(apperr.New(kind, detail))and return;httpx.Errors()renders. Platform code returns sentinel or typed errors, never*apperr.Problem. - Every route goes through
httpx.Protectedwith aworkflow.*permission registered bymember.RegisterPermissionin the module’sinit().api/openapi.yamldocuments every route in the same commit that registers it (internal/app/routes_test.gofails otherwise). - Presence before absence (
AGENTS.md). Every test asserting something is refused, empty or invisible first shows it permitted, present or visible without the mechanism under test. Watch every new guard fail once. postingis a process-global registry. Tests that register a fakePostercallposting.ResetForTest()before registering and int.Cleanup, and never uset.Parallel().- Tests, during development: only the changed packages (
go test ./internal/platform/workflow/... -count=1, etc.). Nevergo test ./...until Task 12, which runs the full suite once. - Record-bearing free text is English or Bahasa Malaysia (§11A.3). Seeded labels ship
enandzh(the supported UI locales,kernel/locale.Supported). - Commits end with the line
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>.
Rulings
Section titled “Rulings”Decisions taken where the specs are silent or ambiguous. Each is Ruling: decision — why — cost if wrong.
- Ruling: drafts are one mutable JSONB row per kind (
workflow_draft, operational); publishing materializes the graph into insert-only tables (workflow_definition/_state/_edge/_gate_grant) with nostatuscolumn, and “current” meansmax(version)— a published version must be append-only (UPDATE and DELETE revoked) while an editor needs replace semantics, and a status column is an UPDATE — cost: rolling back to an old version means republishing its graph as v(n+1), which is one API call. - Ruling: the spec’s columns are typed rather than polymorphic —
semantic_paramsbecomescontrol_kind(CHECKed againstvocab.ControlKinds),grantee_refbecomesgrantee_member_id/grantee_position_id/levels_up, the undefined edgeguardis omitted, and the instance pinsdefinition_id(whose row carries the version) instead of duplicatingdefinition_version— typed columns can carry composite tenant FKs and CHECKs, a JSON blob cannot — cost: a second semantic parameter or a guard language needs an additive migration. - Ruling: a terminal state may have exits, but only into
dispute-semantic states — §6.3’s flow hasreceived(the end) →disputed, and rule 2 (“every non-terminal state has an exit”) needs a notion of terminal that survives it — cost: an org wanting another post-terminal edge is refused at publish. - Ruling: rule 3 is evaluated on the graph with
rejected/cancelledstates removed (they reverse, so they reset the count): no cycle there may contain anaccrualorpayoutstate; fromdraftand from every exit of a reversing state, no path crosses two accruals or two payouts;accrual_firstmeans no path reaches a payout without first crossing an accrual,payout_firstmeans no accrual precedes a payout and every path from a payout to a terminal crosses an accrual; graphs are capped at 64 states / 256 edges / 256 grants — “on any path” is infinite on a graph with cycles, and this is the reading that forbids re-posting — cost: an exotic graph the admin believes safe may be refused; the cap bounds the checker. - Ruling: rule 5’s “reversal action” is a
reverses_accrualflag on the edge; at runtime the reversal is driven byworkflow_instance.accrual_open, passed asposting.Event.PastAccrual— the validator can only see the graph, the runtime knows the instance’s history — cost: none found; the flag is a declaration the validator forces, the posting decision never trusts it alone. - Ruling: rule 6’s “forward edge” is the single
target-gated edge leaving the same state, and anadvanceedge is authorized by the grants on that forward edge’s destination — §5.3 says “grants on the state this rejection replaces” and rule 6 requires exactly one such edge — cost: a state with two target exits cannot carry an advance-gated rejection (usesourceortargetgating there). - Ruling: a reason is mandatory on every edge into a
dispute,rejectedorcancelledstate (validator rule 0), and the member who entered a dispute state can never take asource-gated exit from it — §5.8 “dispute with mandatory reason”, §5.3 “the raiser cannot close their own”, and an auditor asks why a claim was rejected — cost: one more field on rejections. - Ruling: thresholds are accepted only in the organization’s base currency (
min_amount = min_amount_input, currency =company_profile.base_currency); a foreign-currency floor is refused at save until Plan 4 provides FX rates. The columns exist now. — §5.4 converts once at save and Plan 3 has no rate source — cost: an org wanting “USD 500” policies waits for Plan 4; no migration needed then. - Ruling: role sets and quorum are deferred — no
role_set_idcolumn in Plan 3 — neither is on Plan 3’s roadmap lines, and a role set rewriting grant rows conflicts with insert-only published versions (it would have to regenerate a new version) — cost: an additive migration and a generator later; until then grants are per person or dynamic kinds. - Ruling: escalation is evaluated lazily at authorization time, never auto-approves, and needs no worker: once
today − entered_state_on ≥ escalate_after_daysof the state being left, the holders ofNextFilledAbove(primary position of each winning-band approver, grant.max_escalation_levels)may also decide, acting on the escalated-from approver’s band, recorded asauthority_via = 'escalation'— §5.4 “escalation walks up the tree”; an auto-advance would approve money with no human decision, which §11A.4 cannot trace to a person — cost: no proactive notification (no notification platform exists yet); a worker can be added later without changing the rule. - Ruling: delegation is self-service (a member lends their own authority;
workflow.delegation.create), dated in whole days from today onward, optionally scoped to one kind, one hop only (borrowed authority is not re-lent), and a delegate never acts on an instance whose reporter is the delegate or the delegator — §17.4 “the delegate acts on the delegator’s thresholds, and the decision log records both”; retroactive or chained authority is untraceable — cost: an administrator cannot set up a delegation for an absent member in Plan 3. - Ruling:
all_statesgrants (a) join every state’s ladder, (b) exempt their holder from the self-approval refusal only on edges intoaccrualorpayoutstates, (c) give standing onreporter-gated edges except those intoacknowledgedstates — the three properties §5.4 lists, read literally — cost: none beyond the literal reading. - Ruling: submission (any transition out of a
draft-semantic state into a non-draft, non-reversing state) resolves every approval step (everytargetedge’s destination) for this claimant and amount and is refused, naming each gap, when (a) asuperiorgrant with floor ≤ amount resolves to nobody, or (b) the step’s winning band, minus the claimant, is empty. Other grants resolving to nobody are absent. A dynamic grant resolving to the claimant alone escalates one level (NextFilledAbovefrom the claimant’s primary seat) — §5.4’s refusal-not-silence plus rule 7 re-checked at submission, specialized to this amount — cost: a superior-less claimant is refused even when another band would have covered the claim; the fix is the org chart, which §5.4 wants forced. (Amended in final review: the parenthetical originally read “any transition out of adraft-semantic state”, which made withdrawing a draft intocancelled/rejectedneed a routable chain, so a superior-less claimant could never withdraw. Withdrawal asks nobody to decide, so it is not a submission.) - Ruling: Plan 3 exposes no generic instance-transition HTTP endpoint; consumers (Plan 5+) own
POST /v1/<resource>/{id}/transitions, its role permission and itsIdempotency-Key. The engine’sTransitionInputrequires theFromstate the caller saw and refuses a stale one — §4.3 keeps “can this role do this verb” with the owning module;Frommakes a replayed request a 409 instead of a second move — cost: a replay after a lost response gets 409, not the original response, until the consumer adds its key store. - Ruling: the instance stores its amount;
SetAmountis allowed only while the instance is in adraft-semantic state; every decision snapshots the amount it was taken at — authority is computed against an amount, and an amount that moved under an approval would reassign the approver after the fact — cost: partial rejection after submission (Plan 5) needs its own engine call then. - Ruling: the seeded v1 grants are
reviewed ← superior,approved ← unit_head,category_verified,completedanddisputed ← position OWNER, all floors NULL; dispute exits aredisputed → received(dismiss) anddisputed → rejected(uphold, reverses accrual);completedis never re-entered; onlyexpense_claimis seeded, by step 60workflow_definitions, registered frominternal/platform/workflow’sinit()— the only seat provisioning guarantees isOWNER; re-enteringcompletedwould post a second payout; the other §20.2 kinds belong to Plans 7–8 — cost: a fresh org’s owner cannot file their own claim until the chart gives them a superior (§5.4 wants exactly that refusal). - Ruling (flag, not a change):
posting.Eventcannot say “past payout”, so an upheld dispute after payout reaches the Poster only asrejectedwithPastAccrual = true. Plan 3 does not edit the seam (contract §1); this is raised for the seam owner — cost if unaddressed: Plan 4’s Poster reverses the accrual but not the cash leg of a disputed payment. - Ruling: labels are translation rows keyed to the published row ids (
workflow_state/label,workflow_edge/action_label), seeded inenandzh; any label change is a new version — §20.1.6 “every user-facing name is a translation row” and published rows are immutable — cost: a rename does not reach instances pinned to older versions. - Ruling: gate kinds, grantee kinds, posting orders and vacancy policies are Go constants plus OpenAPI enums, not
constants.yamlentries; only permission keys are appended there — they cross the wire (§20.1.9 puts wire enums in OpenAPI) and the contract names only the permissions list as shared — cost: none. - Ruling: the engine writes no
activity_logrows and does not validate the language of decision reasons — the user-facing feed belongs to the consumer that knows what a “claim” is, and §11A.3’s language check is a separate, unticked drift test — cost: Plan 5 must record activity itself. - Ruling: platform SQL is raw pgx, like
orgstruct; nodb/queries/workflow*.sqlfiles — the house pattern in every platform package, and sqlc cannot express the composite FK-driven queries here more safely — cost: none; the tenant-predicate guard scans Go SQL too. - Ruling: kinds are free identifiers matching
^[a-z][a-z0-9_]{0,62}$, one instance per(kind, source_id)— no kind registry exists and a definition for an unused kind is harmless — cost: a typo’d kind in a consumer fails atStartwith “no published definition”.
File Structure
Section titled “File Structure”New migrations (db/migrations/)
Section titled “New migrations (db/migrations/)”| File | Responsibility |
|---|---|
00030_workflow_definition.sql |
workflow_definition, workflow_state, workflow_edge, workflow_gate_grant — insert-only published versions. |
00031_workflow_draft.sql |
workflow_draft — one editable graph per kind. |
00032_workflow_delegation.sql |
workflow_delegation. |
00033_workflow_instance.sql |
workflow_instance, workflow_decision (append-only). |
New Go packages
Section titled “New Go packages”| Path | Responsibility |
|---|---|
internal/platform/workflow/model.go |
Graph types, constants, Definition, graph helpers (GateState, ApprovalStates, GrantsOn). |
internal/platform/workflow/validate.go |
ValidateGraph — pure, rules 0–7. |
internal/platform/workflow/sqlhelp.go |
orgOf, idSet — the two helpers every DB file shares. |
internal/platform/workflow/store.go |
Drafts, Validate (DB-aware), PublishGraph, Publish, Load, Latest, LoadVersion, ListKinds, labels. |
internal/platform/workflow/resolve.go |
Grantee resolution over orgstruct, gaps, vacancy escalation, self-replacement. |
internal/platform/workflow/delegation.go |
Delegation create / revoke / list / active lookup. |
internal/platform/workflow/authorize.go |
Ladder, Route, Authorize (self-approval, all_states, delegation, escalation). |
internal/platform/workflow/instance.go |
Start, Get, GetBySource, SetAmount, Transition, AvailableTransitions, Decisions. |
internal/platform/workflow/seed.go |
DefaultExpenseClaim, SeedDefaults, provisioning step 60 and its readiness warnings. |
internal/platform/workflow/workflowtest/fixture.go |
Test world: org, company profile, unit, seats, members. Test-only helpers (like identitytest). |
internal/modules/workflow/ |
permissions.go, routes.go, decode.go, definitions.go, drafts.go, delegations.go, instances.go, tests. |
Modified (append-only where shared)
Section titled “Modified (append-only where shared)”| Path | Change |
|---|---|
internal/kernel/db/retention_test.go |
Eight tables appended to three lists. |
constants.yaml + regenerated vocab.gen.go, packages/contracts/vocab.ts |
Four workflow.* permissions appended. |
api/openapi.yaml + regenerated internal/gen/api/api.gen.go |
Every /workflow/… operation and its schemas, appended. |
internal/app/app.go, internal/app/routes_test.go |
Register the module; composedModules entry. |
cmd/api/main.go, cmd/migrate/main.go, cmd/orgadmin/main.go, both wiring_test.go |
Link modules/workflow; assert its permission and provisioning step are visible. |
docs/ROADMAP.md |
Tick Plan 3’s lines. |
Task 1: Migrations 00030–00033, retention lists, and the test world
Section titled “Task 1: Migrations 00030–00033, retention lists, and the test world”Files:
- Create:
db/migrations/00030_workflow_definition.sql,db/migrations/00031_workflow_draft.sql,db/migrations/00032_workflow_delegation.sql,db/migrations/00033_workflow_instance.sql - Create:
internal/platform/workflow/workflowtest/fixture.go,internal/platform/workflow/workflowtest/fixture_test.go - Modify:
internal/kernel/db/retention_test.go(append to three lists) - Test: existing
internal/kernel/db/rlscoverage_test.go,retention_test.go,tenantfk_test.go; newworkflowtest/fixture_test.go
Interfaces:
-
Produces tables and columns every later task uses verbatim:
workflow_definition(id, org_id, kind, version, posting_order, published_at, published_by_member_id),workflow_state(id, org_id, definition_id, state_key, semantic, control_kind, display_order, is_terminal, escalate_after_days),workflow_edge(id, org_id, definition_id, from_key, to_key, gate, action_key, requires_reason, reverses_accrual, display_order),workflow_gate_grant(id, org_id, definition_id, state_key, all_states, grantee_kind, grantee_member_id, grantee_position_id, levels_up, min_amount, min_amount_input, min_amount_currency, on_vacancy, max_escalation_levels),workflow_draft(id, org_id, kind, revision, base_version, document, updated_at, updated_by_member_id),workflow_delegation(id, org_id, delegator_member_id, delegate_member_id, kind, valid_from, valid_to, reason, created_at, revoked_at),workflow_instance(id, org_id, definition_id, kind, source_id, reporter_member_id, state_key, amount, currency, accrual_open, entered_state_at, entered_state_on, created_at),workflow_decision(…)as below. -
Produces package
workflowtest:var T0 = time.Date(2026, 9, 21, 9, 0, 0, 0, time.UTC) // the fixed clock's instant (a Monday)var Today = time.Date(2026, 9, 21, 0, 0, 0, 0, time.UTC) // clock.TodayUTC(clock.Fixed(T0)); pass this as `on`var Jan1 = time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)type World struct {T *testing.TH *testdb.HandlesDB *db.DBAdmin *pgxpool.PoolCtx context.Context // tenancy.WithOrg(ctx, Org)Org uuid.UUIDUnit uuid.UUID // root unit "HQ"Clock clock.Clock // clock.Fixed(T0)}func NewWorld(t *testing.T) *Worldfunc (w *World) SecondOrg() *World // same database, another tenantfunc (w *World) Member(name string) uuid.UUIDfunc (w *World) Seat(code string, reportsTo *uuid.UUID) uuid.UUIDfunc (w *World) Sit(member, seat uuid.UUID) // primary, from Jan1, open-endedfunc (w *World) SitSecondary(member, seat uuid.UUID)func (w *World) SetUnitHead(seat *uuid.UUID)func (w *World) DeactivateMember(member uuid.UUID)func (w *World) InTx(fn func(ctx context.Context, tx pgx.Tx) error) errorfunc (w *World) MustTx(fn func(ctx context.Context, tx pgx.Tx) error) -
Step 1: Write
00030_workflow_definition.sql
-- +goose Up
-- PUBLISHED WORKFLOW DEFINITIONS (spec 5.2). One row per (kind, version).---- A published version is never edited: in-flight documents are pinned to-- it (spec 5.2 property 2), so an edit would change the rules a live claim-- is being decided under. Editing happens in workflow_draft (00031);-- publishing inserts a new version here. There is deliberately no status-- column — "current" is max(version) for the kind — because a status column-- would need UPDATE, and UPDATE is not granted on this table.CREATE TABLE workflow_definition ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), org_id UUID NOT NULL REFERENCES organization (id), kind TEXT NOT NULL CHECK (kind ~ '^[a-z][a-z0-9_]{0,62}$'), version INT NOT NULL CHECK (version >= 1), posting_order TEXT NOT NULL CHECK (posting_order IN ('accrual_first', 'payout_first')), published_at TIMESTAMPTZ NOT NULL, -- NULL for the provisioning seed, which no member performs. published_by_member_id UUID, UNIQUE (org_id, kind, version), UNIQUE (org_id, id), CONSTRAINT workflow_definition_publisher_fk FOREIGN KEY (org_id, published_by_member_id) REFERENCES organization_member (org_id, id));
-- States are stable string keys scoped to a definition version (spec 5.2-- property 1). semantic is the closed, system-owned set accounting listens-- to (spec 5.5); control_kind is semantic_params.control_account, typed.CREATE TABLE workflow_state ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), org_id UUID NOT NULL REFERENCES organization (id), definition_id UUID NOT NULL, state_key TEXT NOT NULL CHECK (state_key ~ '^[a-z][a-z0-9_]{0,62}$'), semantic TEXT NOT NULL CHECK (semantic IN ('draft', 'pending', 'accrual', 'payout', 'acknowledged', 'dispute', 'rejected', 'cancelled', 'custom')), control_kind TEXT CHECK (control_kind IN ('ap', 'employee_advances', 'petty_cash_ap', 'ar', 'bank', 'fx_gain_loss')), display_order INT NOT NULL, is_terminal BOOLEAN NOT NULL, escalate_after_days INT CHECK (escalate_after_days BETWEEN 1 AND 365), CHECK ((semantic = 'accrual') = (control_kind IS NOT NULL)), UNIQUE (org_id, definition_id, state_key), UNIQUE (org_id, id), CONSTRAINT workflow_state_definition_fk FOREIGN KEY (org_id, definition_id) REFERENCES workflow_definition (org_id, id));
CREATE TABLE workflow_edge ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), org_id UUID NOT NULL REFERENCES organization (id), definition_id UUID NOT NULL, from_key TEXT NOT NULL, to_key TEXT NOT NULL, gate TEXT NOT NULL CHECK (gate IN ('reporter', 'target', 'source', 'advance')), action_key TEXT NOT NULL CHECK (action_key ~ '^[a-z][a-z0-9_]{0,62}$'), requires_reason BOOLEAN NOT NULL DEFAULT false, -- Spec 5.7 rule 5's "reversal action". The validator forces it on every -- edge from a state at or past accrual into rejected/cancelled; the -- runtime reversal is driven by workflow_instance.accrual_open. reverses_accrual BOOLEAN NOT NULL DEFAULT false, display_order INT NOT NULL DEFAULT 0, CHECK (from_key <> to_key), UNIQUE (org_id, definition_id, from_key, to_key), UNIQUE (org_id, id), CONSTRAINT workflow_edge_definition_fk FOREIGN KEY (org_id, definition_id) REFERENCES workflow_definition (org_id, id), CONSTRAINT workflow_edge_from_fk FOREIGN KEY (org_id, definition_id, from_key) REFERENCES workflow_state (org_id, definition_id, state_key), CONSTRAINT workflow_edge_to_fk FOREIGN KEY (org_id, definition_id, to_key) REFERENCES workflow_state (org_id, definition_id, state_key));
-- THE THRESHOLD IS A COLUMN ON THE GRANT ROW (spec 5.4). There is no-- step-level threshold anywhere. min_amount NULL means "every amount", not-- "unconfigured". Floors only, never ceilings: a band's ceiling is the next-- larger floor on the same step, derived at read time.---- min_amount is the base-currency floor the engine compares against;-- min_amount_input + min_amount_currency are what the admin typed, kept-- because "USD 500" and "MYR 2,235" are different policies. Plan 3 accepts-- base currency only, so today min_amount = min_amount_input.CREATE TABLE workflow_gate_grant ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), org_id UUID NOT NULL REFERENCES organization (id), definition_id UUID NOT NULL, state_key TEXT, all_states BOOLEAN NOT NULL DEFAULT false, grantee_kind TEXT NOT NULL CHECK (grantee_kind IN ('member', 'superior', 'position', 'unit_head', 'levels_up')), grantee_member_id UUID, grantee_position_id UUID, levels_up INT CHECK (levels_up BETWEEN 1 AND 9), min_amount NUMERIC(19,4) CHECK (min_amount >= 0), min_amount_input NUMERIC(19,4) CHECK (min_amount_input >= 0), min_amount_currency TEXT CHECK (min_amount_currency ~ '^[A-Z]{3}$'), on_vacancy TEXT NOT NULL DEFAULT 'escalate' CHECK (on_vacancy IN ('escalate', 'refuse')), max_escalation_levels INT NOT NULL DEFAULT 2 CHECK (max_escalation_levels BETWEEN 1 AND 10), CHECK (all_states = (state_key IS NULL)), CHECK ((min_amount IS NULL) = (min_amount_input IS NULL)), CHECK ((min_amount IS NULL) = (min_amount_currency IS NULL)), CHECK ((grantee_kind = 'member') = (grantee_member_id IS NOT NULL)), CHECK ((grantee_kind = 'position') = (grantee_position_id IS NOT NULL)), CHECK ((grantee_kind = 'levels_up') = (levels_up IS NOT NULL)), UNIQUE (org_id, id), CONSTRAINT workflow_gate_grant_definition_fk FOREIGN KEY (org_id, definition_id) REFERENCES workflow_definition (org_id, id), CONSTRAINT workflow_gate_grant_state_fk FOREIGN KEY (org_id, definition_id, state_key) REFERENCES workflow_state (org_id, definition_id, state_key), CONSTRAINT workflow_gate_grant_member_fk FOREIGN KEY (org_id, grantee_member_id) REFERENCES organization_member (org_id, id), CONSTRAINT workflow_gate_grant_position_fk FOREIGN KEY (org_id, grantee_position_id) REFERENCES org_position (org_id, id));CREATE INDEX workflow_gate_grant_definition_idx ON workflow_gate_grant (org_id, definition_id);
ALTER TABLE workflow_definition ENABLE ROW LEVEL SECURITY;ALTER TABLE workflow_definition FORCE ROW LEVEL SECURITY;CREATE POLICY org_isolation ON workflow_definition USING (org_id = current_setting('app.current_org', true)::uuid) WITH CHECK (org_id = current_setting('app.current_org', true)::uuid);ALTER TABLE workflow_state ENABLE ROW LEVEL SECURITY;ALTER TABLE workflow_state FORCE ROW LEVEL SECURITY;CREATE POLICY org_isolation ON workflow_state USING (org_id = current_setting('app.current_org', true)::uuid) WITH CHECK (org_id = current_setting('app.current_org', true)::uuid);ALTER TABLE workflow_edge ENABLE ROW LEVEL SECURITY;ALTER TABLE workflow_edge FORCE ROW LEVEL SECURITY;CREATE POLICY org_isolation ON workflow_edge USING (org_id = current_setting('app.current_org', true)::uuid) WITH CHECK (org_id = current_setting('app.current_org', true)::uuid);ALTER TABLE workflow_gate_grant ENABLE ROW LEVEL SECURITY;ALTER TABLE workflow_gate_grant FORCE ROW LEVEL SECURITY;CREATE POLICY org_isolation ON workflow_gate_grant USING (org_id = current_setting('app.current_org', true)::uuid) WITH CHECK (org_id = current_setting('app.current_org', true)::uuid);
-- SELECT and INSERT only. A published version is the rulebook a live claim-- is decided under, and the evidence of which rules applied to a decision-- taken seven years ago (spec 11A.4, ITA s.82A). retention_test.go lists-- these four as appendOnly and fails the build if UPDATE or DELETE returns.GRANT SELECT, INSERT ON workflow_definition, workflow_state, workflow_edge, workflow_gate_grant TO erp_app;
-- +goose DownDROP TABLE workflow_gate_grant;DROP TABLE workflow_edge;DROP TABLE workflow_state;DROP TABLE workflow_definition;- Step 2: Write
00031_workflow_draft.sql
-- +goose Up
-- THE EDITABLE GRAPH. One per kind per organization. The whole graph is one-- JSONB document, replaced on every save, because the editor saves the-- graph as a unit and a half-saved graph (a state removed, its edges not)-- is meaningless. It is validated on every save but may be saved invalid:-- the editor shows the violations live (spec 5.7) and only publish refuses.---- A draft is not a record. No claim is ever decided under it, so it is-- operational data: it is deleted when it is published or discarded. Every-- save is still audited.CREATE TABLE workflow_draft ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), org_id UUID NOT NULL REFERENCES organization (id), kind TEXT NOT NULL CHECK (kind ~ '^[a-z][a-z0-9_]{0,62}$'), -- Optimistic concurrency: a save names the revision it was edited from, -- so two administrators editing the same flow cannot silently overwrite -- each other (spec 12.2's If-Match, expressed in the body). revision INT NOT NULL CHECK (revision >= 1), base_version INT, document JSONB NOT NULL, updated_at TIMESTAMPTZ NOT NULL, updated_by_member_id UUID NOT NULL, UNIQUE (org_id, kind), UNIQUE (org_id, id), CONSTRAINT workflow_draft_updated_by_fk FOREIGN KEY (org_id, updated_by_member_id) REFERENCES organization_member (org_id, id));
ALTER TABLE workflow_draft ENABLE ROW LEVEL SECURITY;ALTER TABLE workflow_draft FORCE ROW LEVEL SECURITY;CREATE POLICY org_isolation ON workflow_draft USING (org_id = current_setting('app.current_org', true)::uuid) WITH CHECK (org_id = current_setting('app.current_org', true)::uuid);
GRANT SELECT, INSERT, UPDATE, DELETE ON workflow_draft TO erp_app;
-- +goose DownDROP TABLE workflow_draft;- Step 3: Write
00032_workflow_delegation.sql
-- +goose Up
-- AUTHORITY LENT, NOT REPLACED (spec 17.4). The delegate acts on the-- delegator's thresholds, and the decision log names both.---- Dated in whole days (the business date, kernel/clock UTC). valid_to NULL-- is open-ended; revoking stamps revoked_at and never deletes, because a-- decision taken under a delegation must stay explicable after it ends.CREATE TABLE workflow_delegation ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), org_id UUID NOT NULL REFERENCES organization (id), delegator_member_id UUID NOT NULL, delegate_member_id UUID NOT NULL, -- NULL: every kind. Otherwise one workflow kind. kind TEXT CHECK (kind ~ '^[a-z][a-z0-9_]{0,62}$'), valid_from DATE NOT NULL, valid_to DATE, reason TEXT NOT NULL CHECK (length(btrim(reason)) > 0), created_at TIMESTAMPTZ NOT NULL, revoked_at TIMESTAMPTZ, CHECK (delegator_member_id <> delegate_member_id), CHECK (valid_to IS NULL OR valid_to > valid_from), UNIQUE (org_id, id), CONSTRAINT workflow_delegation_delegator_fk FOREIGN KEY (org_id, delegator_member_id) REFERENCES organization_member (org_id, id), CONSTRAINT workflow_delegation_delegate_fk FOREIGN KEY (org_id, delegate_member_id) REFERENCES organization_member (org_id, id));CREATE INDEX workflow_delegation_delegate_idx ON workflow_delegation (org_id, delegate_member_id);
ALTER TABLE workflow_delegation ENABLE ROW LEVEL SECURITY;ALTER TABLE workflow_delegation FORCE ROW LEVEL SECURITY;CREATE POLICY org_isolation ON workflow_delegation USING (org_id = current_setting('app.current_org', true)::uuid) WITH CHECK (org_id = current_setting('app.current_org', true)::uuid);
GRANT SELECT, INSERT, UPDATE ON workflow_delegation TO erp_app;
-- +goose DownDROP TABLE workflow_delegation;- Step 4: Write
00033_workflow_instance.sql
-- +goose Up
-- ONE DOCUMENT'S JOURNEY THROUGH ONE PINNED DEFINITION VERSION.---- definition_id pins the version (spec 5.2 property 2): editing publishes-- v(n+1) and touches nothing here. The FK on (definition_id, state_key)-- makes a state key that does not exist in the pinned version-- unrepresentable.---- accrual_open is the runtime half of spec 5.7 rule 5: set on entering an-- accrual state, cleared on entering rejected/cancelled. It is what-- posting.Event.PastAccrual carries, so a rejection reverses exactly when-- something was accrued and not already reversed.CREATE TABLE workflow_instance ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), org_id UUID NOT NULL REFERENCES organization (id), definition_id UUID NOT NULL, kind TEXT NOT NULL, source_id UUID NOT NULL, reporter_member_id UUID NOT NULL, state_key TEXT NOT NULL, amount NUMERIC(19,4) NOT NULL CHECK (amount >= 0), currency TEXT NOT NULL CHECK (currency ~ '^[A-Z]{3}$'), accrual_open BOOLEAN NOT NULL DEFAULT false, entered_state_at TIMESTAMPTZ NOT NULL, -- The business date the current state was entered; escalation counts -- whole days from it (kernel/clock, UTC). entered_state_on DATE NOT NULL, created_at TIMESTAMPTZ NOT NULL, UNIQUE (org_id, kind, source_id), UNIQUE (org_id, id), CONSTRAINT workflow_instance_definition_fk FOREIGN KEY (org_id, definition_id) REFERENCES workflow_definition (org_id, id), CONSTRAINT workflow_instance_state_fk FOREIGN KEY (org_id, definition_id, state_key) REFERENCES workflow_state (org_id, definition_id, state_key), CONSTRAINT workflow_instance_reporter_fk FOREIGN KEY (org_id, reporter_member_id) REFERENCES organization_member (org_id, id));CREATE INDEX workflow_instance_state_idx ON workflow_instance (org_id, definition_id, state_key);
-- THE IMMUTABLE DECISION LOG (spec 5.8, 11A.4).---- Every column needed to explain the decision seven years later is written-- HERE, at decision time, and never recomputed: orgstruct.SuperiorOf reads-- reporting lines and unit heads as they stand NOW (org-structure spec 5.5),-- so re-resolving later would explain a March approval with November's-- chart. Hence eligible_member_ids, resolution (every grant on the step and-- whom it resolved to, with its floor and any gap), winning_floor and-- resolved_for.---- actor_member_id did it; authority_member_id is whose grant and band was-- relied on. They differ under delegation (spec 17.4: "names the delegate as-- actor and the delegator as the authority relied on") and escalation.CREATE TABLE workflow_decision ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), org_id UUID NOT NULL REFERENCES organization (id), instance_id UUID NOT NULL, definition_id UUID NOT NULL, from_key TEXT NOT NULL, to_key TEXT NOT NULL, gate TEXT NOT NULL CHECK (gate IN ('reporter', 'target', 'source', 'advance')), action_key TEXT NOT NULL, actor_member_id UUID NOT NULL, authority_member_id UUID NOT NULL, authority_via TEXT NOT NULL CHECK (authority_via IN ('reporter', 'grant', 'delegation', 'escalation')), delegation_id UUID, grant_id UUID, amount NUMERIC(19,4) NOT NULL, currency TEXT NOT NULL, winning_floor NUMERIC(19,4), resolved_for DATE NOT NULL, eligible_member_ids UUID[] NOT NULL, resolution JSONB NOT NULL, reason TEXT, ip INET, request_id TEXT, decided_at TIMESTAMPTZ NOT NULL, -- Order within an instance. decided_at can tie (one clock reading per -- request, a fixed clock in tests); "the latest decision into this state" -- must not depend on a random uuid, so the log carries its own sequence, -- as audit_log does (00007). seq BIGSERIAL, CHECK ((authority_via = 'delegation') = (delegation_id IS NOT NULL)), UNIQUE (org_id, id), CONSTRAINT workflow_decision_instance_fk FOREIGN KEY (org_id, instance_id) REFERENCES workflow_instance (org_id, id), CONSTRAINT workflow_decision_from_fk FOREIGN KEY (org_id, definition_id, from_key) REFERENCES workflow_state (org_id, definition_id, state_key), CONSTRAINT workflow_decision_to_fk FOREIGN KEY (org_id, definition_id, to_key) REFERENCES workflow_state (org_id, definition_id, state_key), CONSTRAINT workflow_decision_actor_fk FOREIGN KEY (org_id, actor_member_id) REFERENCES organization_member (org_id, id), CONSTRAINT workflow_decision_authority_fk FOREIGN KEY (org_id, authority_member_id) REFERENCES organization_member (org_id, id), CONSTRAINT workflow_decision_grant_fk FOREIGN KEY (org_id, grant_id) REFERENCES workflow_gate_grant (org_id, id), CONSTRAINT workflow_decision_delegation_fk FOREIGN KEY (org_id, delegation_id) REFERENCES workflow_delegation (org_id, id));CREATE INDEX workflow_decision_instance_idx ON workflow_decision (org_id, instance_id, seq);
ALTER TABLE workflow_instance ENABLE ROW LEVEL SECURITY;ALTER TABLE workflow_instance FORCE ROW LEVEL SECURITY;CREATE POLICY org_isolation ON workflow_instance USING (org_id = current_setting('app.current_org', true)::uuid) WITH CHECK (org_id = current_setting('app.current_org', true)::uuid);ALTER TABLE workflow_decision ENABLE ROW LEVEL SECURITY;ALTER TABLE workflow_decision FORCE ROW LEVEL SECURITY;CREATE POLICY org_isolation ON workflow_decision USING (org_id = current_setting('app.current_org', true)::uuid) WITH CHECK (org_id = current_setting('app.current_org', true)::uuid);
GRANT SELECT, INSERT, UPDATE ON workflow_instance TO erp_app;-- Append-only. A correction is a further decision, never an edit.GRANT SELECT, INSERT ON workflow_decision TO erp_app;-- BIGSERIAL's sequence is owned by the migration role (see 00007).GRANT USAGE ON SEQUENCE workflow_decision_seq_seq TO erp_app;
-- +goose DownDROP TABLE workflow_decision;DROP TABLE workflow_instance;- Step 5: Append the eight tables to the retention lists
In internal/kernel/db/retention_test.go, append (do not reorder existing entries):
appendOnly := []string{ "audit_log", // Plan 3. A published workflow version is the rulebook an in-flight // document is decided under and the evidence of which rules applied // to a past decision; the decision log is spec 5.8's immutable log. // Corrections are new versions and further decisions, never edits. "workflow_definition", "workflow_state", "workflow_edge", "workflow_gate_grant", "workflow_decision", } // Plan 3. An instance's current state and a delegation's revocation // are amended in place; neither may be erased while a decision refers // to it. "workflow_instance", "workflow_delegation",(into amendable), and
// Plan 3. A draft is never decided under; it is deleted on publish // or discard. Every save is audited. "workflow_draft",(into operational).
- Step 6: Run the drift tests and watch them pass for the right reason
Run: go test ./internal/kernel/db/ -run 'TestRLS|TestOnlyOperational|TestEveryTenantForeignKey' -count=1 -v
Expected: PASS. Then temporarily append GRANT DELETE ON workflow_decision TO erp_app; to 00033’s Up, re-run, and confirm TestOnlyOperationalTablesGrantDeleteToTheAppRole goes RED naming workflow_decision. Temporarily change workflow_edge_from_fk to FOREIGN KEY (definition_id, from_key) REFERENCES workflow_state (definition_id, state_key) (adding the matching UNIQUE), confirm TestEveryTenantForeignKeyIsCompositeOnOrgID goes RED naming it. Restore both; confirm green.
- Step 7: Write the failing fixture test
internal/platform/workflow/workflowtest/fixture_test.go:
package workflowtest_test
import ( "context" "testing"
"github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/stretchr/testify/require"
"github.com/mcom/erp/internal/platform/orgstruct" "github.com/mcom/erp/internal/platform/workflow/workflowtest")
// The world must be the real org structure, not an imitation of it: every// resolution test in Plan 3 stands on it. So the fixture is checked against// orgstruct's own resolver.func TestTheWorldIsAChartOrgstructCanRead(t *testing.T) { w := workflowtest.NewWorld(t) boss := w.Member("Boss") staff := w.Member("Staff") top := w.Seat("CEO", nil) under := w.Seat("STAFF", &top) w.Sit(boss, top) w.Sit(staff, under)
var got []uuid.UUID w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var err error got, err = orgstruct.SuperiorOf(ctx, tx, staff, workflowtest.Today) return err }) require.Equal(t, []uuid.UUID{boss}, got)
// A second tenant in the same database sees none of it. other := w.SecondOrg() var n int other.MustTx(func(ctx context.Context, tx pgx.Tx) error { return tx.QueryRow(ctx, `SELECT count(*) FROM org_position WHERE org_id = $1`, w.Org).Scan(&n) }) require.Zero(t, n)}- Step 8: Run it and watch it fail
Run: go test ./internal/platform/workflow/workflowtest/ -count=1
Expected: FAIL — package workflowtest does not exist.
- Step 9: Write
workflowtest/fixture.go
// Package workflowtest builds the smallest real world a workflow test// needs: an organization with a company profile (the base currency grants// are checked against), a root unit, seats with a real materialized path,// and members sitting in them. Rows are seeded through the admin pool so// that no workflow test can go green or red because of a change in how// orgstruct's own constructors work; paths are computed exactly as// orgstruct.childPath does ("/<root>/…/<self>/").package workflowtest
import ( "context" "testing" "time"
"github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "github.com/stretchr/testify/require"
"github.com/mcom/erp/internal/kernel/clock" "github.com/mcom/erp/internal/kernel/db" "github.com/mcom/erp/internal/kernel/db/testdb" "github.com/mcom/erp/internal/kernel/eventbus" "github.com/mcom/erp/internal/kernel/tenancy")
var ( T0 = time.Date(2026, 9, 21, 9, 0, 0, 0, time.UTC) Today = time.Date(2026, 9, 21, 0, 0, 0, 0, time.UTC) // the business date of T0 Jan1 = time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC))
type World struct { T *testing.T H *testdb.Handles DB *db.DB Admin *pgxpool.Pool Ctx context.Context Org uuid.UUID Unit uuid.UUID Clock clock.Clock}
func NewWorld(t *testing.T) *World { t.Helper() h := testdb.Start(t) d, err := db.Open(context.Background(), h.AppDSN, eventbus.New()) require.NoError(t, err) t.Cleanup(d.Close) return newOrg(t, h, d, "Alpha")}
// SecondOrg is another tenant in the same database, for isolation tests.func (w *World) SecondOrg() *World { return newOrg(w.T, w.H, w.DB, "Beta-"+uuid.NewString()[:8]) }
func newOrg(t *testing.T, h *testdb.Handles, d *db.DB, name string) *World { t.Helper() ctx := context.Background() org := uuid.New() _, err := h.Admin.Exec(ctx, `INSERT INTO organization (id, org_id, name) VALUES ($1,$1,$2)`, org, name) require.NoError(t, err) _, err = h.Admin.Exec(ctx, ` INSERT INTO company_profile (org_id, legal_name, country_code, base_currency, fiscal_year_starts_on) VALUES ($1, $2, 'MY', 'MYR', '2026-01-01')`, org, name+" Sdn Bhd") require.NoError(t, err) unit := uuid.New() _, err = h.Admin.Exec(ctx, ` INSERT INTO org_unit (id, org_id, parent_id, code, name, path, depth) VALUES ($1, $2, NULL, 'HQ', 'Head Office', $3, 0)`, unit, org, "/"+unit.String()+"/") require.NoError(t, err) return &World{T: t, H: h, DB: d, Admin: h.Admin, Ctx: tenancy.WithOrg(ctx, org), Org: org, Unit: unit, Clock: clock.Fixed(T0)}}
func (w *World) Member(name string) uuid.UUID { w.T.Helper() ctx := context.Background() var party, member uuid.UUID require.NoError(w.T, w.Admin.QueryRow(ctx, `INSERT INTO party (org_id, kind, legal_name) VALUES ($1,'employee',$2) RETURNING id`, w.Org, name).Scan(&party)) require.NoError(w.T, w.Admin.QueryRow(ctx, `INSERT INTO organization_member (org_id, user_id, party_id) VALUES ($1,$2,$3) RETURNING id`, w.Org, uuid.New(), party).Scan(&member)) return member}
func (w *World) Seat(code string, reportsTo *uuid.UUID) uuid.UUID { w.T.Helper() ctx := context.Background() parentPath, depth := "/", 0 if reportsTo != nil { require.NoError(w.T, w.Admin.QueryRow(ctx, `SELECT path, depth + 1 FROM org_position WHERE org_id = $1 AND id = $2`, w.Org, *reportsTo).Scan(&parentPath, &depth)) } id := uuid.New() _, err := w.Admin.Exec(ctx, ` INSERT INTO org_position (id, org_id, org_unit_id, code, title, reports_to_position_id, path, depth) VALUES ($1,$2,$3,$4,$4,$5,$6,$7)`, id, w.Org, w.Unit, code, reportsTo, parentPath+id.String()+"/", depth) require.NoError(w.T, err) return id}
func (w *World) sit(member, seat uuid.UUID, primary bool) { w.T.Helper() _, err := w.Admin.Exec(context.Background(), ` INSERT INTO member_position (org_id, member_id, position_id, is_primary, valid_from) VALUES ($1,$2,$3,$4,$5)`, w.Org, member, seat, primary, Jan1) require.NoError(w.T, err)}
func (w *World) Sit(member, seat uuid.UUID) { w.sit(member, seat, true) }func (w *World) SitSecondary(member, seat uuid.UUID) { w.sit(member, seat, false) }
func (w *World) SetUnitHead(seat *uuid.UUID) { w.T.Helper() _, err := w.Admin.Exec(context.Background(), `UPDATE org_unit SET unit_head_position_id = $1 WHERE org_id = $2 AND id = $3`, seat, w.Org, w.Unit) require.NoError(w.T, err)}
func (w *World) DeactivateMember(member uuid.UUID) { w.T.Helper() _, err := w.Admin.Exec(context.Background(), `UPDATE organization_member SET active = false WHERE org_id = $1 AND id = $2`, w.Org, member) require.NoError(w.T, err)}
func (w *World) InTx(fn func(ctx context.Context, tx pgx.Tx) error) error { return w.DB.InTx(w.Ctx, fn)}
func (w *World) MustTx(fn func(ctx context.Context, tx pgx.Tx) error) { w.T.Helper() require.NoError(w.T, w.InTx(fn))}- Step 10: Run it and watch it pass
Run: go test ./internal/platform/workflow/workflowtest/ -count=1 -v
Expected: PASS.
- Step 11: Commit
git add db/migrations/0003[0-3]_workflow_*.sql internal/kernel/db/retention_test.go internal/platform/workflow/workflowtest/git commit -m "Add workflow tables and the workflow test world
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>"Task 2: The graph model and the pure publish-time validator (§5.7, all seven rules)
Section titled “Task 2: The graph model and the pure publish-time validator (§5.7, all seven rules)”Files:
- Create:
internal/platform/workflow/model.go,internal/platform/workflow/validate.go,internal/platform/workflow/sqlhelp.go - Test:
internal/platform/workflow/validate_test.go(pure — no database)
Interfaces:
-
Consumes:
posting.Semanticand its constants;vocab.ControlKinds;locale.Supported;money.Amount. -
Produces (every later task uses these names exactly):
type Gate string // GateReporter, GateTarget, GateSource, GateAdvancetype GranteeKind string // GranteeMember, GranteeSuperior, GranteePosition, GranteeUnitHead, GranteeLevelsUptype PostingOrder string // AccrualFirst, PayoutFirsttype OnVacancy string // VacancyEscalate, VacancyRefuseconst DefaultMaxEscalationLevels = 2const MaxStates, MaxEdges, MaxGrants = 64, 256, 256type Labels map[string]stringtype StateSpec struct { Key string; Semantic posting.Semantic; ControlKind string; DisplayOrder int; IsTerminal bool; EscalateAfterDays *int; Labels Labels }type EdgeSpec struct { From, To string; Gate Gate; ActionKey string; RequiresReason, ReversesAccrual bool; DisplayOrder int; Labels Labels }type GrantSpec struct { StateKey string; AllStates bool; GranteeKind GranteeKind; MemberID, PositionID *uuid.UUID; LevelsUp *int; MinAmount *money.Amount; MinAmountCurrency string; OnVacancy OnVacancy; MaxEscalationLevels int }type Graph struct { Kind string; PostingOrder PostingOrder; States []StateSpec; Edges []EdgeSpec; Grants []GrantSpec }func (g Graph) Normalized() Graphfunc (g Graph) State(key string) (StateSpec, bool)func (g Graph) Edge(from, to string) (EdgeSpec, bool)func (g Graph) EdgesFrom(key string) []EdgeSpecfunc (g Graph) GateState(e EdgeSpec) (string, bool)func (g Graph) ApprovalStates() []stringtype Grant struct { ID uuid.UUID; GrantSpec }type Definition struct { ID uuid.UUID; Version int; PublishedAt time.Time; PublishedBy *uuid.UUID; Graph; StateIDs map[string]uuid.UUID; EdgeIDs map[[2]string]uuid.UUID; GrantIDs []uuid.UUID }func (d Definition) GrantsOn(stateKey string) []Grant // the state's own grants plus every all_states granttype Violation struct { Rule int; Code, Detail, Ref string }func ValidateGraph(g Graph, baseCurrency string) []Violationvar KeyPattern *regexp.Regexp // ^[a-z][a-z0-9_]{0,62}$// sqlhelp.go — shared by Tasks 3, 4, 5 so they can run in parallel:func orgOf(ctx context.Context) (uuid.UUID, error)func idSet(ctx context.Context, tx pgx.Tx, sql string, org uuid.UUID, ids []uuid.UUID) (map[uuid.UUID]bool, error) -
Step 1: Write the failing tests
internal/platform/workflow/validate_test.go. The fixture is a minimal legal flow, and every test breaks exactly one thing in a copy of it — so each test first shows the valid graph passes (presence) before showing the broken one fails (absence).
package workflow_test
import ( "testing"
"github.com/google/uuid" "github.com/stretchr/testify/require"
"github.com/mcom/erp/internal/kernel/money" "github.com/mcom/erp/internal/platform/posting" "github.com/mcom/erp/internal/platform/workflow")
func lbl(s string) workflow.Labels { return workflow.Labels{"en": s} }
func st(key string, sem posting.Semantic, terminal bool) workflow.StateSpec { s := workflow.StateSpec{Key: key, Semantic: sem, IsTerminal: terminal, Labels: lbl(key)} if sem == posting.Accrual { s.ControlKind = "ap" } return s}
func ed(from, to string, gate workflow.Gate) workflow.EdgeSpec { return workflow.EdgeSpec{From: from, To: to, Gate: gate, ActionKey: "go_" + to, Labels: lbl(to)}}
func rejecting(from, to string, gate workflow.Gate, reverses bool) workflow.EdgeSpec { e := ed(from, to, gate) e.RequiresReason, e.ReversesAccrual = true, reverses return e}
func anyone(state string) workflow.GrantSpec { return workflow.GrantSpec{StateKey: state, GranteeKind: workflow.GranteeSuperior}}
// valid is draft -> pending -> approved(accrual) -> paid(payout) -> done,// with rejections from pending and approved, and rejected -> draft.func valid() workflow.Graph { return workflow.Graph{ Kind: "expense_claim", PostingOrder: workflow.AccrualFirst, States: []workflow.StateSpec{ st("draft", posting.Draft, false), st("pending", posting.Pending, false), st("approved", posting.Accrual, false), st("paid", posting.Payout, false), st("done", posting.Acknowledged, true), st("rejected", posting.Rejected, false), }, Edges: []workflow.EdgeSpec{ ed("draft", "pending", workflow.GateReporter), ed("pending", "approved", workflow.GateTarget), ed("approved", "paid", workflow.GateTarget), ed("paid", "done", workflow.GateReporter), rejecting("pending", "rejected", workflow.GateAdvance, false), rejecting("approved", "rejected", workflow.GateAdvance, true), ed("rejected", "draft", workflow.GateReporter), }, Grants: []workflow.GrantSpec{anyone("approved"), anyone("paid")}, }}
func codes(vs []workflow.Violation) []string { out := []string{} for _, v := range vs { out = append(out, v.Code) } return out}
func requireOnly(t *testing.T, g workflow.Graph, rule int, code string) { t.Helper() require.Empty(t, workflow.ValidateGraph(valid(), "MYR"), "the base fixture must be valid, or every absence below is vacuous") vs := workflow.ValidateGraph(g, "MYR") require.NotEmpty(t, vs, "expected %s", code) for _, v := range vs { require.Equal(t, rule, v.Rule, "unexpected violation %+v", v) } require.Contains(t, codes(vs), code)}
func TestTheFixtureIsValid(t *testing.T) { require.Empty(t, workflow.ValidateGraph(valid(), "MYR"))}
func TestRule1ExactlyOneDraftAndATerminal(t *testing.T) { g := valid() g.States[1].Semantic = posting.Draft requireOnly(t, g, 1, "draft_count")
g = valid() g.States[4].IsTerminal = false g.Edges = append(g.Edges, ed("done", "draft", workflow.GateReporter)) // keep rule 2 quiet requireOnly(t, g, 1, "no_terminal")}
func TestRule2ReachableAndEveryNonTerminalHasAnExit(t *testing.T) { g := valid() g.States = append(g.States, st("orphan", posting.Custom, true)) requireOnly(t, g, 2, "unreachable")
g = valid() g.Edges = g.Edges[:len(g.Edges)-1] // drop rejected -> draft requireOnly(t, g, 2, "no_exit")}
func TestRule3AtMostOneAccrualAndOnePayout(t *testing.T) { g := valid() g.States = append(g.States, st("reaccrue", posting.Accrual, false)) g.Edges = []workflow.EdgeSpec{ ed("draft", "pending", workflow.GateReporter), ed("pending", "approved", workflow.GateTarget), ed("approved", "reaccrue", workflow.GateTarget), ed("reaccrue", "paid", workflow.GateTarget), ed("paid", "done", workflow.GateReporter), rejecting("pending", "rejected", workflow.GateAdvance, false), ed("rejected", "draft", workflow.GateReporter), } g.Grants = []workflow.GrantSpec{anyone("approved"), anyone("reaccrue"), anyone("paid")} requireOnly(t, g, 3, "accrual_repeated")}
func TestRule3NoPostingStateOnACycleThatDoesNotReverse(t *testing.T) { // paid -> pending re-enters the approval chain without a reversal, so // approved (accrual) and paid (payout) sit on a cycle and would post twice. g := valid() g.Edges = append(g.Edges, ed("paid", "pending", workflow.GateReporter)) vs := workflow.ValidateGraph(g, "MYR") require.Contains(t, codes(vs), "accrual_on_cycle") require.Contains(t, codes(vs), "payout_on_cycle")
// PRESENCE of the escape hatch: the same loop routed through rejected // (which reverses) is legal. g = valid() g.Edges = append(g.Edges, rejecting("paid", "rejected", workflow.GateSource, true)) g.Grants = append(g.Grants, anyone("paid")) require.Empty(t, workflow.ValidateGraph(g, "MYR"))}
func TestRule3OrderingIsDeclaredNotAssumed(t *testing.T) { // Swap the semantics: pay first, accrue after. g := valid() g.States[2].Semantic, g.States[2].ControlKind = posting.Payout, "" g.States[3].Semantic, g.States[3].ControlKind = posting.Accrual, "employee_advances" requireOnly(t, g, 3, "payout_before_accrual")
g.PostingOrder = workflow.PayoutFirst // Rule 5 now wants the reversal flag on the rejection out of the // post-payout accrual chain; the edge from 'approved' (now payout) is // before any accrual, so it may keep reverses=true harmlessly. require.Empty(t, workflow.ValidateGraph(g, "MYR"))}
func TestRule4EveryGatedEdgeHasAGrant(t *testing.T) { g := valid() g.Grants = g.Grants[:1] // drop the grant on 'paid' requireOnly(t, g, 4, "gate_without_grant")
// An all_states grant counts for every state. g.Grants = append(g.Grants, workflow.GrantSpec{AllStates: true, GranteeKind: workflow.GranteeUnitHead}) require.Empty(t, workflow.ValidateGraph(g, "MYR"))}
func TestRule5RejectingPastAccrualMustReverse(t *testing.T) { g := valid() g.Edges[5].ReversesAccrual = false // approved -> rejected requireOnly(t, g, 5, "reversal_missing")}
func TestRule6AdvanceNeedsExactlyOneForwardEdge(t *testing.T) { g := valid() g.States = append(g.States, st("side", posting.Custom, false)) g.Edges = append(g.Edges, ed("pending", "side", workflow.GateTarget), ed("side", "approved", workflow.GateTarget)) g.Grants = append(g.Grants, anyone("side")) requireOnly(t, g, 6, "advance_forward_count")}
func TestRule7BandsReachDownToZero(t *testing.T) { g := valid() five := money.MustParse("5000") g.Grants[0].MinAmount, g.Grants[0].MinAmountCurrency = &five, "MYR" requireOnly(t, g, 7, "bands_do_not_reach_zero")
zero := money.MustParse("0") g.Grants = append(g.Grants, workflow.GrantSpec{StateKey: "approved", GranteeKind: workflow.GranteeUnitHead, MinAmount: &zero, MinAmountCurrency: "MYR"}) require.Empty(t, workflow.ValidateGraph(g, "MYR"))}
func TestWellFormednessIsRuleZeroAndStopsTheGraphRules(t *testing.T) { cases := map[string]func(g *workflow.Graph){ "kind_invalid": func(g *workflow.Graph) { g.Kind = "Expense Claim" }, "state_key_duplicate": func(g *workflow.Graph) { g.States[1].Key = "draft" }, "control_kind_required": func(g *workflow.Graph) { g.States[2].ControlKind = "" }, "control_kind_forbidden": func(g *workflow.Graph) { g.States[1].ControlKind = "ap" }, "control_kind_unknown": func(g *workflow.Graph) { g.States[2].ControlKind = "cash" }, "label_en_missing": func(g *workflow.Graph) { g.States[0].Labels = workflow.Labels{"zh": "草稿"} }, "label_locale_unsupported": func(g *workflow.Graph) { g.States[0].Labels["fr"] = "brouillon" }, "edge_state_unknown": func(g *workflow.Graph) { g.Edges[0].To = "nowhere" }, "edge_duplicate": func(g *workflow.Graph) { g.Edges = append(g.Edges, g.Edges[0]) }, "gate_unknown": func(g *workflow.Graph) { g.Edges[0].Gate = "anyone" }, "reason_required": func(g *workflow.Graph) { g.Edges[4].RequiresReason = false }, "terminal_exit_not_dispute": func(g *workflow.Graph) { g.Edges = append(g.Edges, ed("done", "draft", workflow.GateReporter)) }, "grant_state_unknown": func(g *workflow.Graph) { g.Grants[0].StateKey = "nowhere" }, "grant_scope_invalid": func(g *workflow.Graph) { g.Grants[0].AllStates = true }, "grantee_ref_invalid": func(g *workflow.Graph) { id := uuid.New(); g.Grants[0].MemberID = &id }, "currency_mismatch": func(g *workflow.Graph) { a := money.MustParse("0") g.Grants[0].MinAmount, g.Grants[0].MinAmountCurrency = &a, "USD" }, "amount_negative": func(g *workflow.Graph) { a := money.MustParse("-1") g.Grants[0].MinAmount, g.Grants[0].MinAmountCurrency = &a, "MYR" }, } for code, breakIt := range cases { t.Run(code, func(t *testing.T) { g := valid() breakIt(&g) requireOnly(t, g, 0, code) }) }}
func TestADisputeFromTheTerminalStateIsAllowed(t *testing.T) { g := valid() g.States = append(g.States, st("disputed", posting.Dispute, false)) g.Edges = append(g.Edges, rejecting("done", "disputed", workflow.GateReporter, false), ed("disputed", "done", workflow.GateSource), ) g.Grants = append(g.Grants, anyone("disputed")) require.Empty(t, workflow.ValidateGraph(g, "MYR"))}Note on TestRule3OrderingIsDeclaredNotAssumed: after the swap, approved is payout and paid is accrual; the edge approved → rejected has ReversesAccrual = true — harmless, rule 5 forbids only its absence.
- Step 2: Run them and watch them fail
Run: go test ./internal/platform/workflow/ -run 'TestRule|TestWellFormed|TestTheFixture|TestADispute' -count=1
Expected: FAIL — the package does not compile.
- Step 3: Write
model.go
// Package workflow is the workflow engine (spec 5): per-organization,// per-kind, versioned definitions; gates and grants with amount floors;// grantee resolution over the org structure; delegation and escalation; and// the transition function that posts through internal/platform/posting in// the same transaction and writes the immutable decision log.//// It is platform, not a module: claims (Plan 5), advances (Plan 7) and// petty cash (Plan 8) call it, and a module cannot be imported downward.package workflow
import ( "regexp" "sort" "time"
"github.com/google/uuid"
"github.com/mcom/erp/internal/kernel/money" "github.com/mcom/erp/internal/platform/posting")
// KeyPattern is the one shape of a kind, a state key and an action key.// Strings, never integers (AGENTS.md): a key can be retired without// reserving its number forever.var KeyPattern = regexp.MustCompile(`^[a-z][a-z0-9_]{0,62}$`)
type Gate string
const ( GateReporter Gate = "reporter" // the submitter, and nobody else GateTarget Gate = "target" // grants on the destination state GateSource Gate = "source" // grants on the state being left GateAdvance Gate = "advance" // grants on the state this rejection replaces)
type GranteeKind string
const ( GranteeMember GranteeKind = "member" GranteeSuperior GranteeKind = "superior" GranteePosition GranteeKind = "position" GranteeUnitHead GranteeKind = "unit_head" GranteeLevelsUp GranteeKind = "levels_up")
type PostingOrder string
const ( AccrualFirst PostingOrder = "accrual_first" PayoutFirst PostingOrder = "payout_first")
type OnVacancy string
const ( VacancyEscalate OnVacancy = "escalate" VacancyRefuse OnVacancy = "refuse")
// DefaultMaxEscalationLevels is spec 5.4's default bound: unbounded// escalation routes an RM100 claim to the CEO.const DefaultMaxEscalationLevels = 2
// The validator's size cap (Ruling 4). Rule 3 enumerates structure over the// graph; the cap bounds its cost and no real approval flow comes near it.const ( MaxStates = 64 MaxEdges = 256 MaxGrants = 256)
// Labels is locale -> text. "en" is required; every key must be a// supported locale. Stored as translation rows on publish (spec 20.1.6).type Labels map[string]string
type StateSpec struct { Key string `json:"key"` Semantic posting.Semantic `json:"semantic"` ControlKind string `json:"control_kind,omitempty"` DisplayOrder int `json:"display_order"` IsTerminal bool `json:"is_terminal"` EscalateAfterDays *int `json:"escalate_after_days,omitempty"` Labels Labels `json:"labels"`}
type EdgeSpec struct { From string `json:"from"` To string `json:"to"` Gate Gate `json:"gate"` ActionKey string `json:"action_key"` RequiresReason bool `json:"requires_reason"` ReversesAccrual bool `json:"reverses_accrual"` DisplayOrder int `json:"display_order"` Labels Labels `json:"labels"`}
// GrantSpec is one (grantee, step, floor) row (spec 5.4). The threshold is// HERE and nowhere else. MinAmount nil means every amount.type GrantSpec struct { StateKey string `json:"state_key,omitempty"` // "" iff AllStates AllStates bool `json:"all_states"` GranteeKind GranteeKind `json:"grantee_kind"` MemberID *uuid.UUID `json:"member_id,omitempty"` PositionID *uuid.UUID `json:"position_id,omitempty"` LevelsUp *int `json:"levels_up,omitempty"` MinAmount *money.Amount `json:"min_amount,omitempty"` MinAmountCurrency string `json:"min_amount_currency,omitempty"` OnVacancy OnVacancy `json:"on_vacancy,omitempty"` MaxEscalationLevels int `json:"max_escalation_levels,omitempty"`}
// Floor is the grant's floor; nil MinAmount is zero.func (g GrantSpec) Floor() money.Amount { if g.MinAmount == nil { return money.Zero() } return *g.MinAmount}
type Graph struct { Kind string `json:"kind"` PostingOrder PostingOrder `json:"posting_order"` States []StateSpec `json:"states"` Edges []EdgeSpec `json:"edges"` Grants []GrantSpec `json:"grants"`}
// Normalized returns a copy with the grant defaults filled in, so that a// graph saved without them and the same graph saved with them are equal.func (g Graph) Normalized() Graph { out := g out.Grants = make([]GrantSpec, len(g.Grants)) for i, gr := range g.Grants { if gr.OnVacancy == "" { gr.OnVacancy = VacancyEscalate } if gr.MaxEscalationLevels == 0 { gr.MaxEscalationLevels = DefaultMaxEscalationLevels } out.Grants[i] = gr } return out}
func (g Graph) State(key string) (StateSpec, bool) { for _, s := range g.States { if s.Key == key { return s, true } } return StateSpec{}, false}
func (g Graph) Edge(from, to string) (EdgeSpec, bool) { for _, e := range g.Edges { if e.From == from && e.To == to { return e, true } } return EdgeSpec{}, false}
// EdgesFrom returns the exits of a state in display order.func (g Graph) EdgesFrom(key string) []EdgeSpec { var out []EdgeSpec for _, e := range g.Edges { if e.From == key { out = append(out, e) } } sort.SliceStable(out, func(i, j int) bool { return out[i].DisplayOrder < out[j].DisplayOrder }) return out}
// GateState is the state whose grants authorize e (spec 5.3), and false for// a reporter edge, which no grant authorizes. For an advance edge it is the// destination of the single target-gated edge leaving the same state// (Ruling 6); false if there is not exactly one.func (g Graph) GateState(e EdgeSpec) (string, bool) { switch e.Gate { case GateTarget: return e.To, true case GateSource: return e.From, true case GateAdvance: var fwd []string for _, o := range g.Edges { if o.From == e.From && o.Gate == GateTarget { fwd = append(fwd, o.To) } } if len(fwd) == 1 { return fwd[0], true } } return "", false}
// ApprovalStates is every state entered through a target-gated edge — the// steps a submission must be routable through (Ruling 13) — in display order.func (g Graph) ApprovalStates() []string { seen := map[string]bool{} var out []StateSpec for _, e := range g.Edges { if e.Gate == GateTarget && !seen[e.To] { seen[e.To] = true if s, ok := g.State(e.To); ok { out = append(out, s) } } } sort.SliceStable(out, func(i, j int) bool { return out[i].DisplayOrder < out[j].DisplayOrder }) keys := make([]string, len(out)) for i, s := range out { keys[i] = s.Key } return keys}
// Grant is a published grant row.type Grant struct { ID uuid.UUID GrantSpec}
// Definition is one published version. Graph is embedded so a Definition// answers every Graph question; the id maps name the published rows.type Definition struct { ID uuid.UUID Version int PublishedAt time.Time PublishedBy *uuid.UUID Graph StateIDs map[string]uuid.UUID EdgeIDs map[[2]string]uuid.UUID GrantIDs []uuid.UUID // parallel to Graph.Grants}
// GrantsOn is the state's own grants plus every all_states grant (spec 5.4:// all_states carries authority at every step).func (d Definition) GrantsOn(stateKey string) []Grant { var out []Grant for i, g := range d.Grants { if g.AllStates || g.StateKey == stateKey { var id uuid.UUID if i < len(d.GrantIDs) { id = d.GrantIDs[i] } out = append(out, Grant{ID: id, GrantSpec: g}) } } return out}
// Violation is one reason a graph cannot be published. Rule 0 is// well-formedness; 1–7 are spec 5.7's rules. Code is stable for the editor// to switch on; Ref names the state key, "from->to", or "grants[i]".type Violation struct { Rule int `json:"rule"` Code string `json:"code"` Detail string `json:"detail"` Ref string `json:"ref,omitempty"`}- Step 4: Write
validate.go
package workflow
import ( "fmt" "slices" "strings"
"github.com/mcom/erp/internal/kernel/locale" "github.com/mcom/erp/internal/kernel/vocab" "github.com/mcom/erp/internal/platform/posting")
var semantics = map[posting.Semantic]bool{ posting.Draft: true, posting.Pending: true, posting.Accrual: true, posting.Payout: true, posting.Acknowledged: true, posting.Dispute: true, posting.Rejected: true, posting.Cancelled: true, posting.Custom: true,}
var gates = map[Gate]bool{GateReporter: true, GateTarget: true, GateSource: true, GateAdvance: true}
func reverses(s posting.Semantic) bool { return s == posting.Rejected || s == posting.Cancelled }
type checker struct { g Graph idx map[string]int adj [][]int out []Violation root int}
func (c *checker) add(rule int, code, ref, format string, args ...any) { c.out = append(c.out, Violation{Rule: rule, Code: code, Ref: ref, Detail: fmt.Sprintf(format, args...)})}
// ValidateGraph is spec 5.7, pure. It runs live in the editor on every save// and again at publish; it touches no database. The DB-aware half (do the// member and position grantees exist and are they active) is Validate in// store.go, which calls this first.//// Rule 0 (well-formedness) failures stop the graph rules: reachability and// path rules over a graph whose edges name missing states would report// nonsense on top of the real defect.func ValidateGraph(g Graph, baseCurrency string) []Violation { g = g.Normalized() c := &checker{g: g, idx: map[string]int{}} c.wellFormed(baseCurrency) if len(c.out) > 0 { return c.out } c.adj = make([][]int, len(g.States)) for _, e := range g.Edges { c.adj[c.idx[e.From]] = append(c.adj[c.idx[e.From]], c.idx[e.To]) } if !c.rule1() { return c.out } c.rule2() c.rule3() c.rule5() c.rules4and6and7() return c.out}
func (c *checker) labels(l Labels, ref string) { if strings.TrimSpace(l["en"]) == "" { c.add(0, "label_en_missing", ref, "an English label is required") } for loc, v := range l { if !slices.Contains(locale.Supported, loc) { c.add(0, "label_locale_unsupported", ref, "locale %q is not supported", loc) } if len(v) > 200 { c.add(0, "label_too_long", ref, "labels are at most 200 bytes") } }}
func (c *checker) wellFormed(base string) { g := c.g if !KeyPattern.MatchString(g.Kind) { c.add(0, "kind_invalid", g.Kind, "kind must match %s", KeyPattern) } if g.PostingOrder != AccrualFirst && g.PostingOrder != PayoutFirst { c.add(0, "posting_order_unknown", string(g.PostingOrder), "posting_order must be accrual_first or payout_first") } if len(g.States) > MaxStates || len(g.Edges) > MaxEdges || len(g.Grants) > MaxGrants { c.add(0, "too_large", "", "at most %d states, %d edges and %d grants", MaxStates, MaxEdges, MaxGrants) return } for i, s := range g.States { switch _, dup := c.idx[s.Key]; { case !KeyPattern.MatchString(s.Key): c.add(0, "state_key_invalid", s.Key, "state keys must match %s", KeyPattern) case dup: c.add(0, "state_key_duplicate", s.Key, "state key %q is used twice", s.Key) default: c.idx[s.Key] = i // position in g.States; the graph rules index by it } if !semantics[s.Semantic] { c.add(0, "semantic_unknown", s.Key, "semantic %q is not one of the system semantics", s.Semantic) } switch { case s.Semantic == posting.Accrual && s.ControlKind == "": c.add(0, "control_kind_required", s.Key, "an accrual state names the control account it credits") case s.Semantic == posting.Accrual && !slices.Contains(vocab.ControlKinds, s.ControlKind): c.add(0, "control_kind_unknown", s.Key, "control kind %q is not known", s.ControlKind) case s.Semantic != posting.Accrual && s.ControlKind != "": c.add(0, "control_kind_forbidden", s.Key, "only an accrual state names a control account") } if d := s.EscalateAfterDays; d != nil && (*d < 1 || *d > 365) { c.add(0, "escalate_after_invalid", s.Key, "escalate_after_days must be 1..365") } c.labels(s.Labels, s.Key) } seen := map[[2]string]bool{} for _, e := range g.Edges { ref := e.From + "->" + e.To fi, okF := c.idx[e.From] ti, okT := c.idx[e.To] if !okF || !okT { c.add(0, "edge_state_unknown", ref, "an edge names a state that does not exist") continue } if e.From == e.To { c.add(0, "edge_self_loop", ref, "an edge cannot leave and enter the same state") } if seen[[2]string{e.From, e.To}] { c.add(0, "edge_duplicate", ref, "two edges join the same pair of states") } seen[[2]string{e.From, e.To}] = true if !gates[e.Gate] { c.add(0, "gate_unknown", ref, "gate %q is not reporter, target, source or advance", e.Gate) } if !KeyPattern.MatchString(e.ActionKey) { c.add(0, "action_key_invalid", ref, "action keys must match %s", KeyPattern) } to, from := g.States[ti], g.States[fi] if (to.Semantic == posting.Dispute || reverses(to.Semantic)) && !e.RequiresReason { c.add(0, "reason_required", ref, "a dispute, rejection or cancellation must require a reason") } if from.IsTerminal && to.Semantic != posting.Dispute { c.add(0, "terminal_exit_not_dispute", ref, "a terminal state may only be left by raising a dispute") } c.labels(e.Labels, ref) }
for i, gr := range g.Grants { ref := fmt.Sprintf("grants[%d]", i) if gr.AllStates != (gr.StateKey == "") { c.add(0, "grant_scope_invalid", ref, "a grant names a state, or is all_states, not both") } else if !gr.AllStates && !c.hasState(gr.StateKey) { c.add(0, "grant_state_unknown", ref, "state %q does not exist", gr.StateKey) } refs := map[GranteeKind][3]bool{ // member, position, levels_up GranteeMember: {true, false, false}, GranteePosition: {false, true, false}, GranteeLevelsUp: {false, false, true}, GranteeSuperior: {}, GranteeUnitHead: {}, } want, known := refs[gr.GranteeKind] if !known { c.add(0, "grantee_kind_unknown", ref, "grantee kind %q is not known", gr.GranteeKind) } else if (gr.MemberID != nil) != want[0] || (gr.PositionID != nil) != want[1] || (gr.LevelsUp != nil) != want[2] { c.add(0, "grantee_ref_invalid", ref, "a %s grant names exactly the reference its kind needs", gr.GranteeKind) } else if gr.LevelsUp != nil && (*gr.LevelsUp < 1 || *gr.LevelsUp > 9) { c.add(0, "levels_up_invalid", ref, "levels_up must be 1..9") } switch { case gr.MinAmount == nil && gr.MinAmountCurrency != "": c.add(0, "currency_without_amount", ref, "a currency is given with no amount") case gr.MinAmount != nil && gr.MinAmount.IsNegative(): c.add(0, "amount_negative", ref, "a floor cannot be negative") case gr.MinAmount != nil && gr.MinAmountCurrency != base: c.add(0, "currency_mismatch", ref, "floors are in the base currency %s until FX rates exist", base) } if gr.OnVacancy != VacancyEscalate && gr.OnVacancy != VacancyRefuse { c.add(0, "on_vacancy_unknown", ref, "on_vacancy must be escalate or refuse") } if gr.MaxEscalationLevels < 1 || gr.MaxEscalationLevels > 10 { c.add(0, "escalation_bound_invalid", ref, "max_escalation_levels must be 1..10") } }}
func (c *checker) hasState(key string) bool { _, ok := c.idx[key] return ok}
func (c *checker) is(i int, sem posting.Semantic) bool { return c.g.States[i].Semantic == sem }
// reach marks every node reachable from starts without entering a blocked// node. A blocked start is not entered.func (c *checker) reach(starts []int, blocked func(int) bool) []bool { seen := make([]bool, len(c.g.States)) stack := []int{} for _, s := range starts { if !blocked(s) && !seen[s] { seen[s] = true stack = append(stack, s) } } for len(stack) > 0 { n := stack[len(stack)-1] stack = stack[:len(stack)-1] for _, m := range c.adj[n] { if !seen[m] && !blocked(m) { seen[m] = true stack = append(stack, m) } } } return seen}
func none(int) bool { return false }
// Rule 1: exactly one draft state; at least one terminal. Returns false on// any rule-1 failure: rules 2–7 need a single root and a notion of where a// path ends, and reporting cycles in a graph with no terminal would bury// the real defect.func (c *checker) rule1() bool { var drafts, terminals int for i, s := range c.g.States { if s.Semantic == posting.Draft { drafts++ c.root = i } if s.IsTerminal { terminals++ } } if drafts != 1 { c.add(1, "draft_count", "", "exactly one draft state is required; found %d", drafts) } if terminals == 0 { c.add(1, "no_terminal", "", "at least one state must be terminal") } return drafts == 1 && terminals > 0}
// Rule 2: every state reachable from draft; every non-terminal has an exit.func (c *checker) rule2() { seen := c.reach([]int{c.root}, none) for i, s := range c.g.States { if !seen[i] { c.add(2, "unreachable", s.Key, "%s cannot be reached from the draft state", s.Key) } if !s.IsTerminal && len(c.adj[i]) == 0 { c.add(2, "no_exit", s.Key, "%s has no exit", s.Key) } }}
// Rule 3, as Ruling 4 reads it. Reversing states (rejected, cancelled) are// removed from the graph: passing through one undoes what was posted, so it// resets the count. On what remains:// (a) no cycle may contain an accrual or payout state — it would post again;// (b) from draft and from every exit of a reversing state, no path crosses// two accruals or two payouts;// (c) the declared order holds.func (c *checker) rule3() { n := len(c.g.States) rev := func(i int) bool { return reverses(c.g.States[i].Semantic) }
// (a) Tarjan's SCC over the non-reversing subgraph. comp, size := c.sccs(rev) cyclic := false for i := 0; i < n; i++ { if rev(i) || size[comp[i]] < 2 { continue } if c.is(i, posting.Accrual) { c.add(3, "accrual_on_cycle", c.g.States[i].Key, "%s can be entered twice without a reversal", c.g.States[i].Key) cyclic = true } if c.is(i, posting.Payout) { c.add(3, "payout_on_cycle", c.g.States[i].Key, "%s can be entered twice without a reversal", c.g.States[i].Key) cyclic = true } } if cyclic { return // (b) and (c) assume the posting states are acyclic }
starts := []int{c.root} for i := 0; i < n; i++ { if rev(i) { for _, m := range c.adj[i] { if !rev(m) { starts = append(starts, m) } } } }
// (b) Longest count of accrual (payout) states on any path, memoized. // Posting states are not on cycles (checked above), so a node whose // count is being computed cannot be re-entered through a posting state; // the guard map breaks the remaining (non-posting) cycles. for _, sem := range []posting.Semantic{posting.Accrual, posting.Payout} { memo := map[int]int{} busy := map[int]bool{} var most func(int) int most = func(i int) int { if v, ok := memo[i]; ok { return v } if busy[i] { return 0 } busy[i] = true best := 0 for _, m := range c.adj[i] { if !rev(m) { best = max(best, most(m)) } } busy[i] = false if c.is(i, sem) { best++ } memo[i] = best return best } for _, s := range starts { if most(s) > 1 { c.add(3, string(sem)+"_repeated", c.g.States[s].Key, "a path from %s crosses more than one %s state", c.g.States[s].Key, sem) break } } }
// (c) Ordering. isA := func(i int) bool { return c.is(i, posting.Accrual) } for p := 0; p < n; p++ { if !c.is(p, posting.Payout) { continue } key := c.g.States[p].Key switch c.g.PostingOrder { case AccrualFirst: if c.reach(starts, func(i int) bool { return rev(i) || isA(i) })[p] { c.add(3, "payout_before_accrual", key, "a path reaches %s without crossing an accrual first", key) } case PayoutFirst: var accruals []int for i := 0; i < n; i++ { if isA(i) { accruals = append(accruals, c.adj[i]...) } } if c.reach(accruals, rev)[p] { c.add(3, "accrual_before_payout", key, "the flow is payout_first but an accrual precedes %s", key) } after := c.reach(c.adj[p], func(i int) bool { return rev(i) || isA(i) }) for i, s := range c.g.States { if s.IsTerminal && (after[i] || i == p) { c.add(3, "payout_without_accrual", key, "a path from %s reaches %s without an accrual", key, s.Key) break } } } }}
func (c *checker) sccs(skip func(int) bool) (comp []int, size map[int]int) { n := len(c.g.States) comp = make([]int, n) for i := range comp { comp[i] = -1 } size = map[int]int{} index, low := make([]int, n), make([]int, n) on := make([]bool, n) for i := range index { index[i] = -1 } var stack []int next, ncomp := 0, 0 var visit func(int) visit = func(v int) { index[v], low[v] = next, next next++ stack = append(stack, v) on[v] = true for _, w := range c.adj[v] { if skip(w) { continue } if index[w] < 0 { visit(w) low[v] = min(low[v], low[w]) } else if on[w] { low[v] = min(low[v], index[w]) } } if low[v] == index[v] { for { w := stack[len(stack)-1] stack = stack[:len(stack)-1] on[w] = false comp[w] = ncomp size[ncomp]++ if w == v { break } } ncomp++ } } for v := 0; v < n; v++ { if !skip(v) && index[v] < 0 { visit(v) } } return comp, size}
// Rule 5: every rejection/cancellation edge leaving a state at or past an// accrual carries the reversal flag. "At or past" is reachability from an// accrual state without passing through a reversing state (which already// reversed it).func (c *checker) rule5() { var accruals []int for i := range c.g.States { if c.is(i, posting.Accrual) { accruals = append(accruals, i) } } past := c.reach(accruals, func(i int) bool { return reverses(c.g.States[i].Semantic) }) for _, e := range c.g.Edges { to := c.g.States[c.idx[e.To]] if past[c.idx[e.From]] && reverses(to.Semantic) && !e.ReversesAccrual { c.add(5, "reversal_missing", e.From+"->"+e.To, "%s is at or past an accrual, so rejecting it must reverse the accrual", e.From) } }}
// Rules 4, 6 and 7, which all look at the grants behind gated edges.func (c *checker) rules4and6and7() { d := Definition{Graph: c.g} checked := map[string]bool{} for _, e := range c.g.Edges { ref := e.From + "->" + e.To if e.Gate == GateAdvance { fwd := 0 for _, o := range c.g.Edges { if o.From == e.From && o.Gate == GateTarget { fwd++ } } if fwd != 1 { c.add(6, "advance_forward_count", ref, "an advance-gated edge needs exactly one target-gated edge leaving %s; found %d", e.From, fwd) continue } } state, gated := c.g.GateState(e) if !gated { continue } grants := d.GrantsOn(state) if len(grants) == 0 { c.add(4, "gate_without_grant", ref, "nobody is granted %s, which authorizes this edge", state) continue } if checked[state] { continue } checked[state] = true reachesZero := false for _, g := range grants { if g.MinAmount == nil || g.MinAmount.IsZero() { reachesZero = true } } if !reachesZero { c.add(7, "bands_do_not_reach_zero", state, "the grants on %s start above zero, so a small amount could be approved by nobody", state) } }}- Step 4b: Write
sqlhelp.go
Two helpers every DB-touching file in this package uses. They live here, in the pure task, so that Tasks 3, 4 and 5 — which run in parallel — do not each define them.
package workflow
import ( "context" "fmt"
"github.com/google/uuid" "github.com/jackc/pgx/v5"
"github.com/mcom/erp/internal/kernel/tenancy")
func orgOf(ctx context.Context) (uuid.UUID, error) { id, ok := tenancy.OrgFromContext(ctx) if !ok { return uuid.Nil, fmt.Errorf("workflow: %w", tenancy.ErrNoOrg) } return id, nil}
// idSet runs a query of the shape `SELECT id FROM t WHERE org_id = $1 AND// id = ANY($2) …` and returns the ids it found. Callers pass a literal SQL// string that carries its own org_id predicate.func idSet(ctx context.Context, tx pgx.Tx, sql string, org uuid.UUID, ids []uuid.UUID) (map[uuid.UUID]bool, error) { out := map[uuid.UUID]bool{} if len(ids) == 0 { return out, nil } rows, err := tx.Query(ctx, sql, org, ids) if err != nil { return nil, fmt.Errorf("workflow: %w", err) } defer rows.Close() for rows.Next() { var id uuid.UUID if err := rows.Scan(&id); err != nil { return nil, err } out[id] = true } return out, rows.Err()}internal/kernel/layering’s tenant-predicate guard scans every string literal in a platform package, so the SQL each caller passes to idSet is checked where it is written; each such literal must carry org_id = $1.
- Step 5: Run the tests
Run: go test ./internal/platform/workflow/ -run 'TestRule|TestWellFormed|TestTheFixture|TestADispute' -count=1 -v
Expected: PASS. A duplicate state key keeps its first position in c.idx; the edges naming it still resolve, which is why state_key_duplicate does not cascade into edge_state_unknown for the first holder of the key.
- Step 6: Watch the rules fail once
Comment out the call to c.rule5() in ValidateGraph; confirm TestRule5RejectingPastAccrualMustReverse goes RED. Change the rev(i) || in the AccrualFirst branch’s blocked function to false ||; confirm the ordering test still passes (the block is on accruals) — then remove isA(i) from it and confirm TestRule3OrderingIsDeclaredNotAssumed goes RED. Restore.
- Step 7: Commit
git add internal/platform/workflow/model.go internal/platform/workflow/validate.go internal/platform/workflow/sqlhelp.go internal/platform/workflow/validate_test.gogit commit -m "Add the workflow graph model and the publish-time validator
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>"Task 3: The definition store — drafts, publish, load
Section titled “Task 3: The definition store — drafts, publish, load”Files:
- Create:
internal/platform/workflow/store.go - Test:
internal/platform/workflow/store_test.go(test helpers in this file are prefixedst)
Interfaces:
-
Consumes: Task 1’s tables and
workflowtest; Task 2’sGraph,Definition,Violation,ValidateGraph;company.Get;audit.Write. -
Produces:
var ErrNoDraft, ErrNoDefinition, ErrRevisionConflict, ErrVersionConflict errortype InvalidError struct{ Violations []Violation } // *InvalidError; Error() lists codestype Draft struct { ID uuid.UUID; Kind string; Revision int; BaseVersion *int; Graph Graph; UpdatedAt time.Time; UpdatedBy uuid.UUID }type KindSummary struct { Kind string; LatestVersion int; DraftRevision int } // 0 = nonefunc Validate(ctx context.Context, tx pgx.Tx, g Graph) ([]Violation, error)func SaveDraft(ctx context.Context, tx pgx.Tx, c clock.Clock, kind string, expectedRevision int, g Graph, actor uuid.UUID) (Draft, []Violation, error)func GetDraft(ctx context.Context, tx pgx.Tx, kind string) (Draft, error)func DiscardDraft(ctx context.Context, tx pgx.Tx, c clock.Clock, kind string, expectedRevision int) errorfunc PublishGraph(ctx context.Context, tx pgx.Tx, c clock.Clock, g Graph, actor *uuid.UUID) (Definition, error)func Publish(ctx context.Context, tx pgx.Tx, c clock.Clock, kind string, expectedRevision int, actor uuid.UUID) (Definition, error)func Load(ctx context.Context, tx pgx.Tx, id uuid.UUID) (Definition, error)func Latest(ctx context.Context, tx pgx.Tx, kind string) (Definition, error)func LoadVersion(ctx context.Context, tx pgx.Tx, kind string, version int) (Definition, error)func ListKinds(ctx context.Context, tx pgx.Tx) ([]KindSummary, error)Uses Task 2’s
orgOfandidSet(sqlhelp.go).expectedRevision == 0means “there is no draft yet”. Audit entity kinds:workflow_draft(actionssave,discard),workflow_definition(publish). -
Step 1: Write the failing tests
internal/platform/workflow/store_test.go:
package workflow_test
import ( "cmp" "context" "slices" "testing"
"github.com/jackc/pgx/v5" "github.com/stretchr/testify/require"
"github.com/mcom/erp/internal/kernel/money" "github.com/mcom/erp/internal/platform/workflow" "github.com/mcom/erp/internal/platform/workflow/workflowtest")
func stPublish(t *testing.T, w *workflowtest.World, g workflow.Graph) workflow.Definition { t.Helper() var d workflow.Definition w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var err error d, err = workflow.PublishGraph(ctx, tx, w.Clock, g, nil) return err }) return d}
func stAuditCount(t *testing.T, w *workflowtest.World, kind string) int { t.Helper() var n int require.NoError(t, w.Admin.QueryRow(context.Background(), `SELECT count(*) FROM audit_log WHERE org_id = $1 AND entity_kind = $2`, w.Org, kind).Scan(&n)) return n}
func TestPublishRoundTripsTheGraphAndNumbersVersions(t *testing.T) { w := workflowtest.NewWorld(t) g := valid() g.States[0].Labels = workflow.Labels{"en": "Draft", "zh": "草稿"} floor := money.MustParse("2000") g.Grants = append(g.Grants, workflow.GrantSpec{StateKey: "approved", GranteeKind: workflow.GranteeUnitHead, MinAmount: &floor, MinAmountCurrency: "MYR"})
v1 := stPublish(t, w, g) v2 := stPublish(t, w, g) require.Equal(t, 1, v1.Version) require.Equal(t, 2, v2.Version) require.Equal(t, 2, stAuditCount(t, w, "workflow_definition"), "one audit row per publish")
var got workflow.Definition w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var err error got, err = workflow.Latest(ctx, tx, "expense_claim") return err }) require.Equal(t, v2.ID, got.ID) want, have := stSorted(g.Normalized()), stSorted(got.Graph) require.Equal(t, want.States, have.States, "states must load back exactly as published") require.Equal(t, want.Edges, have.Edges, "edges must load back exactly as published") require.ElementsMatch(t, want.Grants, have.Grants, "grants must load back exactly as published") draft, _ := have.State("draft") require.Equal(t, "草稿", draft.Labels["zh"]) var floors []string for _, gr := range have.Grants { if gr.MinAmount != nil { floors = append(floors, gr.MinAmount.String()) } } require.Equal(t, []string{"2000.0000"}, floors)}
// stSorted puts states and edges in the order Load returns them —// (display_order, key) and (display_order, from, to). Grants load in id// order, which is random, so callers compare them with ElementsMatch.func stSorted(g workflow.Graph) workflow.Graph { g.States = slices.Clone(g.States) g.Edges = slices.Clone(g.Edges) slices.SortStableFunc(g.States, func(a, b workflow.StateSpec) int { return cmp.Or(cmp.Compare(a.DisplayOrder, b.DisplayOrder), cmp.Compare(a.Key, b.Key)) }) slices.SortStableFunc(g.Edges, func(a, b workflow.EdgeSpec) int { return cmp.Or(cmp.Compare(a.DisplayOrder, b.DisplayOrder), cmp.Compare(a.From, b.From), cmp.Compare(a.To, b.To)) }) return g}
func TestAnInvalidGraphIsNotPublished(t *testing.T) { w := workflowtest.NewWorld(t) stPublish(t, w, valid()) // PRESENCE: a valid graph publishes
g := valid() g.Grants = nil err := w.InTx(func(ctx context.Context, tx pgx.Tx) error { _, err := workflow.PublishGraph(ctx, tx, w.Clock, g, nil) return err }) var inv *workflow.InvalidError require.ErrorAs(t, err, &inv) require.NotEmpty(t, inv.Violations)}
func TestValidateRefusesAGranteeThatIsNotAnActiveMemberOrLiveSeat(t *testing.T) { w := workflowtest.NewWorld(t) m := w.Member("Approver") g := valid() g.Grants = append(g.Grants, workflow.GrantSpec{StateKey: "approved", GranteeKind: workflow.GranteeMember, MemberID: &m})
check := func() []workflow.Violation { var vs []workflow.Violation w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var err error vs, err = workflow.Validate(ctx, tx, g) return err }) return vs } require.Empty(t, check(), "PRESENCE: an active member is a valid grantee") w.DeactivateMember(m) require.Contains(t, codes(check()), "grantee_member_unknown")
other := w.SecondOrg().Member("Stranger") // exists, but in another tenant g.Grants[len(g.Grants)-1].MemberID = &other require.Contains(t, codes(check()), "grantee_member_unknown")}
func TestDraftsAreOptimisticAndPublishConsumesThem(t *testing.T) { w := workflowtest.NewWorld(t) admin := w.Member("Admin") g := valid()
var d workflow.Draft w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var err error var vs []workflow.Violation d, vs, err = workflow.SaveDraft(ctx, tx, w.Clock, "expense_claim", 0, g, admin) require.Empty(t, vs) return err }) require.Equal(t, 1, d.Revision)
// A stale revision is refused; the current one is accepted. err := w.InTx(func(ctx context.Context, tx pgx.Tx) error { _, _, err := workflow.SaveDraft(ctx, tx, w.Clock, "expense_claim", 0, g, admin) return err }) require.ErrorIs(t, err, workflow.ErrRevisionConflict)
broken := valid() broken.Grants = nil w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var vs []workflow.Violation var err error d, vs, err = workflow.SaveDraft(ctx, tx, w.Clock, "expense_claim", 1, broken, admin) require.NotEmpty(t, vs, "an invalid draft saves, with its violations — the editor shows them live") return err }) require.Equal(t, 2, d.Revision)
// Publishing an invalid draft is refused and leaves it in place. err = w.InTx(func(ctx context.Context, tx pgx.Tx) error { _, err := workflow.Publish(ctx, tx, w.Clock, "expense_claim", 2, admin) return err }) var inv *workflow.InvalidError require.ErrorAs(t, err, &inv)
w.MustTx(func(ctx context.Context, tx pgx.Tx) error { _, _, err := workflow.SaveDraft(ctx, tx, w.Clock, "expense_claim", 2, g, admin) return err }) var def workflow.Definition w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var err error def, err = workflow.Publish(ctx, tx, w.Clock, "expense_claim", 3, admin) return err }) require.Equal(t, 1, def.Version) require.Equal(t, admin, *def.PublishedBy) err = w.InTx(func(ctx context.Context, tx pgx.Tx) error { _, err := workflow.GetDraft(ctx, tx, "expense_claim") return err }) require.ErrorIs(t, err, workflow.ErrNoDraft) require.Equal(t, 3, stAuditCount(t, w, "workflow_draft"), "three saves")}
func TestPublishedRowsCannotBeChangedByTheApplication(t *testing.T) { w := workflowtest.NewWorld(t) d := stPublish(t, w, valid()) // PRESENCE: the row is visible to the application role. var n int w.MustTx(func(ctx context.Context, tx pgx.Tx) error { return tx.QueryRow(ctx, `SELECT count(*) FROM workflow_state WHERE org_id = $1 AND definition_id = $2`, w.Org, d.ID).Scan(&n) }) require.Positive(t, n) // ABSENCE: neither UPDATE nor DELETE is permitted to it. for _, stmt := range []string{ `UPDATE workflow_state SET display_order = 9 WHERE org_id = $1 AND definition_id = $2`, `DELETE FROM workflow_state WHERE org_id = $1 AND definition_id = $2`, } { err := w.InTx(func(ctx context.Context, tx pgx.Tx) error { _, err := tx.Exec(ctx, stmt, w.Org, d.ID) return err }) require.ErrorContains(t, err, "permission denied", stmt) }}
func TestDefinitionsAreInvisibleAcrossTenants(t *testing.T) { w := workflowtest.NewWorld(t) d := stPublish(t, w, valid()) w.MustTx(func(ctx context.Context, tx pgx.Tx) error { // PRESENCE _, err := workflow.Load(ctx, tx, d.ID) return err }) other := w.SecondOrg() err := other.InTx(func(ctx context.Context, tx pgx.Tx) error { _, err := workflow.Load(ctx, tx, d.ID) return err }) require.ErrorIs(t, err, workflow.ErrNoDefinition)}- Step 2: Run them and watch them fail
Run: go test ./internal/platform/workflow/ -run 'TestPublish|TestAnInvalid|TestValidateRefuses|TestDrafts|TestDefinitionsAre' -count=1
Expected: FAIL — PublishGraph, SaveDraft and friends are undefined.
- Step 3: Write
store.go
package workflow
import ( "context" "encoding/json" "errors" "fmt" "strings" "time"
"github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn"
"github.com/mcom/erp/internal/kernel/audit" "github.com/mcom/erp/internal/kernel/clock" "github.com/mcom/erp/internal/kernel/money" "github.com/mcom/erp/internal/platform/company" "github.com/mcom/erp/internal/platform/posting")
var ( ErrNoDraft = errors.New("workflow: no draft for this kind") ErrNoDefinition = errors.New("workflow: no published definition") ErrRevisionConflict = errors.New("workflow: the draft has changed since it was read") ErrVersionConflict = errors.New("workflow: another version of this kind was published concurrently"))
// InvalidError carries every reason a graph cannot be published.type InvalidError struct{ Violations []Violation }
func (e *InvalidError) Error() string { cs := make([]string, len(e.Violations)) for i, v := range e.Violations { cs[i] = v.Code } return "workflow: definition is invalid: " + strings.Join(cs, ", ")}
type Draft struct { ID uuid.UUID Kind string Revision int BaseVersion *int Graph Graph UpdatedAt time.Time UpdatedBy uuid.UUID}
type KindSummary struct { Kind string LatestVersion int // 0: never published DraftRevision int // 0: no draft}
// Validate is ValidateGraph plus the checks only the database can answer:// the base currency floors are compared against, and whether every member// and position a grant names is active in THIS organization. A grantee in// another tenant is reported exactly like a missing one.func Validate(ctx context.Context, tx pgx.Tx, g Graph) ([]Violation, error) { org, err := orgOf(ctx) if err != nil { return nil, err } p, err := company.Get(ctx, tx) if err != nil { return nil, fmt.Errorf("workflow: validate: %w", err) } vs := ValidateGraph(g, p.BaseCurrency)
var members, positions []uuid.UUID for _, gr := range g.Grants { if gr.MemberID != nil { members = append(members, *gr.MemberID) } if gr.PositionID != nil { positions = append(positions, *gr.PositionID) } } liveMembers, err := idSet(ctx, tx, `SELECT id FROM organization_member WHERE org_id = $1 AND id = ANY($2) AND active`, org, members) if err != nil { return nil, err } liveSeats, err := idSet(ctx, tx, `SELECT id FROM org_position WHERE org_id = $1 AND id = ANY($2) AND is_active`, org, positions) if err != nil { return nil, err } for i, gr := range g.Grants { ref := fmt.Sprintf("grants[%d]", i) if gr.MemberID != nil && !liveMembers[*gr.MemberID] { vs = append(vs, Violation{Rule: 0, Code: "grantee_member_unknown", Ref: ref, Detail: "the member is not an active member of this organization"}) } if gr.PositionID != nil && !liveSeats[*gr.PositionID] { vs = append(vs, Violation{Rule: 0, Code: "grantee_position_unknown", Ref: ref, Detail: "the position is not an active position of this organization"}) } } return vs, nil}
// SaveDraft replaces the kind's draft. expectedRevision is the revision the// caller edited (0: there is no draft yet). An invalid graph is saved and// its violations returned — only Publish refuses.func SaveDraft(ctx context.Context, tx pgx.Tx, c clock.Clock, kind string, expectedRevision int, g Graph, actor uuid.UUID) (Draft, []Violation, error) { org, err := orgOf(ctx) if err != nil { return Draft{}, nil, err } g.Kind = kind g = g.Normalized() doc, err := json.Marshal(g) if err != nil { return Draft{}, nil, fmt.Errorf("workflow: encode draft: %w", err) } now := c.Now()
var id uuid.UUID var rev int err = tx.QueryRow(ctx, `SELECT id, revision FROM workflow_draft WHERE org_id = $1 AND kind = $2 FOR UPDATE`, org, kind).Scan(&id, &rev) switch { case errors.Is(err, pgx.ErrNoRows): if expectedRevision != 0 { return Draft{}, nil, ErrRevisionConflict } var base *int if latest, err := latestVersion(ctx, tx, org, kind); err != nil { return Draft{}, nil, err } else if latest > 0 { base = &latest } rev = 1 err = tx.QueryRow(ctx, ` INSERT INTO workflow_draft (org_id, kind, revision, base_version, document, updated_at, updated_by_member_id) VALUES ($1, $2, 1, $3, $4, $5, $6) RETURNING id`, org, kind, base, doc, now, actor).Scan(&id) if isUnique(err) { return Draft{}, nil, ErrRevisionConflict } if err != nil { return Draft{}, nil, fmt.Errorf("workflow: insert draft: %w", err) } case err != nil: return Draft{}, nil, fmt.Errorf("workflow: read draft: %w", err) default: if expectedRevision != rev { return Draft{}, nil, ErrRevisionConflict } rev++ if _, err := tx.Exec(ctx, ` UPDATE workflow_draft SET revision = $3, document = $4, updated_at = $5, updated_by_member_id = $6 WHERE org_id = $1 AND id = $2`, org, id, rev, doc, now, actor); err != nil { return Draft{}, nil, fmt.Errorf("workflow: update draft: %w", err) } }
vs, err := Validate(ctx, tx, g) if err != nil { return Draft{}, nil, err } if err := audit.Write(ctx, tx, c, audit.Entry{EntityKind: "workflow_draft", EntityID: id, Action: "save", ActorID: &actor, After: map[string]any{"kind": kind, "revision": rev, "graph": g}}); err != nil { return Draft{}, nil, err } d, err := GetDraft(ctx, tx, kind) return d, vs, err}
func GetDraft(ctx context.Context, tx pgx.Tx, kind string) (Draft, error) { org, err := orgOf(ctx) if err != nil { return Draft{}, err } var d Draft var doc []byte err = tx.QueryRow(ctx, ` SELECT id, kind, revision, base_version, document, updated_at, updated_by_member_id FROM workflow_draft WHERE org_id = $1 AND kind = $2`, org, kind). Scan(&d.ID, &d.Kind, &d.Revision, &d.BaseVersion, &doc, &d.UpdatedAt, &d.UpdatedBy) if errors.Is(err, pgx.ErrNoRows) { return Draft{}, ErrNoDraft } if err != nil { return Draft{}, fmt.Errorf("workflow: read draft: %w", err) } if err := json.Unmarshal(doc, &d.Graph); err != nil { return Draft{}, fmt.Errorf("workflow: decode draft: %w", err) } return d, nil}
func DiscardDraft(ctx context.Context, tx pgx.Tx, c clock.Clock, kind string, expectedRevision int) error { org, err := orgOf(ctx) if err != nil { return err } id, err := lockDraft(ctx, tx, org, kind, expectedRevision) if err != nil { return err } if _, err := tx.Exec(ctx, `DELETE FROM workflow_draft WHERE org_id = $1 AND id = $2`, org, id); err != nil { return fmt.Errorf("workflow: discard draft: %w", err) } return audit.Write(ctx, tx, c, audit.Entry{EntityKind: "workflow_draft", EntityID: id, Action: "discard", Before: map[string]any{"kind": kind, "revision": expectedRevision}})}
func lockDraft(ctx context.Context, tx pgx.Tx, org uuid.UUID, kind string, expectedRevision int) (uuid.UUID, error) { var id uuid.UUID var rev int err := tx.QueryRow(ctx, `SELECT id, revision FROM workflow_draft WHERE org_id = $1 AND kind = $2 FOR UPDATE`, org, kind).Scan(&id, &rev) if errors.Is(err, pgx.ErrNoRows) { return uuid.Nil, ErrNoDraft } if err != nil { return uuid.Nil, fmt.Errorf("workflow: lock draft: %w", err) } if rev != expectedRevision { return uuid.Nil, ErrRevisionConflict } return id, nil}
// Publish turns the kind's draft into the next version and deletes the// draft, in the caller's transaction.func Publish(ctx context.Context, tx pgx.Tx, c clock.Clock, kind string, expectedRevision int, actor uuid.UUID) (Definition, error) { org, err := orgOf(ctx) if err != nil { return Definition{}, err } id, err := lockDraft(ctx, tx, org, kind, expectedRevision) if err != nil { return Definition{}, err } d, err := GetDraft(ctx, tx, kind) if err != nil { return Definition{}, err } def, err := PublishGraph(ctx, tx, c, d.Graph, &actor) if err != nil { return Definition{}, err } if _, err := tx.Exec(ctx, `DELETE FROM workflow_draft WHERE org_id = $1 AND id = $2`, org, id); err != nil { return Definition{}, fmt.Errorf("workflow: consume draft: %w", err) } return def, nil}
// PublishGraph validates g and inserts it as the kind's next version. It is// the one path into the published tables: Publish uses it for a draft, and// provisioning uses it for the seeded default (actor nil).func PublishGraph(ctx context.Context, tx pgx.Tx, c clock.Clock, g Graph, actor *uuid.UUID) (Definition, error) { org, err := orgOf(ctx) if err != nil { return Definition{}, err } g = g.Normalized() vs, err := Validate(ctx, tx, g) if err != nil { return Definition{}, err } if len(vs) > 0 { return Definition{}, &InvalidError{Violations: vs} } latest, err := latestVersion(ctx, tx, org, g.Kind) if err != nil { return Definition{}, err } def := Definition{Version: latest + 1, PublishedAt: c.Now(), PublishedBy: actor, Graph: g, StateIDs: map[string]uuid.UUID{}, EdgeIDs: map[[2]string]uuid.UUID{}, GrantIDs: make([]uuid.UUID, len(g.Grants))}
err = tx.QueryRow(ctx, ` INSERT INTO workflow_definition (org_id, kind, version, posting_order, published_at, published_by_member_id) VALUES ($1, $2, $3, $4, $5, $6) RETURNING id`, org, g.Kind, def.Version, string(g.PostingOrder), def.PublishedAt, actor).Scan(&def.ID) if isUnique(err) { return Definition{}, ErrVersionConflict } if err != nil { return Definition{}, fmt.Errorf("workflow: insert definition: %w", err) }
for _, s := range g.States { var ctrl *string if s.ControlKind != "" { ctrl = &s.ControlKind } var id uuid.UUID if err := tx.QueryRow(ctx, ` INSERT INTO workflow_state (org_id, definition_id, state_key, semantic, control_kind, display_order, is_terminal, escalate_after_days) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING id`, org, def.ID, s.Key, string(s.Semantic), ctrl, s.DisplayOrder, s.IsTerminal, s.EscalateAfterDays).Scan(&id); err != nil { return Definition{}, fmt.Errorf("workflow: insert state %s: %w", s.Key, err) } def.StateIDs[s.Key] = id if err := writeLabels(ctx, tx, org, "workflow_state", id, "label", s.Labels); err != nil { return Definition{}, err } } for _, e := range g.Edges { var id uuid.UUID if err := tx.QueryRow(ctx, ` INSERT INTO workflow_edge (org_id, definition_id, from_key, to_key, gate, action_key, requires_reason, reverses_accrual, display_order) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) RETURNING id`, org, def.ID, e.From, e.To, string(e.Gate), e.ActionKey, e.RequiresReason, e.ReversesAccrual, e.DisplayOrder).Scan(&id); err != nil { return Definition{}, fmt.Errorf("workflow: insert edge %s->%s: %w", e.From, e.To, err) } def.EdgeIDs[[2]string{e.From, e.To}] = id if err := writeLabels(ctx, tx, org, "workflow_edge", id, "action_label", e.Labels); err != nil { return Definition{}, err } } for i, gr := range g.Grants { var state, cur *string if !gr.AllStates { state = &gr.StateKey } if gr.MinAmountCurrency != "" { cur = &gr.MinAmountCurrency } // Base currency only (Ruling 8): the compared floor IS the input. if err := tx.QueryRow(ctx, ` INSERT INTO workflow_gate_grant (org_id, definition_id, state_key, all_states, grantee_kind, grantee_member_id, grantee_position_id, levels_up, min_amount, min_amount_input, min_amount_currency, on_vacancy, max_escalation_levels) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $9, $10, $11, $12) RETURNING id`, org, def.ID, state, gr.AllStates, string(gr.GranteeKind), gr.MemberID, gr.PositionID, gr.LevelsUp, amountArg(gr.MinAmount), cur, string(gr.OnVacancy), gr.MaxEscalationLevels).Scan(&def.GrantIDs[i]); err != nil { return Definition{}, fmt.Errorf("workflow: insert grant %d: %w", i, err) } }
if err := audit.Write(ctx, tx, c, audit.Entry{EntityKind: "workflow_definition", EntityID: def.ID, Action: "publish", ActorID: actor, After: map[string]any{"kind": g.Kind, "version": def.Version, "graph": g}}); err != nil { return Definition{}, err } return def, nil}
// amountArg binds an optional amount as NUMERIC text, or NULL.func amountArg(a *money.Amount) *string { if a == nil { return nil } s := a.String() return &s}
func parseAmount(s *string) (*money.Amount, error) { if s == nil { return nil, nil } a, err := money.Parse(*s) if err != nil { return nil, err } return &a, nil}
func isUnique(err error) bool { var pg *pgconn.PgError return errors.As(err, &pg) && pg.Code == "23505"}
func latestVersion(ctx context.Context, tx pgx.Tx, org uuid.UUID, kind string) (int, error) { var v int if err := tx.QueryRow(ctx, `SELECT COALESCE(max(version), 0) FROM workflow_definition WHERE org_id = $1 AND kind = $2`, org, kind).Scan(&v); err != nil { return 0, fmt.Errorf("workflow: latest version: %w", err) } return v, nil}
func writeLabels(ctx context.Context, tx pgx.Tx, org uuid.UUID, kind string, id uuid.UUID, field string, l Labels) error { for loc, v := range l { if _, err := tx.Exec(ctx, ` INSERT INTO translation (org_id, entity_kind, entity_id, field, locale, value) VALUES ($1, $2, $3, $4, $5, $6)`, org, kind, id, field, loc, v); err != nil { return fmt.Errorf("workflow: write label: %w", err) } } return nil}
// loadLabels reads every locale of a field for a batch of ids, for the// editor. (i18n.Load answers "in the request's locale" — a different// question, and the one a status chip asks.)func loadLabels(ctx context.Context, tx pgx.Tx, org uuid.UUID, kind, field string, ids []uuid.UUID) (map[uuid.UUID]Labels, error) { out := map[uuid.UUID]Labels{} rows, err := tx.Query(ctx, ` SELECT entity_id, locale, value FROM translation WHERE org_id = $1 AND entity_kind = $2 AND field = $3 AND entity_id = ANY($4)`, org, kind, field, ids) if err != nil { return nil, fmt.Errorf("workflow: load labels: %w", err) } defer rows.Close() for rows.Next() { var id uuid.UUID var loc, v string if err := rows.Scan(&id, &loc, &v); err != nil { return nil, err } if out[id] == nil { out[id] = Labels{} } out[id][loc] = v } return out, rows.Err()}
func Latest(ctx context.Context, tx pgx.Tx, kind string) (Definition, error) { return loadWhere(ctx, tx, `kind = $2 ORDER BY version DESC LIMIT 1`, kind)}
func LoadVersion(ctx context.Context, tx pgx.Tx, kind string, version int) (Definition, error) { return loadWhere(ctx, tx, `kind = $2 AND version = $3`, kind, version)}
func Load(ctx context.Context, tx pgx.Tx, id uuid.UUID) (Definition, error) { return loadWhere(ctx, tx, `id = $2`, id)}
// loadWhere reads one definition row selected by a fixed predicate (every// caller passes a literal) plus its states, edges, grants and labels.func loadWhere(ctx context.Context, tx pgx.Tx, where string, args ...any) (Definition, error) { org, err := orgOf(ctx) if err != nil { return Definition{}, err } var d Definition var order string err = tx.QueryRow(ctx, ` SELECT id, kind, version, posting_order, published_at, published_by_member_id FROM workflow_definition WHERE org_id = $1 AND `+where, append([]any{org}, args...)...). Scan(&d.ID, &d.Kind, &d.Version, &order, &d.PublishedAt, &d.PublishedBy) if errors.Is(err, pgx.ErrNoRows) { return Definition{}, ErrNoDefinition } if err != nil { return Definition{}, fmt.Errorf("workflow: load definition: %w", err) } d.PostingOrder = PostingOrder(order) d.StateIDs, d.EdgeIDs = map[string]uuid.UUID{}, map[[2]string]uuid.UUID{}
rows, err := tx.Query(ctx, ` SELECT id, state_key, semantic, COALESCE(control_kind, ''), display_order, is_terminal, escalate_after_days FROM workflow_state WHERE org_id = $1 AND definition_id = $2 ORDER BY display_order, state_key`, org, d.ID) if err != nil { return Definition{}, fmt.Errorf("workflow: load states: %w", err) } var stateIDs []uuid.UUID for rows.Next() { var id uuid.UUID var s StateSpec var sem string if err := rows.Scan(&id, &s.Key, &sem, &s.ControlKind, &s.DisplayOrder, &s.IsTerminal, &s.EscalateAfterDays); err != nil { rows.Close() return Definition{}, err } s.Semantic = posting.Semantic(sem) d.States = append(d.States, s) d.StateIDs[s.Key] = id stateIDs = append(stateIDs, id) } rows.Close() if err := rows.Err(); err != nil { return Definition{}, err }
rows, err = tx.Query(ctx, ` SELECT id, from_key, to_key, gate, action_key, requires_reason, reverses_accrual, display_order FROM workflow_edge WHERE org_id = $1 AND definition_id = $2 ORDER BY display_order, from_key, to_key`, org, d.ID) if err != nil { return Definition{}, fmt.Errorf("workflow: load edges: %w", err) } var edgeIDs []uuid.UUID for rows.Next() { var id uuid.UUID var e EdgeSpec var gate string if err := rows.Scan(&id, &e.From, &e.To, &gate, &e.ActionKey, &e.RequiresReason, &e.ReversesAccrual, &e.DisplayOrder); err != nil { rows.Close() return Definition{}, err } e.Gate = Gate(gate) d.Edges = append(d.Edges, e) d.EdgeIDs[[2]string{e.From, e.To}] = id edgeIDs = append(edgeIDs, id) } rows.Close() if err := rows.Err(); err != nil { return Definition{}, err }
rows, err = tx.Query(ctx, ` SELECT id, COALESCE(state_key, ''), all_states, grantee_kind, grantee_member_id, grantee_position_id, levels_up, min_amount_input::text, COALESCE(min_amount_currency, ''), on_vacancy, max_escalation_levels FROM workflow_gate_grant WHERE org_id = $1 AND definition_id = $2 ORDER BY id`, org, d.ID) if err != nil { return Definition{}, fmt.Errorf("workflow: load grants: %w", err) } for rows.Next() { var id uuid.UUID var g GrantSpec var kind, vac string var min *string if err := rows.Scan(&id, &g.StateKey, &g.AllStates, &kind, &g.MemberID, &g.PositionID, &g.LevelsUp, &min, &g.MinAmountCurrency, &vac, &g.MaxEscalationLevels); err != nil { rows.Close() return Definition{}, err } g.GranteeKind, g.OnVacancy = GranteeKind(kind), OnVacancy(vac) if g.MinAmount, err = parseAmount(min); err != nil { rows.Close() return Definition{}, err } d.Grants = append(d.Grants, g) d.GrantIDs = append(d.GrantIDs, id) } rows.Close() if err := rows.Err(); err != nil { return Definition{}, err }
sl, err := loadLabels(ctx, tx, org, "workflow_state", "label", stateIDs) if err != nil { return Definition{}, err } for i := range d.States { d.States[i].Labels = sl[d.StateIDs[d.States[i].Key]] } el, err := loadLabels(ctx, tx, org, "workflow_edge", "action_label", edgeIDs) if err != nil { return Definition{}, err } for i := range d.Edges { d.Edges[i].Labels = el[d.EdgeIDs[[2]string{d.Edges[i].From, d.Edges[i].To}]] } return d, nil}
func ListKinds(ctx context.Context, tx pgx.Tx) ([]KindSummary, error) { org, err := orgOf(ctx) if err != nil { return nil, err } byKind := map[string]*KindSummary{} var order []string get := func(k string) *KindSummary { if byKind[k] == nil { byKind[k] = &KindSummary{Kind: k} order = append(order, k) } return byKind[k] } rows, err := tx.Query(ctx, `SELECT kind, max(version) FROM workflow_definition WHERE org_id = $1 GROUP BY kind ORDER BY kind`, org) if err != nil { return nil, fmt.Errorf("workflow: list kinds: %w", err) } for rows.Next() { var k string var v int if err := rows.Scan(&k, &v); err != nil { rows.Close() return nil, err } get(k).LatestVersion = v } rows.Close() rows, err = tx.Query(ctx, `SELECT kind, revision FROM workflow_draft WHERE org_id = $1 ORDER BY kind`, org) if err != nil { return nil, fmt.Errorf("workflow: list drafts: %w", err) } for rows.Next() { var k string var r int if err := rows.Scan(&k, &r); err != nil { rows.Close() return nil, err } get(k).DraftRevision = r } rows.Close() out := make([]KindSummary, 0, len(order)) for _, k := range order { out = append(out, *byKind[k]) } return out, rows.Err()}Load must return ErrNoDefinition for another tenant’s id: RLS returns zero rows, and the explicit org_id = $1 returns zero rows even without RLS.
- Step 4: Run the tests
Run: go test ./internal/platform/workflow/ -count=1
Expected: PASS (including Task 2’s tests). Also run go test ./internal/kernel/layering/ -run TestEveryTenantQueryCarriesAnExplicitOrgPredicate -count=1 — expected PASS; every statement above names org_id = $1 or inserts org_id.
- Step 5: Watch the tenant predicate matter
In loadWhere’s first query, delete org_id = $1 AND (keeping the $1 bind by prepending $1::uuid IS NOT NULL AND ). Run the layering test: RED naming store.go. Restore.
- Step 6: Commit
git add internal/platform/workflow/store.go internal/platform/workflow/store_test.gogit commit -m "Add workflow drafts, publishing and definition loading
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>"Task 4: Grantee resolution — who a grant points at, for this claimant, on this day
Section titled “Task 4: Grantee resolution — who a grant points at, for this claimant, on this day”Files:
- Create:
internal/platform/workflow/resolve.go - Test:
internal/platform/workflow/resolve_test.go(helpers prefixedrs)
Interfaces:
-
Consumes: Task 2’s
Grant,GrantSpec, grantee kinds,OnVacancy;orgstruct.SuperiorOf,orgstruct.NextFilledAbove,orgstruct.HoldersOf,orgstruct.PrimaryPositionOf,orgstruct.GetPosition,orgstruct.GetUnit,*orgstruct.NoSuperiorError,orgstruct.ErrNoPrimaryPosition,orgstruct.ErrNotFound. -
Produces:
type Gap struct {GrantID uuid.UUID `json:"grant_id"`Reason string `json:"reason"`PositionID *uuid.UUID `json:"position_id,omitempty"`MemberID *uuid.UUID `json:"member_id,omitempty"`}// Gap reasons (strings, stable for the UI):const (GapMemberInactive = "member_inactive"GapPositionRetired = "position_retired"GapVacant = "vacant"GapNoPrimaryPosition = "no_primary_position"GapNoUnitHead = "no_unit_head"GapAboveRoot = "above_root"GapOnlyTheClaimant = "only_the_claimant")// superior grants use orgstruct.NoSuperiorReason values verbatim as Reason.type Resolved struct {Grant Grant `json:"-"`GrantID uuid.UUID `json:"grant_id"`Kind GranteeKind `json:"grantee_kind"`Floor string `json:"floor"` // money string, "0.0000" for nilMembers []uuid.UUID `json:"members"` // active members, sorted, deduplicatedEscalated bool `json:"escalated"` // members came from vacancy escalation or self-replacementGap *Gap `json:"gap,omitempty"`}func Resolve(ctx context.Context, tx pgx.Tx, grants []Grant, reporter uuid.UUID, on time.Time) ([]Resolved, error)Resolvedis what the decision log persists (its JSON isworkflow_decision.resolution). -
Step 1: Write the failing tests
internal/platform/workflow/resolve_test.go. One test per grantee kind and per edge case §5.4 names, each asserting on the exact member set and on Gap.Reason — not merely on “empty”.
package workflow_test
import ( "context" "testing"
"github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/stretchr/testify/require"
"github.com/mcom/erp/internal/platform/workflow" "github.com/mcom/erp/internal/platform/workflow/workflowtest")
func rsResolve(t *testing.T, w *workflowtest.World, reporter uuid.UUID, specs ...workflow.GrantSpec) []workflow.Resolved { t.Helper() grants := make([]workflow.Grant, len(specs)) for i, s := range specs { grants[i] = workflow.Grant{ID: uuid.New(), GrantSpec: s} } var out []workflow.Resolved w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var err error out, err = workflow.Resolve(ctx, tx, grants, reporter, workflowtest.Today) return err }) require.Len(t, out, len(specs)) return out}
func rsGrant(kind workflow.GranteeKind) workflow.GrantSpec { return workflow.GrantSpec{StateKey: "approved", GranteeKind: kind, OnVacancy: workflow.VacancyEscalate, MaxEscalationLevels: 2}}
// rsChart is CEO > HEAD > STAFF, one person in each, HEAD heads the unit.type rsChart struct { w *workflowtest.World ceo, head, staff uuid.UUID // members ceoSeat, headSeat, staffSeat uuid.UUID}
func rsNewChart(t *testing.T) rsChart { w := workflowtest.NewWorld(t) c := rsChart{w: w} c.ceoSeat = w.Seat("CEO", nil) c.headSeat = w.Seat("HEAD", &c.ceoSeat) c.staffSeat = w.Seat("STAFF", &c.headSeat) c.ceo, c.head, c.staff = w.Member("Ceo"), w.Member("Head"), w.Member("Staff") w.Sit(c.ceo, c.ceoSeat) w.Sit(c.head, c.headSeat) w.Sit(c.staff, c.staffSeat) w.SetUnitHead(&c.headSeat) return c}
func TestEachGranteeKindResolvesPerClaimant(t *testing.T) { c := rsNewChart(t) m := c.ceo pos := c.ceoSeat two := 2 sup := rsGrant(workflow.GranteeSuperior) mem := rsGrant(workflow.GranteeMember) mem.MemberID = &m seat := rsGrant(workflow.GranteePosition) seat.PositionID = &pos up := rsGrant(workflow.GranteeLevelsUp) up.LevelsUp = &two
got := rsResolve(t, c.w, c.staff, sup, mem, seat, rsGrant(workflow.GranteeUnitHead), up) require.Equal(t, []uuid.UUID{c.head}, got[0].Members, "superior of STAFF is HEAD") require.Equal(t, []uuid.UUID{c.ceo}, got[1].Members) require.Equal(t, []uuid.UUID{c.ceo}, got[2].Members) require.Equal(t, []uuid.UUID{c.head}, got[3].Members, "the unit's head seat") require.Equal(t, []uuid.UUID{c.ceo}, got[4].Members, "two levels up from STAFF") for _, r := range got { require.Nil(t, r.Gap) require.False(t, r.Escalated) }}
func TestAVacantSeatEscalatesWithinItsBoundOrRefuses(t *testing.T) { c := rsNewChart(t) vacant := c.w.Seat("VACANT", &c.ceoSeat) esc := rsGrant(workflow.GranteePosition) esc.PositionID = &vacant ref := esc ref.OnVacancy = workflow.VacancyRefuse
got := rsResolve(t, c.w, c.staff, esc, ref) require.Equal(t, []uuid.UUID{c.ceo}, got[0].Members, "escalate walks to the next filled seat above") require.True(t, got[0].Escalated) require.Empty(t, got[1].Members) require.Equal(t, workflow.GapVacant, got[1].Gap.Reason) require.Equal(t, vacant, *got[1].Gap.PositionID)}
func TestASuperiorThatCannotResolveIsAGapNamingTheMissingLink(t *testing.T) { c := rsNewChart(t) loner := c.w.Member("Loner") // no seat at all got := rsResolve(t, c.w, loner, rsGrant(workflow.GranteeSuperior)) require.Empty(t, got[0].Members) require.Equal(t, "no_primary_position", got[0].Gap.Reason)
// PRESENCE for the vacancy branch: STAFF's superior resolves... require.Equal(t, []uuid.UUID{c.head}, rsResolve(t, c.w, c.staff, rsGrant(workflow.GranteeSuperior))[0].Members)}
func TestTheSuperiorOfAVacantSeatEscalatesUpTheTree(t *testing.T) { w := workflowtest.NewWorld(t) ceoSeat := w.Seat("CEO", nil) gap := w.Seat("GAP", &ceoSeat) // vacant, and no unit head is set staffSeat := w.Seat("STAFF", &gap) ceo, staff := w.Member("Ceo"), w.Member("Staff") w.Sit(ceo, ceoSeat) w.Sit(staff, staffSeat)
esc := rsGrant(workflow.GranteeSuperior) got := rsResolve(t, w, staff, esc) require.Equal(t, []uuid.UUID{ceo}, got[0].Members) require.True(t, got[0].Escalated)
ref := esc ref.OnVacancy = workflow.VacancyRefuse got = rsResolve(t, w, staff, ref) require.Empty(t, got[0].Members) require.NotNil(t, got[0].Gap)}
func TestAGrantResolvingToTheClaimantEscalatesOneLevel(t *testing.T) { c := rsNewChart(t) // HEAD files a claim; the unit head is HEAD's own seat. got := rsResolve(t, c.w, c.head, rsGrant(workflow.GranteeUnitHead)) require.Equal(t, []uuid.UUID{c.ceo}, got[0].Members, "self is replaced by the next filled seat above") require.True(t, got[0].Escalated)
// A member grant naming the claimant is NOT escalated: it simply cannot // be used by them, and the self-approval refusal says so later. m := c.head mem := rsGrant(workflow.GranteeMember) mem.MemberID = &m require.Equal(t, []uuid.UUID{c.head}, rsResolve(t, c.w, c.head, mem)[0].Members)}
func TestInactiveMembersAndRetiredSeatsResolveToNobody(t *testing.T) { c := rsNewChart(t) m := c.ceo mem := rsGrant(workflow.GranteeMember) mem.MemberID = &m require.Equal(t, []uuid.UUID{c.ceo}, rsResolve(t, c.w, c.staff, mem)[0].Members) // PRESENCE c.w.DeactivateMember(c.ceo) got := rsResolve(t, c.w, c.staff, mem) require.Empty(t, got[0].Members) require.Equal(t, workflow.GapMemberInactive, got[0].Gap.Reason)}
func TestLevelsUpBeyondTheRootIsAGap(t *testing.T) { c := rsNewChart(t) five := 5 up := rsGrant(workflow.GranteeLevelsUp) up.LevelsUp = &five got := rsResolve(t, c.w, c.staff, up) require.Equal(t, workflow.GapAboveRoot, got[0].Gap.Reason)}- Step 2: Run them and watch them fail
Run: go test ./internal/platform/workflow/ -run 'TestEachGrantee|TestAVacant|TestASuperior|TestTheSuperior|TestAGrantResolving|TestInactive|TestLevelsUp' -count=1
Expected: FAIL — Resolve undefined.
- Step 3: Write
resolve.go
package workflow
import ( "context" "errors" "fmt" "slices" "time"
"github.com/google/uuid" "github.com/jackc/pgx/v5"
"github.com/mcom/erp/internal/platform/orgstruct")
const ( GapMemberInactive = "member_inactive" GapPositionRetired = "position_retired" GapVacant = "vacant" GapNoPrimaryPosition = "no_primary_position" GapNoUnitHead = "no_unit_head" GapAboveRoot = "above_root" GapOnlyTheClaimant = "only_the_claimant")
type Gap struct { GrantID uuid.UUID `json:"grant_id"` Reason string `json:"reason"` PositionID *uuid.UUID `json:"position_id,omitempty"` MemberID *uuid.UUID `json:"member_id,omitempty"`}
type Resolved struct { Grant Grant `json:"-"` GrantID uuid.UUID `json:"grant_id"` Kind GranteeKind `json:"grantee_kind"` Floor string `json:"floor"` Members []uuid.UUID `json:"members"` Escalated bool `json:"escalated"` Gap *Gap `json:"gap,omitempty"`}
// Resolve turns every grant into a concrete member set for this claimant on// this day (spec 5.4: "resolve every grant on the step to a concrete member// set → compute the winning floor over that resolved list → test the// actor's membership"). It is the FIRST of those three steps and nothing// else: floors are carried, not compared.//// Only member_position is dated (org-structure spec 5.5): the chart itself// is read as it stands now. That is why the caller persists the result at// decision time and never re-resolves to explain a past decision.func Resolve(ctx context.Context, tx pgx.Tx, grants []Grant, reporter uuid.UUID, on time.Time) ([]Resolved, error) { org, err := orgOf(ctx) if err != nil { return nil, err } out := make([]Resolved, len(grants)) for i, g := range grants { r := Resolved{Grant: g, GrantID: g.ID, Kind: g.GranteeKind, Floor: g.Floor().String()} members, escalated, gap, err := resolveOne(ctx, tx, g, reporter, on) if err != nil { return nil, fmt.Errorf("workflow: resolve grant %s: %w", g.ID, err) } members, err = activeOnly(ctx, tx, org, members) if err != nil { return nil, err } // A dynamic grant that lands on the claimant escalates one level // rather than blocking (spec 5.4). A member grant naming the // claimant does not: it names a person, not a relationship, and the // self-approval refusal is what applies to it. Nor does an // all_states grant: its holder keeps standing on their own claim's // posting transitions (Ruling 12), which Authorize decides. if g.GranteeKind != GranteeMember && !g.AllStates && slices.Contains(members, reporter) { members = slices.DeleteFunc(members, func(m uuid.UUID) bool { return m == reporter }) if len(members) == 0 { above, err := aboveClaimant(ctx, tx, g, reporter, on) if err != nil { return nil, err } if members, err = activeOnly(ctx, tx, org, above); err != nil { return nil, err } members = slices.DeleteFunc(members, func(m uuid.UUID) bool { return m == reporter }) escalated = true if len(members) == 0 { gap = &Gap{Reason: GapOnlyTheClaimant, MemberID: &reporter} } } } // Holders that all turned out to have left: name it, never return an // empty set without a reason. if gap == nil && len(members) == 0 { if g.GranteeKind == GranteeMember { gap = &Gap{Reason: GapMemberInactive, MemberID: g.MemberID} } else { gap = &Gap{Reason: GapVacant, PositionID: g.PositionID} } } if len(members) > 0 { gap = nil } if gap != nil { gap.GrantID = g.ID } slices.SortFunc(members, func(a, b uuid.UUID) int { return slices.Compare(a[:], b[:]) }) r.Members, r.Escalated, r.Gap = slices.Compact(members), escalated, gap if r.Members == nil { r.Members = []uuid.UUID{} } out[i] = r } return out, nil}
func resolveOne(ctx context.Context, tx pgx.Tx, g Grant, reporter uuid.UUID, on time.Time) ([]uuid.UUID, bool, *Gap, error) { switch g.GranteeKind { case GranteeMember: return []uuid.UUID{*g.MemberID}, false, nil, nil
case GranteePosition: return seat(ctx, tx, g, *g.PositionID, on)
case GranteeUnitHead: pos, gap, err := primarySeat(ctx, tx, reporter, on) if gap != nil || err != nil { return nil, false, gap, err } unit, err := orgstruct.GetUnit(ctx, tx, pos.OrgUnitID) if err != nil { return nil, false, nil, err } if unit.UnitHeadPositionID == nil { return nil, false, &Gap{Reason: GapNoUnitHead, PositionID: &pos.ID}, nil } return seat(ctx, tx, g, *unit.UnitHeadPositionID, on)
case GranteeLevelsUp: pos, gap, err := primarySeat(ctx, tx, reporter, on) if gap != nil || err != nil { return nil, false, gap, err } for i := 0; i < *g.LevelsUp; i++ { if pos.ReportsToPositionID == nil { return nil, false, &Gap{Reason: GapAboveRoot, PositionID: &pos.ID}, nil } if pos, err = orgstruct.GetPosition(ctx, tx, *pos.ReportsToPositionID); err != nil { return nil, false, nil, err } } return seat(ctx, tx, g, pos.ID, on)
case GranteeSuperior: members, err := orgstruct.SuperiorOf(ctx, tx, reporter, on) var nse *orgstruct.NoSuperiorError if errors.As(err, &nse) { // PositionID is set exactly when the claimant HAS a primary seat // (it is nil for no_primary_position and invalid_override, which // only an administrator can repair). With a seat, the remaining // reasons all mean the reporting seat above is empty or absent — // SuperiorOf reports no_unit_head when reports_to exists but is // vacant AND the unit has no head — so on_vacancy: escalate walks // up from the claimant's own seat. For no_reports_to the walk // finds nothing and the gap stands. if nse.PositionID != nil && g.OnVacancy == VacancyEscalate { up, err := orgstruct.NextFilledAbove(ctx, tx, *nse.PositionID, g.MaxEscalationLevels, on) if err != nil { return nil, false, nil, err } if len(up) > 0 { return up, true, nil, nil } } return nil, false, &Gap{Reason: string(nse.Reason), PositionID: nse.PositionID, MemberID: &reporter}, nil } return members, false, nil, err } return nil, false, nil, fmt.Errorf("unknown grantee kind %q", g.GranteeKind)}
// seat resolves a position grant or a derived seat: its current holders,// else — per the grant's on_vacancy — the next filled seat above it.func seat(ctx context.Context, tx pgx.Tx, g Grant, positionID uuid.UUID, on time.Time) ([]uuid.UUID, bool, *Gap, error) { pos, err := orgstruct.GetPosition(ctx, tx, positionID) if errors.Is(err, orgstruct.ErrNotFound) || (err == nil && !pos.IsActive) { return nil, false, &Gap{Reason: GapPositionRetired, PositionID: &positionID}, nil } if err != nil { return nil, false, nil, err } holders, err := orgstruct.HoldersOf(ctx, tx, positionID, on) if err != nil || len(holders) > 0 { return holders, false, nil, err } if g.OnVacancy == VacancyEscalate { up, err := orgstruct.NextFilledAbove(ctx, tx, positionID, g.MaxEscalationLevels, on) if err != nil || len(up) > 0 { return up, len(up) > 0, nil, err } } return nil, false, &Gap{Reason: GapVacant, PositionID: &positionID}, nil}
func primarySeat(ctx context.Context, tx pgx.Tx, member uuid.UUID, on time.Time) (orgstruct.Position, *Gap, error) { id, err := orgstruct.PrimaryPositionOf(ctx, tx, member, on) if errors.Is(err, orgstruct.ErrNoPrimaryPosition) { return orgstruct.Position{}, &Gap{Reason: GapNoPrimaryPosition, MemberID: &member}, nil } if err != nil { return orgstruct.Position{}, nil, err } pos, err := orgstruct.GetPosition(ctx, tx, id) return pos, nil, err}
// aboveClaimant is the "one level up" of spec 5.4's self case: the next// filled seat above the claimant's own primary seat, within the grant's// escalation bound.func aboveClaimant(ctx context.Context, tx pgx.Tx, g Grant, reporter uuid.UUID, on time.Time) ([]uuid.UUID, error) { id, err := orgstruct.PrimaryPositionOf(ctx, tx, reporter, on) if errors.Is(err, orgstruct.ErrNoPrimaryPosition) { return nil, nil } if err != nil { return nil, err } return orgstruct.NextFilledAbove(ctx, tx, id, g.MaxEscalationLevels, on)}
// activeOnly drops members who have left. A seat's holders are dated// assignments, and an assignment can outlive the membership it belonged to.func activeOnly(ctx context.Context, tx pgx.Tx, org uuid.UUID, ids []uuid.UUID) ([]uuid.UUID, error) { live, err := idSet(ctx, tx, `SELECT id FROM organization_member WHERE org_id = $1 AND id = ANY($2) AND active`, org, ids) if err != nil { return nil, err } return slices.DeleteFunc(slices.Clone(ids), func(id uuid.UUID) bool { return !live[id] }), nil}idSet and orgOf are Task 2’s sqlhelp.go.
- Step 4: Run the tests
Run: go test ./internal/platform/workflow/ -count=1
Expected: PASS.
- Step 5: Watch the self-replacement fail once
Delete the if g.GranteeKind != GranteeMember && slices.Contains(members, reporter) block; confirm TestAGrantResolvingToTheClaimantEscalatesOneLevel goes RED with [head] instead of [ceo]. Restore.
- Step 6: Commit
git add internal/platform/workflow/resolve.go internal/platform/workflow/resolve_test.gogit commit -m "Resolve workflow grants to members over the org structure
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>"Task 5: Delegation — authority lent, dated, one hop
Section titled “Task 5: Delegation — authority lent, dated, one hop”Files:
- Create:
internal/platform/workflow/delegation.go - Test:
internal/platform/workflow/delegation_test.go(helpers prefixeddl)
Interfaces:
-
Consumes: Task 1’s
workflow_delegation;audit.Write;clock.TodayUTC. -
Produces:
var ErrSelfDelegation, ErrDelegateInactive, ErrBackdated, ErrBadWindow, ErrNoDelegation, ErrAlreadyRevoked errortype Delegation struct {ID, DelegatorMemberID, DelegateMemberID uuid.UUIDKind *stringValidFrom time.TimeValidTo *time.TimeReason stringCreatedAt time.TimeRevokedAt *time.Time}type DelegationInput struct { Delegator, Delegate uuid.UUID; Kind *string; ValidFrom time.Time; ValidTo *time.Time; Reason string }func CreateDelegation(ctx context.Context, tx pgx.Tx, c clock.Clock, in DelegationInput) (Delegation, error)func RevokeDelegation(ctx context.Context, tx pgx.Tx, c clock.Clock, id, by uuid.UUID) errorfunc ListDelegations(ctx context.Context, tx pgx.Tx, member uuid.UUID) ([]Delegation, error) // as delegator or delegate, newest firstfunc ActiveDelegationsTo(ctx context.Context, tx pgx.Tx, delegate uuid.UUID, kind string, on time.Time) ([]Delegation, error) -
Step 1: Write the failing tests
package workflow_test
import ( "context" "testing" "time"
"github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/stretchr/testify/require"
"github.com/mcom/erp/internal/platform/workflow" "github.com/mcom/erp/internal/platform/workflow/workflowtest")
var dlToday = workflowtest.Today
func dlCreate(w *workflowtest.World, in workflow.DelegationInput) (workflow.Delegation, error) { var d workflow.Delegation err := w.InTx(func(ctx context.Context, tx pgx.Tx) error { var err error d, err = workflow.CreateDelegation(ctx, tx, w.Clock, in) return err }) return d, err}
func dlActive(t *testing.T, w *workflowtest.World, delegate uuid.UUID, kind string, on time.Time) []workflow.Delegation { t.Helper() var out []workflow.Delegation w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var err error out, err = workflow.ActiveDelegationsTo(ctx, tx, delegate, kind, on) return err }) return out}
func TestADelegationIsActiveOnlyInsideItsWindowAndScope(t *testing.T) { w := workflowtest.NewWorld(t) boss, deputy := w.Member("Boss"), w.Member("Deputy") end := dlToday.AddDate(0, 0, 7) claims := "expense_claim" d, err := dlCreate(w, workflow.DelegationInput{Delegator: boss, Delegate: deputy, Kind: &claims, ValidFrom: dlToday, ValidTo: &end, Reason: "annual leave"}) require.NoError(t, err)
require.Len(t, dlActive(t, w, deputy, "expense_claim", dlToday), 1, "PRESENCE: active today for its kind") require.Empty(t, dlActive(t, w, deputy, "advance", dlToday), "another kind") require.Empty(t, dlActive(t, w, deputy, "expense_claim", end), "valid_to is exclusive") require.Empty(t, dlActive(t, w, boss, "expense_claim", dlToday), "it lends to the deputy, not back")
w.MustTx(func(ctx context.Context, tx pgx.Tx) error { return workflow.RevokeDelegation(ctx, tx, w.Clock, d.ID, boss) }) require.Empty(t, dlActive(t, w, deputy, "expense_claim", dlToday), "revoked")}
func TestADelegationCannotBeBackdatedSelfOrToALeaver(t *testing.T) { w := workflowtest.NewWorld(t) boss, deputy := w.Member("Boss"), w.Member("Deputy") _, err := dlCreate(w, workflow.DelegationInput{Delegator: boss, Delegate: deputy, ValidFrom: dlToday, Reason: "leave"}) require.NoError(t, err, "PRESENCE: the ordinary case succeeds")
_, err = dlCreate(w, workflow.DelegationInput{Delegator: boss, Delegate: deputy, ValidFrom: dlToday.AddDate(0, 0, -1), Reason: "leave"}) require.ErrorIs(t, err, workflow.ErrBackdated) _, err = dlCreate(w, workflow.DelegationInput{Delegator: boss, Delegate: boss, ValidFrom: dlToday, Reason: "x"}) require.ErrorIs(t, err, workflow.ErrSelfDelegation) w.DeactivateMember(deputy) _, err = dlCreate(w, workflow.DelegationInput{Delegator: boss, Delegate: deputy, ValidFrom: dlToday, Reason: "x"}) require.ErrorIs(t, err, workflow.ErrDelegateInactive)}
func TestOnlyTheDelegatorRevokesAndARevocationIsAudited(t *testing.T) { w := workflowtest.NewWorld(t) boss, deputy, other := w.Member("Boss"), w.Member("Deputy"), w.Member("Other") d, err := dlCreate(w, workflow.DelegationInput{Delegator: boss, Delegate: deputy, ValidFrom: dlToday, Reason: "leave"}) require.NoError(t, err) err = w.InTx(func(ctx context.Context, tx pgx.Tx) error { return workflow.RevokeDelegation(ctx, tx, w.Clock, d.ID, other) }) require.ErrorIs(t, err, workflow.ErrNoDelegation) w.MustTx(func(ctx context.Context, tx pgx.Tx) error { return workflow.RevokeDelegation(ctx, tx, w.Clock, d.ID, boss) }) var n int require.NoError(t, w.Admin.QueryRow(context.Background(), `SELECT count(*) FROM audit_log WHERE org_id = $1 AND entity_kind = 'workflow_delegation'`, w.Org).Scan(&n)) require.Equal(t, 2, n, "create and revoke")}- Step 2: Run them and watch them fail
Run: go test ./internal/platform/workflow/ -run 'Delegation' -count=1
Expected: FAIL — undefined.
- Step 3: Write
delegation.go
package workflow
import ( "context" "errors" "fmt" "strings" "time"
"github.com/google/uuid" "github.com/jackc/pgx/v5"
"github.com/mcom/erp/internal/kernel/audit" "github.com/mcom/erp/internal/kernel/clock")
var ( ErrSelfDelegation = errors.New("workflow: a member cannot delegate to themselves") ErrDelegateInactive = errors.New("workflow: the delegate is not an active member of this organization") ErrBackdated = errors.New("workflow: a delegation cannot start before today") ErrBadWindow = errors.New("workflow: a delegation must end after it starts, and needs a reason") ErrNoDelegation = errors.New("workflow: no such delegation of yours") ErrAlreadyRevoked = errors.New("workflow: the delegation is already revoked"))
type Delegation struct { ID uuid.UUID DelegatorMemberID uuid.UUID DelegateMemberID uuid.UUID Kind *string ValidFrom time.Time ValidTo *time.Time Reason string CreatedAt time.Time RevokedAt *time.Time}
type DelegationInput struct { Delegator, Delegate uuid.UUID Kind *string ValidFrom time.Time ValidTo *time.Time Reason string}
const delegationColumns = `id, delegator_member_id, delegate_member_id, kind, valid_from, valid_to, reason, created_at, revoked_at`
func scanDelegation(row pgx.Row) (Delegation, error) { var d Delegation err := row.Scan(&d.ID, &d.DelegatorMemberID, &d.DelegateMemberID, &d.Kind, &d.ValidFrom, &d.ValidTo, &d.Reason, &d.CreatedAt, &d.RevokedAt) return d, err}
// CreateDelegation lends the delegator's authority (spec 17.4). Authority is// never retroactive: a delegation starting before today would make a past// decision's authority something invented after it was taken.func CreateDelegation(ctx context.Context, tx pgx.Tx, c clock.Clock, in DelegationInput) (Delegation, error) { org, err := orgOf(ctx) if err != nil { return Delegation{}, err } switch { case in.Delegator == in.Delegate: return Delegation{}, ErrSelfDelegation case in.ValidFrom.Before(clock.TodayUTC(c)): return Delegation{}, ErrBackdated case in.ValidTo != nil && !in.ValidTo.After(in.ValidFrom), strings.TrimSpace(in.Reason) == "": return Delegation{}, ErrBadWindow case in.Kind != nil && !KeyPattern.MatchString(*in.Kind): return Delegation{}, ErrBadWindow } live, err := idSet(ctx, tx, `SELECT id FROM organization_member WHERE org_id = $1 AND id = ANY($2) AND active`, org, []uuid.UUID{in.Delegate}) if err != nil { return Delegation{}, err } if !live[in.Delegate] { return Delegation{}, ErrDelegateInactive } d, err := scanDelegation(tx.QueryRow(ctx, ` INSERT INTO workflow_delegation (org_id, delegator_member_id, delegate_member_id, kind, valid_from, valid_to, reason, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING `+delegationColumns, org, in.Delegator, in.Delegate, in.Kind, in.ValidFrom, in.ValidTo, in.Reason, c.Now())) if err != nil { return Delegation{}, fmt.Errorf("workflow: create delegation: %w", err) } if err := audit.Write(ctx, tx, c, audit.Entry{EntityKind: "workflow_delegation", EntityID: d.ID, Action: "create", ActorID: &in.Delegator, After: d}); err != nil { return Delegation{}, err } return d, nil}
// RevokeDelegation ends a delegation now. Only its delegator may: another// member's attempt is indistinguishable from a missing id.func RevokeDelegation(ctx context.Context, tx pgx.Tx, c clock.Clock, id, by uuid.UUID) error { org, err := orgOf(ctx) if err != nil { return err } d, err := scanDelegation(tx.QueryRow(ctx, `SELECT `+delegationColumns+` FROM workflow_delegation WHERE org_id = $1 AND id = $2 AND delegator_member_id = $3 FOR UPDATE`, org, id, by)) if errors.Is(err, pgx.ErrNoRows) { return ErrNoDelegation } if err != nil { return fmt.Errorf("workflow: revoke delegation: %w", err) } if d.RevokedAt != nil { return ErrAlreadyRevoked } now := c.Now() if _, err := tx.Exec(ctx, `UPDATE workflow_delegation SET revoked_at = $3 WHERE org_id = $1 AND id = $2`, org, id, now); err != nil { return fmt.Errorf("workflow: revoke delegation: %w", err) } after := d after.RevokedAt = &now return audit.Write(ctx, tx, c, audit.Entry{EntityKind: "workflow_delegation", EntityID: id, Action: "revoke", ActorID: &by, Before: d, After: after})}
func ListDelegations(ctx context.Context, tx pgx.Tx, member uuid.UUID) ([]Delegation, error) { org, err := orgOf(ctx) if err != nil { return nil, err } return queryDelegations(ctx, tx, `SELECT `+delegationColumns+` FROM workflow_delegation WHERE org_id = $1 AND (delegator_member_id = $2 OR delegate_member_id = $2) ORDER BY created_at DESC, id`, org, member)}
// ActiveDelegationsTo is every delegation lending authority to delegate for// this kind on this day. One hop only: authority a delegate borrowed is not// itself lendable, because nothing here follows delegator -> delegate chains.func ActiveDelegationsTo(ctx context.Context, tx pgx.Tx, delegate uuid.UUID, kind string, on time.Time) ([]Delegation, error) { org, err := orgOf(ctx) if err != nil { return nil, err } return queryDelegations(ctx, tx, `SELECT `+delegationColumns+` FROM workflow_delegation WHERE org_id = $1 AND delegate_member_id = $2 AND revoked_at IS NULL AND (kind IS NULL OR kind = $3) AND valid_from <= $4 AND (valid_to IS NULL OR valid_to > $4) ORDER BY created_at, id`, org, delegate, kind, on)}
func queryDelegations(ctx context.Context, tx pgx.Tx, sql string, args ...any) ([]Delegation, error) { rows, err := tx.Query(ctx, sql, args...) if err != nil { return nil, fmt.Errorf("workflow: delegations: %w", err) } defer rows.Close() var out []Delegation for rows.Next() { d, err := scanDelegation(rows) if err != nil { return nil, err } out = append(out, d) } return out, rows.Err()}orgOf and idSet are Task 2’s sqlhelp.go.
- Step 4: Run the tests
Run: go test ./internal/platform/workflow/ -count=1
Expected: PASS.
- Step 5: Commit
git add internal/platform/workflow/delegation.go internal/platform/workflow/delegation_test.gogit commit -m "Add workflow delegations
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>"Task 6: The ladder, routing, and authorization (thresholds, self-approval, delegation, escalation)
Section titled “Task 6: The ladder, routing, and authorization (thresholds, self-approval, delegation, escalation)”Files:
- Create:
internal/platform/workflow/authorize.go - Test:
internal/platform/workflow/ladder_test.go(pure),internal/platform/workflow/authorize_test.go(helpers prefixedaz)
Interfaces:
-
Consumes: Task 2 (
Definition,EdgeSpec,GateState,ApprovalStates,GrantsOn); Task 4 (Resolve,Resolved,Gap); Task 5 (ActiveDelegationsTo);orgstruct.PrimaryPositionOf,orgstruct.NextFilledAbove. -
Produces:
type Via string // ViaReporter "reporter", ViaGrant "grant", ViaDelegation "delegation", ViaEscalation "escalation" — the values of workflow_decision.authority_viaconst (RefuseNotReporter = "not_reporter"RefuseNoBand = "no_band_covers_amount"RefuseNotInBand = "not_in_band"RefuseSelfApproval = "self_approval"RefuseDisputeRaiser = "dispute_raiser")type Refusal struct { Code string `json:"code"`; Detail string `json:"detail"`; Gaps []Gap `json:"gaps,omitempty"` }type Subject struct { Kind string; Reporter uuid.UUID; Amount money.Amount; EnteredStateOn time.Time; DisputeRaiser *uuid.UUID }type Authority struct {Actor, AuthorityMember uuid.UUIDVia ViaGrantID *uuid.UUIDDelegationID *uuid.UUIDGateState string // "" for a reporter edgeWinningFloor *money.Amount // nil for a reporter edgeEligible []uuid.UUID // members of the winning band (reporter edge: the reporter plus all_states holders)Resolution []Resolved // every grant consulted, as resolvedResolvedFor time.Time}func Ladder(res []Resolved, amount money.Amount) (floor money.Amount, band []Resolved, ok bool)type StepRouting struct { State string; Resolution []Resolved; WinningFloor *string; Eligible []uuid.UUID; Gaps []Gap; Routable bool }func Route(ctx context.Context, tx pgx.Tx, def Definition, reporter uuid.UUID, amount money.Amount, on time.Time) ([]StepRouting, error)func Authorize(ctx context.Context, tx pgx.Tx, def Definition, e EdgeSpec, s Subject, actor uuid.UUID, on time.Time) (Authority, *Refusal, error) -
Step 1: Write the failing ladder tests (pure)
internal/platform/workflow/ladder_test.go. These pin spec 5.4’s load-bearing sentences one each.
package workflow_test
import ( "testing"
"github.com/google/uuid" "github.com/stretchr/testify/require"
"github.com/mcom/erp/internal/kernel/money" "github.com/mcom/erp/internal/platform/workflow")
func rung(floor string, members ...uuid.UUID) workflow.Resolved { g := workflow.GrantSpec{GranteeKind: workflow.GranteeMember} if floor != "" { f := money.MustParse(floor) g.MinAmount = &f } if members == nil { members = []uuid.UUID{} } return workflow.Resolved{Grant: workflow.Grant{ID: uuid.New(), GrantSpec: g}, Members: members}}
func bandOf(t *testing.T, amount string, rungs ...workflow.Resolved) (string, []uuid.UUID) { t.Helper() floor, band, ok := workflow.Ladder(rungs, money.MustParse(amount)) if !ok { return "", nil } var ms []uuid.UUID for _, r := range band { ms = append(ms, r.Members...) } return floor.String(), ms}
func TestTheRM0ApproverDoesNotAuthorizeTheCEOsClaims(t *testing.T) { clerk, ceo := uuid.New(), uuid.New() _, got := bandOf(t, "4999.99", rung("", clerk), rung("5000", ceo)) require.Equal(t, []uuid.UUID{clerk}, got, "PRESENCE: below the CEO's floor the clerk approves") _, got = bandOf(t, "5000", rung("", clerk), rung("5000", ceo)) require.Equal(t, []uuid.UUID{ceo}, got, "the winning floor is computed over the whole step, not the actor's rows")}
func TestBandsAreHalfOpen(t *testing.T) { a, b := uuid.New(), uuid.New() floor, got := bandOf(t, "2000.00", rung("0", a), rung("2000", b)) require.Equal(t, "2000.0000", floor) require.Equal(t, []uuid.UUID{b}, got, "RM2,000.00 exactly lands in the upper band and nowhere else")}
func TestAGrantResolvingToNobodyIsAbsentFromTheLadder(t *testing.T) { a := uuid.New() _, got := bandOf(t, "6000", rung("", a), rung("5000")) require.Equal(t, []uuid.UUID{a}, got)}
func TestThereIsNoFallbackWhenNoBandReachesDown(t *testing.T) { floor, got := bandOf(t, "100", rung("500", uuid.New())) require.Empty(t, floor) require.Nil(t, got, "nobody is authorized and the move is refused")}
func TestOneFloorCanHoldSeveralRows(t *testing.T) { a, b := uuid.New(), uuid.New() _, got := bandOf(t, "10", rung("", a), rung("0", b)) require.ElementsMatch(t, []uuid.UUID{a, b}, got, "NULL and zero are the same floor")}- Step 2: Write the failing authorization tests
internal/platform/workflow/authorize_test.go. The chart is CEO > HEAD > STAFF, plus DEPUTY (no seat) and OUTSIDER.
package workflow_test
import ( "context" "testing" "time"
"github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/stretchr/testify/require"
"github.com/mcom/erp/internal/kernel/money" "github.com/mcom/erp/internal/platform/workflow" "github.com/mcom/erp/internal/platform/workflow/workflowtest")
type azEnv struct { rsChart deputy, outsider uuid.UUID def workflow.Definition}
// azDef is valid() with approved <- superior and paid <- all_states CEO seat,// wrapped as a Definition with fresh grant ids (Authorize never reads the// grant table; it is handed the Definition).func azNew(t *testing.T, mutate func(g *workflow.Graph, e *azEnv)) *azEnv { e := &azEnv{rsChart: rsNewChart(t)} e.deputy, e.outsider = e.w.Member("Deputy"), e.w.Member("Outsider") g := valid() ceoSeat := e.ceoSeat g.Grants = []workflow.GrantSpec{ {StateKey: "approved", GranteeKind: workflow.GranteeSuperior}, {AllStates: true, GranteeKind: workflow.GranteePosition, PositionID: &ceoSeat}, } if mutate != nil { mutate(&g, e) } g = g.Normalized() e.def = workflow.Definition{ID: uuid.New(), Version: 1, Graph: g} for range g.Grants { e.def.GrantIDs = append(e.def.GrantIDs, uuid.New()) } return e}
func (e *azEnv) authorize(t *testing.T, from, to string, s workflow.Subject, actor uuid.UUID) (workflow.Authority, *workflow.Refusal) { t.Helper() edge, ok := e.def.Edge(from, to) require.True(t, ok) var a workflow.Authority var r *workflow.Refusal e.w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var err error a, r, err = workflow.Authorize(ctx, tx, e.def, edge, s, actor, workflowtest.Today) return err }) return a, r}
func (e *azEnv) claim(reporter uuid.UUID, amount string) workflow.Subject { return workflow.Subject{Kind: "expense_claim", Reporter: reporter, Amount: money.MustParse(amount), EnteredStateOn: time.Date(2026, 9, 21, 0, 0, 0, 0, time.UTC)}}
func TestTheClaimantsSuperiorApprovesAndNobodyElseDoes(t *testing.T) { e := azNew(t, nil) a, r := e.authorize(t, "pending", "approved", e.claim(e.staff, "100"), e.head) require.Nil(t, r, "PRESENCE: STAFF's superior approves") require.Equal(t, workflow.ViaGrant, a.Via) require.Equal(t, e.head, a.AuthorityMember) require.ElementsMatch(t, []uuid.UUID{e.head, e.ceo}, a.Eligible, "superior and the all_states CEO share the zero band") require.Len(t, a.Resolution, 2, "every grant on the step is recorded as resolved")
_, r = e.authorize(t, "pending", "approved", e.claim(e.staff, "100"), e.outsider) require.Equal(t, workflow.RefuseNotInBand, r.Code)}
func TestTheReporterAloneTakesReporterEdges(t *testing.T) { e := azNew(t, nil) _, r := e.authorize(t, "draft", "pending", e.claim(e.staff, "100"), e.staff) require.Nil(t, r) _, r = e.authorize(t, "draft", "pending", e.claim(e.staff, "100"), e.outsider) require.Equal(t, workflow.RefuseNotReporter, r.Code) // all_states has standing on the claimant's own edges... _, r = e.authorize(t, "draft", "pending", e.claim(e.staff, "100"), e.ceo) require.Nil(t, r) // ...except the final acknowledgement. _, r = e.authorize(t, "paid", "done", e.claim(e.staff, "100"), e.ceo) require.Equal(t, workflow.RefuseNotReporter, r.Code)}
func TestSelfApprovalIsRefusedExceptAllStatesOnPostingEdges(t *testing.T) { e := azNew(t, nil) // The CEO files a claim. The CEO holds all_states. _, r := e.authorize(t, "pending", "approved", e.claim(e.ceo, "100"), e.ceo) require.Nil(t, r, "approved is an accrual state: all_states is exempt on posting transitions")
e = azNew(t, func(g *workflow.Graph, _ *azEnv) { g.States[2].Semantic, g.States[2].ControlKind = "pending", "" // approved is no longer a posting state g.States[3].Semantic = "custom" g.PostingOrder = workflow.AccrualFirst }) _, r = e.authorize(t, "pending", "approved", e.claim(e.ceo, "100"), e.ceo) require.Equal(t, workflow.RefuseSelfApproval, r.Code)}
func TestADelegateActsOnTheDelegatorsBandAndTheLogNamesBoth(t *testing.T) { e := azNew(t, nil) _, r := e.authorize(t, "pending", "approved", e.claim(e.staff, "100"), e.deputy) require.Equal(t, workflow.RefuseNotInBand, r.Code, "PRESENCE of the refusal the delegation lifts")
_, err := dlCreate(e.w, workflow.DelegationInput{Delegator: e.head, Delegate: e.deputy, ValidFrom: dlToday, Reason: "leave"}) require.NoError(t, err) a, r := e.authorize(t, "pending", "approved", e.claim(e.staff, "100"), e.deputy) require.Nil(t, r) require.Equal(t, workflow.ViaDelegation, a.Via) require.Equal(t, e.deputy, a.Actor) require.Equal(t, e.head, a.AuthorityMember, "authority lent, not replaced (spec 17.4)") require.NotNil(t, a.DelegationID)
// A delegate never acts on the delegator's own claim. _, err = dlCreate(e.w, workflow.DelegationInput{Delegator: e.staff, Delegate: e.outsider, ValidFrom: dlToday, Reason: "leave"}) require.NoError(t, err) _, r = e.authorize(t, "pending", "approved", e.claim(e.staff, "100"), e.outsider) require.NotNil(t, r)}
func TestEscalationAddsTheNextSeatUpOnlyAfterTheSLA(t *testing.T) { three := 3 e := azNew(t, func(g *workflow.Graph, _ *azEnv) { g.States[1].EscalateAfterDays = &three }) // Make the CEO not an all_states holder for this test, so only escalation can admit them. e.def.Grants = e.def.Grants[:1] e.def.GrantIDs = e.def.GrantIDs[:1]
fresh := e.claim(e.staff, "100") fresh.EnteredStateOn = time.Date(2026, 9, 19, 0, 0, 0, 0, time.UTC) // 2 days idle _, r := e.authorize(t, "pending", "approved", fresh, e.ceo) require.Equal(t, workflow.RefuseNotInBand, r.Code, "PRESENCE: before the SLA the CEO cannot decide")
idle := e.claim(e.staff, "100") idle.EnteredStateOn = time.Date(2026, 9, 18, 0, 0, 0, 0, time.UTC) // 3 days idle a, r := e.authorize(t, "pending", "approved", idle, e.ceo) require.Nil(t, r) require.Equal(t, workflow.ViaEscalation, a.Via) require.Equal(t, e.head, a.AuthorityMember, "the escalatee acts on the escalated-from approver's band")}
func TestTheRaiserOfADisputeCannotCloseIt(t *testing.T) { e := azNew(t, func(g *workflow.Graph, _ *azEnv) { g.States = append(g.States, st("disputed", "dispute", false)) g.Edges = append(g.Edges, rejecting("done", "disputed", workflow.GateReporter, false), ed("disputed", "done", workflow.GateSource)) g.Grants = append(g.Grants, workflow.GrantSpec{StateKey: "disputed", GranteeKind: workflow.GranteeUnitHead}) }) s := e.claim(e.staff, "100") _, r := e.authorize(t, "disputed", "done", s, e.head) require.Nil(t, r, "PRESENCE: the unit head closes a dispute someone else raised") s.DisputeRaiser = &e.head _, r = e.authorize(t, "disputed", "done", s, e.head) require.Equal(t, workflow.RefuseDisputeRaiser, r.Code)}
func TestRouteNamesTheGapsThatWouldStrandAClaim(t *testing.T) { e := azNew(t, nil) e.def.Grants = e.def.Grants[:1] // superior only e.def.GrantIDs = e.def.GrantIDs[:1] var ok, stranded []workflow.StepRouting e.w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var err error if ok, err = workflow.Route(ctx, tx, e.def, e.staff, money.MustParse("100"), workflowtest.Today); err != nil { return err } loner := e.w.Member("Loner") stranded, err = workflow.Route(ctx, tx, e.def, loner, money.MustParse("100"), workflowtest.Today) return err }) require.True(t, ok[0].Routable, "PRESENCE") require.Equal(t, "approved", stranded[0].State) require.False(t, stranded[0].Routable) require.Equal(t, "no_primary_position", stranded[0].Gaps[0].Reason)}Note: TestEscalationAddsTheNextSeatUpOnlyAfterTheSLA trims the all_states grant after azNew has normalized the graph; Authorize reads def.Grants and def.GrantIDs in parallel, so trim both. valid()’s paid-state grant is gone, which does not matter here — Authorize does not validate.
- Step 3: Run them and watch them fail
Run: go test ./internal/platform/workflow/ -run 'TestTheRM0|TestBands|TestAGrantResolvingToNobody|TestThereIsNo|TestOneFloor|TestTheClaimants|TestTheReporter|TestSelfApproval|TestADelegate|TestEscalation|TestTheRaiser|TestRoute' -count=1
Expected: FAIL — undefined.
- Step 4: Write
authorize.go
package workflow
import ( "context" "errors" "fmt" "slices" "time"
"github.com/google/uuid" "github.com/jackc/pgx/v5"
"github.com/mcom/erp/internal/kernel/money" "github.com/mcom/erp/internal/platform/orgstruct" "github.com/mcom/erp/internal/platform/posting")
type Via string
const ( ViaReporter Via = "reporter" ViaGrant Via = "grant" ViaDelegation Via = "delegation" ViaEscalation Via = "escalation")
const ( RefuseNotReporter = "not_reporter" RefuseNoBand = "no_band_covers_amount" RefuseNotInBand = "not_in_band" RefuseSelfApproval = "self_approval" RefuseDisputeRaiser = "dispute_raiser")
type Refusal struct { Code string `json:"code"` Detail string `json:"detail"` Gaps []Gap `json:"gaps,omitempty"`}
// Subject is what authorization needs to know about the document.type Subject struct { Kind string Reporter uuid.UUID Amount money.Amount EnteredStateOn time.Time // DisputeRaiser is the member who entered the current state, when that // state is a dispute: they may not take a source-gated exit (Ruling 7). DisputeRaiser *uuid.UUID}
type Authority struct { Actor uuid.UUID AuthorityMember uuid.UUID Via Via GrantID *uuid.UUID DelegationID *uuid.UUID GateState string WinningFloor *money.Amount Eligible []uuid.UUID Resolution []Resolved ResolvedFor time.Time}
// Ladder is spec 5.4's floor computation over RESOLVED grants: the winning// floor is the largest floor at or below the amount among grants that// resolved to somebody, computed over the whole step — never over the// actor's own rows, which would make every actor's own floor the largest// and let the RM0 approver authorize the CEO's claims. The band is every// resolved grant at exactly that floor. Floors only: a band's ceiling is the// next larger floor, so [floor, next) is half-open by construction.func Ladder(res []Resolved, amount money.Amount) (money.Amount, []Resolved, bool) { var best *money.Amount for _, r := range res { f := r.Grant.Floor() if len(r.Members) == 0 || f.Cmp(amount) > 0 { continue } if best == nil || f.Cmp(*best) > 0 { best = &f } } if best == nil { return money.Zero(), nil, false } var band []Resolved for _, r := range res { if len(r.Members) > 0 && r.Grant.Floor().Cmp(*best) == 0 { band = append(band, r) } } return *best, band, true}
type StepRouting struct { State string `json:"state"` Resolution []Resolved `json:"resolution"` WinningFloor *string `json:"winning_floor,omitempty"` Eligible []uuid.UUID `json:"eligible"` Gaps []Gap `json:"gaps,omitempty"` Routable bool `json:"routable"`}
// Route answers, for one claimant and amount, who would decide each approval// step — and which gaps would strand the claim. It is the submission check// (Ruling 13) and the editor's "who would approve this?" preview; the// readiness surface of spec 20.2 can call it too.func Route(ctx context.Context, tx pgx.Tx, def Definition, reporter uuid.UUID, amount money.Amount, on time.Time) ([]StepRouting, error) { var out []StepRouting for _, state := range def.ApprovalStates() { res, err := Resolve(ctx, tx, def.GrantsOn(state), reporter, on) if err != nil { return nil, err } st := StepRouting{State: state, Resolution: res, Eligible: []uuid.UUID{}} spec, _ := def.State(state) posts := spec.Semantic == posting.Accrual || spec.Semantic == posting.Payout for _, r := range res { if r.Gap != nil { st.Gaps = append(st.Gaps, *r.Gap) } } floor, band, ok := Ladder(res, amount) if ok { f := floor.String() st.WinningFloor = &f for _, r := range band { for _, m := range r.Members { if m != reporter || (r.Grant.AllStates && posts) { st.Eligible = append(st.Eligible, m) } } } } st.Routable = ok && len(st.Eligible) > 0 for _, r := range res { // A superior who is relevant at this amount and cannot be found // refuses the submission (spec 5.4 "refusal, not silence"). if r.Kind == GranteeSuperior && len(r.Members) == 0 && r.Grant.Floor().Cmp(amount) <= 0 { st.Routable = false } } if !st.Routable && len(st.Gaps) == 0 { gap := Gap{Reason: GapOnlyTheClaimant, MemberID: &reporter} if !ok { gap.Reason = RefuseNoBand } st.Gaps = append(st.Gaps, gap) } slices.SortFunc(st.Eligible, func(a, b uuid.UUID) int { return slices.Compare(a[:], b[:]) }) st.Eligible = slices.Compact(st.Eligible) out = append(out, st) } return out, nil}
// Authorize answers spec 4.3's second question for one edge: may this actor// take it, on this document, at this amount, today — and on whose// authority. It never returns an error for a refusal; the *Refusal is the// answer, and the Authority it returns alongside is still filled with what// was resolved, so a caller can show why.func Authorize(ctx context.Context, tx pgx.Tx, def Definition, e EdgeSpec, s Subject, actor uuid.UUID, on time.Time) (Authority, *Refusal, error) { a := Authority{Actor: actor, ResolvedFor: on, Resolution: []Resolved{}, Eligible: []uuid.UUID{}} to, _ := def.State(e.To) from, _ := def.State(e.From) posts := to.Semantic == posting.Accrual || to.Semantic == posting.Payout
if e.Gate == GateReporter { a.Eligible = []uuid.UUID{s.Reporter} if actor == s.Reporter { a.AuthorityMember, a.Via = actor, ViaReporter return a, nil, nil } // all_states: standing on the claimant's own edges, except the // final acknowledgement — only the claimant can say they were paid. if to.Semantic != posting.Acknowledged { var all []Grant for _, g := range def.GrantsOn(e.To) { if g.AllStates { all = append(all, g) } } res, err := Resolve(ctx, tx, all, s.Reporter, on) if err != nil { return a, nil, err } a.Resolution = res for _, r := range res { if slices.Contains(r.Members, actor) && r.Grant.Floor().Cmp(s.Amount) <= 0 { id := r.GrantID a.AuthorityMember, a.Via, a.GrantID = actor, ViaGrant, &id return a, nil, nil } } } return a, &Refusal{Code: RefuseNotReporter, Detail: "only the person who submitted this can take this action"}, nil }
gs, ok := def.GateState(e) if !ok { return a, nil, fmt.Errorf("workflow: edge %s->%s has no gate state; the definition is invalid", e.From, e.To) } a.GateState = gs res, err := Resolve(ctx, tx, def.GrantsOn(gs), s.Reporter, on) if err != nil { return a, nil, err } a.Resolution = res floor, band, ok := Ladder(res, s.Amount) if !ok { var gaps []Gap for _, r := range res { if r.Gap != nil { gaps = append(gaps, *r.Gap) } } return a, &Refusal{Code: RefuseNoBand, Gaps: gaps, Detail: fmt.Sprintf("nobody is granted %s at %s", gs, s.Amount)}, nil } a.WinningFloor = &floor for _, r := range band { a.Eligible = append(a.Eligible, r.Members...) } slices.SortFunc(a.Eligible, func(x, y uuid.UUID) int { return slices.Compare(x[:], y[:]) }) a.Eligible = slices.Compact(a.Eligible)
if e.Gate == GateSource && s.DisputeRaiser != nil && *s.DisputeRaiser == actor { return a, &Refusal{Code: RefuseDisputeRaiser, Detail: "the person who raised a dispute cannot resolve it"}, nil }
// 1. The actor's own grant in the winning band. For the reporter, only // an all_states grant on a posting transition counts (Ruling 12). for _, r := range band { if !slices.Contains(r.Members, actor) { continue } if actor == s.Reporter && !(r.Grant.AllStates && posts) { continue } id := r.GrantID a.AuthorityMember, a.Via, a.GrantID = actor, ViaGrant, &id return a, nil, nil } if actor == s.Reporter { return a, &Refusal{Code: RefuseSelfApproval, Detail: "you cannot approve your own submission"}, nil }
// 2. Delegation: authority lent by a member of the winning band, never // by the reporter (that would be self-approval by proxy). ds, err := ActiveDelegationsTo(ctx, tx, actor, s.Kind, on) if err != nil { return a, nil, err } for _, d := range ds { if d.DelegatorMemberID == s.Reporter { continue } for _, r := range band { if slices.Contains(r.Members, d.DelegatorMemberID) { gid, did := r.GrantID, d.ID a.AuthorityMember, a.Via, a.GrantID, a.DelegationID = d.DelegatorMemberID, ViaDelegation, &gid, &did return a, nil, nil } } }
// 3. Escalation (Ruling 10): once the state being left has been idle for // its SLA, the next filled seat above each band member may decide on // that member's authority. Never on source-gated (dispute) exits. if (e.Gate == GateTarget || e.Gate == GateAdvance) && from.EscalateAfterDays != nil && !on.Before(s.EnteredStateOn.AddDate(0, 0, *from.EscalateAfterDays)) { org, err := orgOf(ctx) if err != nil { return a, nil, err } for _, r := range band { for _, m := range r.Members { if m == s.Reporter { continue } pos, err := orgstruct.PrimaryPositionOf(ctx, tx, m, on) if errors.Is(err, orgstruct.ErrNoPrimaryPosition) { continue } if err != nil { return a, nil, err } up, err := orgstruct.NextFilledAbove(ctx, tx, pos, r.Grant.MaxEscalationLevels, on) if err != nil { return a, nil, err } if up, err = activeOnly(ctx, tx, org, up); err != nil { return a, nil, err } if slices.Contains(up, actor) { gid := r.GrantID a.AuthorityMember, a.Via, a.GrantID = m, ViaEscalation, &gid return a, nil, nil } } } }
return a, &Refusal{Code: RefuseNotInBand, Detail: fmt.Sprintf("you are not among those granted %s at this amount", gs)}, nil}Resolved.Grant must carry the grant’s MaxEscalationLevels; Normalized() (Task 2) guarantees it is ≥ 1, which NextFilledAbove requires.
- Step 5: Run the tests
Run: go test ./internal/platform/workflow/ -count=1
Expected: PASS.
- Step 6: Watch the two load-bearing checks fail once
(a) In Ladder, replace the band selection with “every resolved grant whose floor ≤ amount”: TestTheRM0ApproverDoesNotAuthorizeTheCEOsClaims goes RED. (b) Delete the if actor == s.Reporter && !(…) guard: TestSelfApprovalIsRefusedExceptAllStatesOnPostingEdges goes RED. Restore both.
- Step 7: Commit
git add internal/platform/workflow/authorize.go internal/platform/workflow/ladder_test.go internal/platform/workflow/authorize_test.gogit commit -m "Add the threshold ladder, routing and workflow authorization
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>"Task 7: The instance engine — start, transition, post, decide, log
Section titled “Task 7: The instance engine — start, transition, post, decide, log”Files:
- Create:
internal/platform/workflow/instance.go - Test:
internal/platform/workflow/instance_test.go(helpers prefixedin)
Interfaces:
-
Consumes: Task 3 (
Latest,Load,PublishGraph), Task 6 (Authorize,Route,Subject,Authority,Refusal,StepRouting);posting.Dispatch,posting.Event;audit.Write,audit.OriginFromContext;company.Get;clock.TodayUTC. -
Produces:
var ErrNoInstance, ErrInstanceExists, ErrStaleState, ErrNoSuchEdge, ErrReasonRequired, ErrNotDraft, ErrNegativeAmount errorconst RefuseNotRoutable = "not_routable"type RefusedError struct{ Refusal Refusal } // 403 at the edgetype SubmissionRefusedError struct{ Steps []StepRouting } // 422 at the edge; only the unroutable stepstype Instance struct {ID, DefinitionID uuid.UUID; Kind string; Version int; SourceID, Reporter uuid.UUIDStateKey string; Amount money.Amount; Currency string; AccrualOpen boolEnteredStateAt, EnteredStateOn, CreatedAt time.Time}type StartInput struct { Kind string; SourceID, Reporter uuid.UUID; Amount money.Amount }type TransitionInput struct { InstanceID uuid.UUID; From, To string; Actor uuid.UUID; Reason string }type Decision struct {ID, InstanceID, DefinitionID uuid.UUID; From, To string; Gate Gate; ActionKey stringActor, AuthorityMember uuid.UUID; Via Via; DelegationID, GrantID *uuid.UUIDAmount money.Amount; Currency string; WinningFloor *money.AmountResolvedFor time.Time; Eligible []uuid.UUID; Resolution json.RawMessageReason string; IP, RequestID *string; DecidedAt time.Time}type Available struct { To, ActionKey string; Gate Gate; RequiresReason, Allowed bool; Labels Labels; Refusal *Refusal }func Start(ctx context.Context, tx pgx.Tx, c clock.Clock, in StartInput) (Instance, error)func Get(ctx context.Context, tx pgx.Tx, id uuid.UUID) (Instance, error)func GetBySource(ctx context.Context, tx pgx.Tx, kind string, sourceID uuid.UUID) (Instance, error)func SetAmount(ctx context.Context, tx pgx.Tx, c clock.Clock, id uuid.UUID, amount money.Amount) errorfunc Transition(ctx context.Context, tx pgx.Tx, c clock.Clock, in TransitionInput) (Decision, error)func AvailableTransitions(ctx context.Context, tx pgx.Tx, c clock.Clock, id, actor uuid.UUID) ([]Available, error)func Decisions(ctx context.Context, tx pgx.Tx, instanceID uuid.UUID) ([]Decision, error)This is the contract Plans 5, 7 and 8 call. A consumer creates its document and calls
Startin the same transaction, callsSetAmountwhile its document is editable, exposesPOST /v1/<resource>/{id}/transitionsthat callsTransition(owning the role permission andIdempotency-Key, Ruling 14), and embedsAvailableTransitionsin its read response (spec 12.2). -
Step 1: Write the failing tests
internal/platform/workflow/instance_test.go:
package workflow_test
import ( "context" "encoding/json" "errors" "testing"
"github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/stretchr/testify/require"
"github.com/mcom/erp/internal/kernel/money" "github.com/mcom/erp/internal/platform/posting" "github.com/mcom/erp/internal/platform/workflow" "github.com/mcom/erp/internal/platform/workflow/workflowtest")
// inPoster is the fake ledger. The real one is Plan 4's; this plan only// proves it is called, with the right event, inside the transition.type inPoster struct { events []posting.Event fail error probe func(ctx context.Context, tx pgx.Tx, ev posting.Event)}
func (p *inPoster) Post(ctx context.Context, tx pgx.Tx, ev posting.Event) error { if p.probe != nil { p.probe(ctx, tx, ev) } if p.fail != nil { return p.fail } p.events = append(p.events, ev) return nil}
func inUsePoster(t *testing.T, p posting.Poster) { t.Helper() posting.ResetForTest() posting.Register(p) t.Cleanup(posting.ResetForTest)}
// inEnv is rsNewChart plus valid() published (approved <- superior,// paid <- superior), and a fake poster.type inEnv struct { rsChart def workflow.Definition poster *inPoster}
func inNew(t *testing.T) *inEnv { e := &inEnv{rsChart: rsNewChart(t), poster: &inPoster{}} inUsePoster(t, e.poster) e.def = stPublish(t, e.w, valid()) return e}
func (e *inEnv) start(t *testing.T, reporter uuid.UUID, amount string) workflow.Instance { t.Helper() var inst workflow.Instance e.w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var err error inst, err = workflow.Start(ctx, tx, e.w.Clock, workflow.StartInput{Kind: "expense_claim", SourceID: uuid.New(), Reporter: reporter, Amount: money.MustParse(amount)}) return err }) return inst}
func (e *inEnv) move(id uuid.UUID, from, to string, actor uuid.UUID, reason string) (workflow.Decision, error) { var d workflow.Decision err := e.w.InTx(func(ctx context.Context, tx pgx.Tx) error { var err error d, err = workflow.Transition(ctx, tx, e.w.Clock, workflow.TransitionInput{ InstanceID: id, From: from, To: to, Actor: actor, Reason: reason}) return err }) return d, err}
func (e *inEnv) mustMove(t *testing.T, id uuid.UUID, from, to string, actor uuid.UUID) workflow.Decision { t.Helper() d, err := e.move(id, from, to, actor, "because") require.NoError(t, err) return d}
func (e *inEnv) count(t *testing.T, sql string, args ...any) int { t.Helper() var n int require.NoError(t, e.w.Admin.QueryRow(context.Background(), sql, args...).Scan(&n)) return n}
func TestAnInstanceStaysPinnedToTheVersionItStartedOn(t *testing.T) { e := inNew(t) inst := e.start(t, e.staff, "100") require.Equal(t, 1, inst.Version) require.Equal(t, "draft", inst.StateKey)
stPublish(t, e.w, valid()) // v2 var again workflow.Instance e.w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var err error again, err = workflow.Get(ctx, tx, inst.ID) return err }) require.Equal(t, e.def.ID, again.DefinitionID, "publishing v2 touches nothing in flight (spec 5.2)") require.Equal(t, 2, e.start(t, e.staff, "5").Version, "a new document starts on the latest version")}
func TestAFullWalkPostsAccrualThenPayoutThroughTheSeam(t *testing.T) { e := inNew(t) inst := e.start(t, e.staff, "250.50") e.mustMove(t, inst.ID, "draft", "pending", e.staff) e.mustMove(t, inst.ID, "pending", "approved", e.head) e.mustMove(t, inst.ID, "approved", "paid", e.head) e.mustMove(t, inst.ID, "paid", "done", e.staff)
require.Len(t, e.poster.events, 2, "only accrual and payout reach the ledger") acc, pay := e.poster.events[0], e.poster.events[1] require.Equal(t, posting.Accrual, acc.Semantic) require.Equal(t, "ap", acc.ControlKind) require.False(t, acc.PastAccrual) require.Equal(t, posting.Payout, pay.Semantic) require.Equal(t, inst.SourceID, pay.SourceID) require.Equal(t, "expense_claim", pay.Kind) require.Equal(t, e.head, pay.ActorMemberID) require.Equal(t, workflowtest.Today, pay.BusinessDate)
require.Equal(t, 4, e.count(t, `SELECT count(*) FROM workflow_decision WHERE org_id = $1 AND instance_id = $2`, e.w.Org, inst.ID)) require.Equal(t, 4, e.count(t, `SELECT count(*) FROM audit_log WHERE org_id = $1 AND entity_id = $2 AND action = 'transition'`, e.w.Org, inst.ID))}
func TestRejectingReversesExactlyWhenSomethingWasAccrued(t *testing.T) { e := inNew(t) inst := e.start(t, e.staff, "100") e.mustMove(t, inst.ID, "draft", "pending", e.staff) e.mustMove(t, inst.ID, "pending", "rejected", e.head) // advance: the approver of 'approved' may reject require.Empty(t, e.poster.events, "rejecting before accrual posts nothing")
e.mustMove(t, inst.ID, "rejected", "draft", e.staff) e.mustMove(t, inst.ID, "draft", "pending", e.staff) e.mustMove(t, inst.ID, "pending", "approved", e.head) e.mustMove(t, inst.ID, "approved", "rejected", e.head) require.Len(t, e.poster.events, 2) require.Equal(t, posting.Rejected, e.poster.events[1].Semantic) require.True(t, e.poster.events[1].PastAccrual, "rejecting the accrued claim reverses it (spec 6.3)")}
func TestAPosterFailureAbortsTheWholeTransition(t *testing.T) { e := inNew(t) inst := e.start(t, e.staff, "100") e.mustMove(t, inst.ID, "draft", "pending", e.staff) decisions := func() int { return e.count(t, `SELECT count(*) FROM workflow_decision WHERE org_id = $1 AND instance_id = $2`, e.w.Org, inst.ID) } require.Equal(t, 1, decisions(), "PRESENCE: a successful transition leaves its decision")
e.poster.fail = errors.New("ledger says no") _, err := e.move(inst.ID, "pending", "approved", e.head, "") require.ErrorContains(t, err, "ledger says no") require.Equal(t, 1, decisions(), "no decision without its posting") require.Equal(t, 0, e.count(t, `SELECT count(*) FROM workflow_instance WHERE org_id = $1 AND id = $2 AND state_key = 'approved'`, e.w.Org, inst.ID))}
func TestThePosterRunsInsideTheTransactionBeforeItsAuditRow(t *testing.T) { e := inNew(t) inst := e.start(t, e.staff, "100") e.mustMove(t, inst.ID, "draft", "pending", e.staff) var decisionsSeen, auditSeen int = -1, -1 e.poster.probe = func(ctx context.Context, tx pgx.Tx, ev posting.Event) { _ = tx.QueryRow(ctx, `SELECT count(*) FROM workflow_decision WHERE org_id = $1 AND instance_id = $2 AND to_key = 'approved'`, ev.OrgID, ev.InstanceID).Scan(&decisionsSeen) _ = tx.QueryRow(ctx, `SELECT count(*) FROM audit_log WHERE org_id = $1 AND entity_id = $2 AND action = 'transition'`, ev.OrgID, ev.InstanceID).Scan(&auditSeen) } e.mustMove(t, inst.ID, "pending", "approved", e.head) require.Equal(t, 1, decisionsSeen, "the poster sees the decision: same transaction") require.Equal(t, 1, auditSeen, "only draft->pending's audit row exists yet: Dispatch precedes this transition's audit.Write, so a Poster's advisory locks precede class 1")}
func TestTheDecisionLogKeepsTheApproverSetWhenTheChartLaterChanges(t *testing.T) { e := inNew(t) inst := e.start(t, e.staff, "100") e.mustMove(t, inst.ID, "draft", "pending", e.staff) d := e.mustMove(t, inst.ID, "pending", "approved", e.head) require.Equal(t, []uuid.UUID{e.head}, d.Eligible) require.Equal(t, e.head, d.AuthorityMember) require.Equal(t, workflow.ViaGrant, d.Via) require.Equal(t, workflowtest.Today, d.ResolvedFor) var res []workflow.Resolved require.NoError(t, json.Unmarshal(d.Resolution, &res)) require.Equal(t, []uuid.UUID{e.head}, res[0].Members)
// Re-draw the chart: STAFF now reports straight to the CEO. _, err := e.w.Admin.Exec(context.Background(), `UPDATE org_position SET reports_to_position_id = $1 WHERE org_id = $2 AND id = $3`, e.ceoSeat, e.w.Org, e.staffSeat) require.NoError(t, err) var now []workflow.StepRouting var logged []workflow.Decision e.w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var err error if now, err = workflow.Route(ctx, tx, e.def, e.staff, money.MustParse("100"), workflowtest.Today); err != nil { return err } logged, err = workflow.Decisions(ctx, tx, inst.ID) return err }) require.Equal(t, []uuid.UUID{e.ceo}, now[0].Eligible, "PRESENCE: re-resolving today gives a different answer") require.Equal(t, []uuid.UUID{e.head}, logged[1].Eligible, "the log still says who was entitled then")}
func TestAStaleFromOrAMissingReasonIsRefused(t *testing.T) { e := inNew(t) inst := e.start(t, e.staff, "100") _, err := e.move(inst.ID, "pending", "approved", e.head, "") require.ErrorIs(t, err, workflow.ErrStaleState) e.mustMove(t, inst.ID, "draft", "pending", e.staff) _, err = e.move(inst.ID, "pending", "rejected", e.head, " ") require.ErrorIs(t, err, workflow.ErrReasonRequired) _, err = e.move(inst.ID, "pending", "paid", e.head, "") require.ErrorIs(t, err, workflow.ErrNoSuchEdge) _, err = e.move(inst.ID, "pending", "approved", e.staff, "") var refused *workflow.RefusedError require.ErrorAs(t, err, &refused) require.Equal(t, workflow.RefuseSelfApproval, refused.Refusal.Code)}
func TestASubmissionThatCannotBeRoutedIsRefusedNamingTheGap(t *testing.T) { e := inNew(t) ok := e.start(t, e.staff, "100") e.mustMove(t, ok.ID, "draft", "pending", e.staff) // PRESENCE
loner := e.w.Member("Loner") inst := e.start(t, loner, "100") _, err := e.move(inst.ID, "draft", "pending", loner, "") var sub *workflow.SubmissionRefusedError require.ErrorAs(t, err, &sub) require.Equal(t, "approved", sub.Steps[0].State) require.Equal(t, "no_primary_position", sub.Steps[0].Gaps[0].Reason)}
func TestTheAmountIsFrozenOnceSubmitted(t *testing.T) { e := inNew(t) inst := e.start(t, e.staff, "100") set := func(a string) error { return e.w.InTx(func(ctx context.Context, tx pgx.Tx) error { return workflow.SetAmount(ctx, tx, e.w.Clock, inst.ID, money.MustParse(a)) }) } require.NoError(t, set("120"), "PRESENCE: a draft's amount changes") e.mustMove(t, inst.ID, "draft", "pending", e.staff) require.ErrorIs(t, set("130"), workflow.ErrNotDraft)}
func TestTheDecisionLogIsAppendOnlyForTheApplication(t *testing.T) { e := inNew(t) inst := e.start(t, e.staff, "100") e.mustMove(t, inst.ID, "draft", "pending", e.staff) for _, stmt := range []string{ `UPDATE workflow_decision SET reason = 'edited' WHERE org_id = $1 AND instance_id = $2`, `DELETE FROM workflow_decision WHERE org_id = $1 AND instance_id = $2`, } { err := e.w.InTx(func(ctx context.Context, tx pgx.Tx) error { _, err := tx.Exec(ctx, stmt, e.w.Org, inst.ID) return err }) require.ErrorContains(t, err, "permission denied", stmt) }}
func TestAvailableTransitionsSayWhyNot(t *testing.T) { e := inNew(t) inst := e.start(t, e.staff, "100") outsider := e.w.Member("Outsider") for actor, allowed := range map[uuid.UUID]bool{e.staff: true, outsider: false} { var av []workflow.Available e.w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var err error av, err = workflow.AvailableTransitions(ctx, tx, e.w.Clock, inst.ID, actor) return err }) require.Len(t, av, 1) require.Equal(t, "pending", av[0].To) require.Equal(t, allowed, av[0].Allowed) if !allowed { require.Equal(t, workflow.RefuseNotReporter, av[0].Refusal.Code) } }}
func TestAnInstanceIsInvisibleToAnotherTenant(t *testing.T) { e := inNew(t) inst := e.start(t, e.staff, "100") e.w.MustTx(func(ctx context.Context, tx pgx.Tx) error { // PRESENCE _, err := workflow.Get(ctx, tx, inst.ID) return err }) err := e.w.SecondOrg().InTx(func(ctx context.Context, tx pgx.Tx) error { _, err := workflow.Get(ctx, tx, inst.ID) return err }) require.ErrorIs(t, err, workflow.ErrNoInstance)}- Step 2: Run them and watch them fail
Run: go test ./internal/platform/workflow/ -run 'TestAnInstance|TestAFullWalk|TestRejecting|TestAPoster|TestThePoster|TestTheDecisionLog|TestAStale|TestASubmission|TestTheAmount|TestAvailable' -count=1
Expected: FAIL — Start, Transition and friends undefined.
- Step 3: Write
instance.go
package workflow
import ( "context" "encoding/json" "errors" "fmt" "strings" "time"
"github.com/google/uuid" "github.com/jackc/pgx/v5"
"github.com/mcom/erp/internal/kernel/audit" "github.com/mcom/erp/internal/kernel/clock" "github.com/mcom/erp/internal/kernel/money" "github.com/mcom/erp/internal/platform/company" "github.com/mcom/erp/internal/platform/posting")
var ( ErrNoInstance = errors.New("workflow: no such instance") ErrInstanceExists = errors.New("workflow: this document already has an instance of this kind") ErrStaleState = errors.New("workflow: the instance is no longer in the state the caller saw") ErrNoSuchEdge = errors.New("workflow: no such transition from the current state") ErrReasonRequired = errors.New("workflow: this transition requires a reason") ErrNotDraft = errors.New("workflow: the amount can only change while the instance is a draft") ErrNegativeAmount = errors.New("workflow: an amount cannot be negative") errNoDraftSemantic = errors.New("workflow: the definition has no draft state"))
const RefuseNotRoutable = "not_routable"
type RefusedError struct{ Refusal Refusal }
func (e *RefusedError) Error() string { return "workflow: refused: " + e.Refusal.Code + ": " + e.Refusal.Detail}
type SubmissionRefusedError struct{ Steps []StepRouting }
func (e *SubmissionRefusedError) Error() string { parts := make([]string, len(e.Steps)) for i, s := range e.Steps { parts[i] = s.State } return "workflow: submission refused: nobody can decide " + strings.Join(parts, ", ")}
type Instance struct { ID uuid.UUID DefinitionID uuid.UUID Kind string Version int SourceID uuid.UUID Reporter uuid.UUID StateKey string Amount money.Amount Currency string AccrualOpen bool EnteredStateAt time.Time EnteredStateOn time.Time CreatedAt time.Time}
type StartInput struct { Kind string SourceID uuid.UUID Reporter uuid.UUID Amount money.Amount}
type TransitionInput struct { InstanceID uuid.UUID From, To string Actor uuid.UUID Reason string}
type Decision struct { ID, InstanceID, DefinitionID uuid.UUID From, To string Gate Gate ActionKey string Actor, AuthorityMember uuid.UUID Via Via DelegationID, GrantID *uuid.UUID Amount money.Amount Currency string WinningFloor *money.Amount ResolvedFor time.Time Eligible []uuid.UUID Resolution json.RawMessage Reason string IP, RequestID *string DecidedAt time.Time}
type Available struct { To string `json:"to"` ActionKey string `json:"action_key"` Gate Gate `json:"gate"` RequiresReason bool `json:"requires_reason"` Allowed bool `json:"allowed"` Labels Labels `json:"labels"` Refusal *Refusal `json:"refusal,omitempty"`}
// instanceSelect joins the pinned definition for its version. Both tables// are tied to $1: the join equality carries the tenant predicate to d.const instanceSelect = ` SELECT i.id, i.definition_id, i.kind, d.version, i.source_id, i.reporter_member_id, i.state_key, i.amount::text, i.currency, i.accrual_open, i.entered_state_at, i.entered_state_on, i.created_at FROM workflow_instance i JOIN workflow_definition d ON d.org_id = i.org_id AND d.id = i.definition_id WHERE i.org_id = $1 AND `
func scanInstance(row pgx.Row) (Instance, error) { var in Instance var amount string err := row.Scan(&in.ID, &in.DefinitionID, &in.Kind, &in.Version, &in.SourceID, &in.Reporter, &in.StateKey, &amount, &in.Currency, &in.AccrualOpen, &in.EnteredStateAt, &in.EnteredStateOn, &in.CreatedAt) if errors.Is(err, pgx.ErrNoRows) { return Instance{}, ErrNoInstance } if err != nil { return Instance{}, fmt.Errorf("workflow: read instance: %w", err) } in.Amount, err = money.Parse(amount) return in, err}
func Get(ctx context.Context, tx pgx.Tx, id uuid.UUID) (Instance, error) { org, err := orgOf(ctx) if err != nil { return Instance{}, err } return scanInstance(tx.QueryRow(ctx, instanceSelect+`i.id = $2`, org, id))}
func GetBySource(ctx context.Context, tx pgx.Tx, kind string, sourceID uuid.UUID) (Instance, error) { org, err := orgOf(ctx) if err != nil { return Instance{}, err } return scanInstance(tx.QueryRow(ctx, instanceSelect+`i.kind = $2 AND i.source_id = $3`, org, kind, sourceID))}
// lockInstance reads the instance FOR UPDATE: two concurrent transitions of// one document serialize here, and the second sees the first's state and// fails ErrStaleState. This is the only serialization a transition needs —// no advisory lock class is added by this package.func lockInstance(ctx context.Context, tx pgx.Tx, org, id uuid.UUID) (Instance, error) { return scanInstance(tx.QueryRow(ctx, instanceSelect+`i.id = $2 FOR UPDATE OF i`, org, id))}
// Start pins a new document to the kind's latest published version, in its// draft state, in the base currency.func Start(ctx context.Context, tx pgx.Tx, c clock.Clock, in StartInput) (Instance, error) { org, err := orgOf(ctx) if err != nil { return Instance{}, err } if in.Amount.IsNegative() { return Instance{}, ErrNegativeAmount } def, err := Latest(ctx, tx, in.Kind) if err != nil { return Instance{}, err } draft := "" for _, s := range def.States { if s.Semantic == posting.Draft { draft = s.Key } } if draft == "" { return Instance{}, errNoDraftSemantic } p, err := company.Get(ctx, tx) if err != nil { return Instance{}, fmt.Errorf("workflow: start: %w", err) } now, today := c.Now(), clock.TodayUTC(c) var id uuid.UUID err = tx.QueryRow(ctx, ` INSERT INTO workflow_instance (org_id, definition_id, kind, source_id, reporter_member_id, state_key, amount, currency, accrual_open, entered_state_at, entered_state_on, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, false, $9, $10, $9) RETURNING id`, org, def.ID, in.Kind, in.SourceID, in.Reporter, draft, in.Amount.String(), p.BaseCurrency, now, today).Scan(&id) if isUnique(err) { return Instance{}, ErrInstanceExists } if err != nil { return Instance{}, fmt.Errorf("workflow: start: %w", err) } inst, err := Get(ctx, tx, id) if err != nil { return Instance{}, err } if err := audit.Write(ctx, tx, c, audit.Entry{EntityKind: "workflow_instance", EntityID: id, Action: "start", After: inst}); err != nil { return Instance{}, err } return inst, nil}
// SetAmount changes the amount a document will be decided at. Only while it// is a draft (Ruling 15): an approval is given at an amount, and an amount// moving under it would reassign the approver after the fact.func SetAmount(ctx context.Context, tx pgx.Tx, c clock.Clock, id uuid.UUID, amount money.Amount) error { org, err := orgOf(ctx) if err != nil { return err } if amount.IsNegative() { return ErrNegativeAmount } inst, err := lockInstance(ctx, tx, org, id) if err != nil { return err } def, err := Load(ctx, tx, inst.DefinitionID) if err != nil { return err } if s, _ := def.State(inst.StateKey); s.Semantic != posting.Draft { return ErrNotDraft } if _, err := tx.Exec(ctx, `UPDATE workflow_instance SET amount = $3 WHERE org_id = $1 AND id = $2`, org, id, amount.String()); err != nil { return fmt.Errorf("workflow: set amount: %w", err) } return audit.Write(ctx, tx, c, audit.Entry{EntityKind: "workflow_instance", EntityID: id, Action: "amount", Before: map[string]string{"amount": inst.Amount.String()}, After: map[string]string{"amount": amount.String()}})}
func subjectOf(ctx context.Context, tx pgx.Tx, org uuid.UUID, def Definition, inst Instance) (Subject, error) { s := Subject{Kind: inst.Kind, Reporter: inst.Reporter, Amount: inst.Amount, EnteredStateOn: inst.EnteredStateOn} if cur, _ := def.State(inst.StateKey); cur.Semantic == posting.Dispute { var raiser uuid.UUID err := tx.QueryRow(ctx, ` SELECT actor_member_id FROM workflow_decision WHERE org_id = $1 AND instance_id = $2 AND to_key = $3 ORDER BY seq DESC LIMIT 1`, org, inst.ID, inst.StateKey).Scan(&raiser) if err != nil && !errors.Is(err, pgx.ErrNoRows) { return s, fmt.Errorf("workflow: dispute raiser: %w", err) } if err == nil { s.DisputeRaiser = &raiser } } return s, nil}
func unroutable(steps []StepRouting) []StepRouting { var bad []StepRouting for _, s := range steps { if !s.Routable { bad = append(bad, s) } } return bad}
// Transition moves one instance along one edge. In ONE transaction, in this// order://// 1. lock the instance row; refuse a stale From;// 2. authorize (spec 5.3/5.4, delegation, escalation);// 3. on submission, refuse a claim no one could decide (Ruling 13);// 4. write the new state and the decision row, with the resolved approver// set persisted (org-structure spec 5.5);// 5. posting.Dispatch — the ledger, in this transaction; any error aborts// everything above (wave-1 contract §1);// 6. audit.Write — last, so every advisory lock a Poster takes is acquired// before the audit chain's class 1.func Transition(ctx context.Context, tx pgx.Tx, c clock.Clock, in TransitionInput) (Decision, error) { org, err := orgOf(ctx) if err != nil { return Decision{}, err } inst, err := lockInstance(ctx, tx, org, in.InstanceID) if err != nil { return Decision{}, err } if inst.StateKey != in.From { return Decision{}, ErrStaleState } def, err := Load(ctx, tx, inst.DefinitionID) if err != nil { return Decision{}, err } e, ok := def.Edge(inst.StateKey, in.To) if !ok { return Decision{}, ErrNoSuchEdge } reason := strings.TrimSpace(in.Reason) if e.RequiresReason && reason == "" { return Decision{}, ErrReasonRequired } today := clock.TodayUTC(c) subj, err := subjectOf(ctx, tx, org, def, inst) if err != nil { return Decision{}, err } auth, refusal, err := Authorize(ctx, tx, def, e, subj, in.Actor, today) if err != nil { return Decision{}, err } if refusal != nil { return Decision{}, &RefusedError{Refusal: *refusal} } from, _ := def.State(e.From) to, _ := def.State(e.To) if from.Semantic == posting.Draft && to.Semantic != posting.Draft { steps, err := Route(ctx, tx, def, inst.Reporter, inst.Amount, today) if err != nil { return Decision{}, err } if bad := unroutable(steps); len(bad) > 0 { return Decision{}, &SubmissionRefusedError{Steps: bad} } }
pastAccrual := inst.AccrualOpen open := pastAccrual switch to.Semantic { case posting.Accrual: open = true case posting.Rejected, posting.Cancelled: open = false } now := c.Now() if _, err := tx.Exec(ctx, ` UPDATE workflow_instance SET state_key = $3, accrual_open = $4, entered_state_at = $5, entered_state_on = $6 WHERE org_id = $1 AND id = $2`, org, inst.ID, e.To, open, now, today); err != nil { return Decision{}, fmt.Errorf("workflow: move instance: %w", err) }
resolution, err := json.Marshal(auth.Resolution) if err != nil { return Decision{}, fmt.Errorf("workflow: encode resolution: %w", err) } var ip, reqID *string if o, ok := audit.OriginFromContext(ctx); ok { if o.IP.IsValid() { s := o.IP.String() ip = &s } if o.RequestID != "" { reqID = &o.RequestID } } var reasonArg *string if reason != "" { reasonArg = &reason } d := Decision{InstanceID: inst.ID, DefinitionID: def.ID, From: e.From, To: e.To, Gate: e.Gate, ActionKey: e.ActionKey, Actor: in.Actor, AuthorityMember: auth.AuthorityMember, Via: auth.Via, DelegationID: auth.DelegationID, GrantID: auth.GrantID, Amount: inst.Amount, Currency: inst.Currency, WinningFloor: auth.WinningFloor, ResolvedFor: today, Eligible: auth.Eligible, Resolution: resolution, Reason: reason, IP: ip, RequestID: reqID, DecidedAt: now} if err := tx.QueryRow(ctx, ` INSERT INTO workflow_decision (org_id, instance_id, definition_id, from_key, to_key, gate, action_key, actor_member_id, authority_member_id, authority_via, delegation_id, grant_id, amount, currency, winning_floor, resolved_for, eligible_member_ids, resolution, reason, ip, request_id, decided_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20::inet, $21, $22) RETURNING id`, org, inst.ID, def.ID, e.From, e.To, string(e.Gate), e.ActionKey, in.Actor, auth.AuthorityMember, string(auth.Via), auth.DelegationID, auth.GrantID, inst.Amount.String(), inst.Currency, amountArg(auth.WinningFloor), today, auth.Eligible, resolution, reasonArg, ip, reqID, now).Scan(&d.ID); err != nil { return Decision{}, fmt.Errorf("workflow: record decision: %w", err) }
if err := posting.Dispatch(ctx, tx, posting.Event{ OrgID: org, Kind: def.Kind, InstanceID: inst.ID, SourceID: inst.SourceID, Semantic: to.Semantic, ControlKind: to.ControlKind, PastAccrual: pastAccrual, BusinessDate: today, ActorMemberID: in.Actor, }); err != nil { return Decision{}, fmt.Errorf("workflow: posting %s->%s: %w", e.From, e.To, err) }
if err := audit.Write(ctx, tx, c, audit.Entry{EntityKind: "workflow_instance", EntityID: inst.ID, Action: "transition", ActorID: &in.Actor, Before: map[string]any{"state": inst.StateKey, "accrual_open": inst.AccrualOpen}, After: map[string]any{"state": e.To, "accrual_open": open, "decision_id": d.ID, "authority_member_id": auth.AuthorityMember, "via": auth.Via}}); err != nil { return Decision{}, err } return d, nil}
// AvailableTransitions is spec 12.2's list: every exit of the current state,// whether this actor may take it, and why not. The frontend renders buttons// from it and never evaluates a gate or a threshold itself (spec 13.4).func AvailableTransitions(ctx context.Context, tx pgx.Tx, c clock.Clock, id, actor uuid.UUID) ([]Available, error) { org, err := orgOf(ctx) if err != nil { return nil, err } inst, err := Get(ctx, tx, id) if err != nil { return nil, err } def, err := Load(ctx, tx, inst.DefinitionID) if err != nil { return nil, err } subj, err := subjectOf(ctx, tx, org, def, inst) if err != nil { return nil, err } today := clock.TodayUTC(c) cur, _ := def.State(inst.StateKey) out := []Available{} for _, e := range def.EdgesFrom(inst.StateKey) { _, ref, err := Authorize(ctx, tx, def, e, subj, actor, today) if err != nil { return nil, err } to, _ := def.State(e.To) if ref == nil && cur.Semantic == posting.Draft && to.Semantic != posting.Draft { steps, err := Route(ctx, tx, def, inst.Reporter, inst.Amount, today) if err != nil { return nil, err } if bad := unroutable(steps); len(bad) > 0 { var gaps []Gap for _, s := range bad { gaps = append(gaps, s.Gaps...) } ref = &Refusal{Code: RefuseNotRoutable, Gaps: gaps, Detail: "nobody can currently decide this submission; the organization chart needs fixing"} } } out = append(out, Available{To: e.To, ActionKey: e.ActionKey, Gate: e.Gate, RequiresReason: e.RequiresReason, Allowed: ref == nil, Labels: e.Labels, Refusal: ref}) } return out, nil}
// Decisions is an instance's decision log, oldest first. Read as written —// never re-resolved.func Decisions(ctx context.Context, tx pgx.Tx, instanceID uuid.UUID) ([]Decision, error) { org, err := orgOf(ctx) if err != nil { return nil, err } rows, err := tx.Query(ctx, ` SELECT id, instance_id, definition_id, from_key, to_key, gate, action_key, actor_member_id, authority_member_id, authority_via, delegation_id, grant_id, amount::text, currency, winning_floor::text, resolved_for, eligible_member_ids, resolution, COALESCE(reason, ''), host(ip), request_id, decided_at FROM workflow_decision WHERE org_id = $1 AND instance_id = $2 ORDER BY seq`, org, instanceID) if err != nil { return nil, fmt.Errorf("workflow: decisions: %w", err) } defer rows.Close() var out []Decision for rows.Next() { var d Decision var gate, via, amount string var floor *string if err := rows.Scan(&d.ID, &d.InstanceID, &d.DefinitionID, &d.From, &d.To, &gate, &d.ActionKey, &d.Actor, &d.AuthorityMember, &via, &d.DelegationID, &d.GrantID, &amount, &d.Currency, &floor, &d.ResolvedFor, &d.Eligible, &d.Resolution, &d.Reason, &d.IP, &d.RequestID, &d.DecidedAt); err != nil { return nil, err } d.Gate, d.Via = Gate(gate), Via(via) if d.Amount, err = money.Parse(amount); err != nil { return nil, err } if d.WinningFloor, err = parseAmount(floor); err != nil { return nil, err } out = append(out, d) } return out, rows.Err()}- Step 4: Run the tests
Run: go test ./internal/platform/workflow/ -count=1
Expected: PASS. Then go test ./internal/kernel/layering/ -count=1 — expected PASS (every statement names org_id = $1 or inserts org_id; instanceSelect’s join carries it to workflow_definition).
- Step 5: Watch the ordering and the abort fail once
(a) Move the posting.Dispatch block below audit.Write: TestThePosterRunsInsideTheTransactionBeforeItsAuditRow goes RED (auditSeen = 2). (b) Change return Decision{}, fmt.Errorf("workflow: posting … to _ = err and continue: TestAPosterFailureAbortsTheWholeTransition goes RED. Restore both.
- Step 6: Commit
git add internal/platform/workflow/instance.go internal/platform/workflow/instance_test.gogit commit -m "Add the workflow instance engine: transitions, posting dispatch and the decision log
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>"Task 8: Seed the korre default flow as v1 — provisioning step 60 and its readiness warnings
Section titled “Task 8: Seed the korre default flow as v1 — provisioning step 60 and its readiness warnings”Files:
- Create:
internal/platform/workflow/seed.go - Test:
internal/platform/workflow/seed_test.go(helpers prefixedsd)
Interfaces:
-
Consumes: Task 3 (
PublishGraph,latestVersion), Task 7 (Start,Transitionfor the end-to-end test);provisioning.Register,provisioning.Warning,provisioning.Steps;orgstruct.ListPositions,orgstruct.HoldersOf. -
Produces:
const KindExpenseClaim = "expense_claim"const StepKey = "workflow_definitions" // provisioning step keyconst StepOrder = 60const (WarningGrantMemberInactive = "workflow_grant_member_inactive"WarningGrantPositionRetired = "workflow_grant_position_retired"WarningGrantVacant = "workflow_grant_vacant")var ErrNoOwnerSeat errorfunc DefaultExpenseClaim(ownerSeat uuid.UUID) Graphfunc SeededGraphs(ownerSeat uuid.UUID) []Graph // every graph provisioning seeds; the drift test iterates itfunc SeedDefaults(ctx context.Context, tx pgx.Tx, c clock.Clock) errorfunc CheckDefinitions(ctx context.Context, tx pgx.Tx, c clock.Clock) (bool, []provisioning.Warning, error)Registered at
init():provisioning.Register(definitionsStep{}). A binary sees this step only if it linksinternal/platform/workflow— Task 9 makes every binary link it throughmodules/workflow, and Task 12 asserts it. -
Step 1: Write the failing tests
internal/platform/workflow/seed_test.go:
package workflow_test
import ( "context" "testing"
"github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/stretchr/testify/require"
"github.com/mcom/erp/internal/kernel/money" "github.com/mcom/erp/internal/platform/posting" "github.com/mcom/erp/internal/platform/provisioning" "github.com/mcom/erp/internal/platform/workflow" "github.com/mcom/erp/internal/platform/workflow/workflowtest")
// TestEverySeededDefinitionPassesTheValidator is spec 20.1.1's fifth drift// test. A seed that fails its own validator would be refused at// provisioning time, for every new organization, with no one to fix it.func TestEverySeededDefinitionPassesTheValidator(t *testing.T) { graphs := workflow.SeededGraphs(uuid.New()) require.NotEmpty(t, graphs, "no seeded graphs; this guard would be vacuous") for _, g := range graphs { require.Empty(t, workflow.ValidateGraph(g, "MYR"), "seeded %s is invalid", g.Kind) }}
func TestTheSeededFlowIsSpec63(t *testing.T) { g := workflow.DefaultExpenseClaim(uuid.New()) sem := map[string]posting.Semantic{} for _, s := range g.States { sem[s.Key] = s.Semantic require.NotEmpty(t, s.Labels["zh"], "seeded labels ship en and zh together (spec 20.1.6): %s", s.Key) } require.Equal(t, posting.Accrual, sem["category_verified"]) require.Equal(t, posting.Payout, sem["completed"]) require.Equal(t, posting.Acknowledged, sem["received"]) cv, _ := g.State("category_verified") require.Equal(t, "ap", cv.ControlKind) _, back := g.Edge("completed", "approved") require.False(t, back, "completed -> approved does not exist; a missing payment is a dispute (spec 6.3)")}
func TestStep60IsRegisteredAfterOrgStructure(t *testing.T) { var keys []string for _, s := range provisioning.Steps() { keys = append(keys, s.Key()) if s.Key() == workflow.StepKey { require.Equal(t, 60, s.Order()) } } require.Contains(t, keys, "org_structure", "PRESENCE: the registry is the real one") require.Contains(t, keys, workflow.StepKey)}
// sdWorld is a provisioned-looking org: a root OWNER seat held by the owner,// a STAFF seat reporting to it, the root unit headed by OWNER.type sdWorld struct { w *workflowtest.World owner, staff uuid.UUID ownerSeat, staffSeat uuid.UUID}
func sdNew(t *testing.T) sdWorld { w := workflowtest.NewWorld(t) s := sdWorld{w: w} s.ownerSeat = w.Seat("OWNER", nil) s.staffSeat = w.Seat("STAFF", &s.ownerSeat) s.owner, s.staff = w.Member("Owner"), w.Member("Staff") w.Sit(s.owner, s.ownerSeat) w.Sit(s.staff, s.staffSeat) w.SetUnitHead(&s.ownerSeat) return s}
func sdSeed(w *workflowtest.World) error { return w.InTx(func(ctx context.Context, tx pgx.Tx) error { return workflow.SeedDefaults(ctx, tx, w.Clock) })}
func TestSeedingIsIdempotentAndKeepsTheCustomersEdits(t *testing.T) { s := sdNew(t) require.NoError(t, sdSeed(s.w)) var def workflow.Definition s.w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var err error def, err = workflow.Latest(ctx, tx, workflow.KindExpenseClaim) return err }) require.Equal(t, 1, def.Version) require.Nil(t, def.PublishedBy, "the seed is published by no member")
edited := def.Graph edited.States[1].Labels = workflow.Labels{"en": "Awaiting HOD"} stPublish(t, s.w, edited) // v2, the customer's require.NoError(t, sdSeed(s.w)) s.w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var err error def, err = workflow.Latest(ctx, tx, workflow.KindExpenseClaim) return err }) require.Equal(t, 2, def.Version, "a re-run never republishes over an org's own version (spec 20.2)")}
func TestSeedingRefusesWithoutAnOwnerSeat(t *testing.T) { w := workflowtest.NewWorld(t) require.ErrorIs(t, sdSeed(w), workflow.ErrNoOwnerSeat)}
func TestTheSeededClaimWalksEndToEnd(t *testing.T) { s := sdNew(t) p := &inPoster{} inUsePoster(t, p) require.NoError(t, sdSeed(s.w))
var inst workflow.Instance s.w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var err error inst, err = workflow.Start(ctx, tx, s.w.Clock, workflow.StartInput{Kind: workflow.KindExpenseClaim, SourceID: uuid.New(), Reporter: s.staff, Amount: money.MustParse("480.00")}) return err }) e := &inEnv{rsChart: rsChart{w: s.w}, poster: p} walk := []struct { from, to string actor uuid.UUID }{ {"draft", "pending_review", s.staff}, {"pending_review", "reviewed", s.owner}, // superior {"reviewed", "approved", s.owner}, // unit head {"approved", "category_verified", s.owner}, // OWNER seat — accrual {"category_verified", "completed", s.owner}, // payout {"completed", "received", s.staff}, {"received", "disputed", s.staff}, } for _, step := range walk { e.mustMove(t, inst.ID, step.from, step.to, step.actor) } // The claimant cannot close the dispute they raised. _, err := e.move(inst.ID, "disputed", "received", s.staff, "") var refused *workflow.RefusedError require.ErrorAs(t, err, &refused) // A second person upholds it, reversing the accrual. e.mustMove(t, inst.ID, "disputed", "rejected", s.owner)
var sems []posting.Semantic for _, ev := range p.events { sems = append(sems, ev.Semantic) } require.Equal(t, []posting.Semantic{posting.Accrual, posting.Payout, posting.Rejected}, sems) require.True(t, p.events[2].PastAccrual)}
func TestReadinessWarnsAboutAGrantThatCanNoLongerResolve(t *testing.T) { s := sdNew(t) require.NoError(t, sdSeed(s.w)) check := func() []provisioning.Warning { var ws []provisioning.Warning s.w.MustTx(func(ctx context.Context, tx pgx.Tx) error { done, got, err := workflow.CheckDefinitions(ctx, tx, s.w.Clock) require.True(t, done) ws = got return err }) return ws } require.Empty(t, check(), "PRESENCE: a healthy seed warns about nothing")
g := workflow.DefaultExpenseClaim(s.ownerSeat) m := s.w.Member("Named Approver") g.Grants = append(g.Grants, workflow.GrantSpec{StateKey: "reviewed", GranteeKind: workflow.GranteeMember, MemberID: &m}) stPublish(t, s.w, g) s.w.DeactivateMember(m) ws := check() require.Len(t, ws, 1) require.Equal(t, workflow.WarningGrantMemberInactive, ws[0].Code)}- Step 2: Run them and watch them fail
Run: go test ./internal/platform/workflow/ -run 'Seed|Step60|TheSeeded|Readiness' -count=1
Expected: FAIL — undefined.
- Step 3: Write
seed.go
package workflow
import ( "context" "errors" "fmt"
"github.com/google/uuid" "github.com/jackc/pgx/v5"
"github.com/mcom/erp/internal/kernel/clock" "github.com/mcom/erp/internal/platform/orgstruct" "github.com/mcom/erp/internal/platform/posting" "github.com/mcom/erp/internal/platform/provisioning")
const ( KindExpenseClaim = "expense_claim" StepKey = "workflow_definitions" // StepOrder is in Plan 3's reserved range (wave-1 contract §2: 60–69), // after org_structure (50), whose OWNER seat the seed's grants name. StepOrder = 60
WarningGrantMemberInactive = "workflow_grant_member_inactive" WarningGrantPositionRetired = "workflow_grant_position_retired" WarningGrantVacant = "workflow_grant_vacant")
var ErrNoOwnerSeat = errors.New( "workflow: provisioning needs the root OWNER position (step org_structure) before it can seed a workflow")
// Registered here, not in provisioning/steps.go: that file is shared with// Plan 4's branch (wave-1 contract §3), and each module that needs seeded// data registers its own step (spec 20.3). The cost is spec 2b §2.3's// lesson — a registry filled by init() is only as complete as the import// graph of the binary reading it — which cmd/*/wiring_test.go pins.func init() { provisioning.Register(definitionsStep{}) }
func ls(en, zh string) Labels { return Labels{"en": en, "zh": zh} }
// DefaultExpenseClaim is spec 6.3's flow, taken from korre://// draft → pending_review → reviewed → approved → category_verified → completed → received// │ │ │ │// └────────────────┴───────────┴──────────────┴──→ rejected → draft// completed ─→ disputed// received ─→ disputed//// Approval precedes the category check, and the accrual sits AT the check:// "never accrue to an account no human has verified" is kept by where the// accrual posts. category_verified is the posted-but-unpaid state, so// rejecting from it reverses. A dispute is dismissed to received or upheld// to rejected (reversal) — never back to completed, which would pay twice// (Ruling 16). The seeded grants name only what provisioning guarantees: the// claimant's superior, their unit head, and the root OWNER seat.func DefaultExpenseClaim(ownerSeat uuid.UUID) Graph { st := func(order int, key string, sem posting.Semantic, terminal bool, l Labels) StateSpec { s := StateSpec{Key: key, Semantic: sem, DisplayOrder: order, IsTerminal: terminal, Labels: l} if sem == posting.Accrual { s.ControlKind = "ap" // AP control for a receipt claim (spec 5.5) } return s } ed := func(order int, from, to string, gate Gate, action string, l Labels) EdgeSpec { return EdgeSpec{From: from, To: to, Gate: gate, ActionKey: action, DisplayOrder: order, Labels: l} } why := func(e EdgeSpec, reverses bool) EdgeSpec { e.RequiresReason, e.ReversesAccrual = true, reverses return e } owner := func(state string) GrantSpec { return GrantSpec{StateKey: state, GranteeKind: GranteePosition, PositionID: &ownerSeat} } reject := ls("Reject", "拒绝") return Graph{ Kind: KindExpenseClaim, PostingOrder: AccrualFirst, States: []StateSpec{ st(1, "draft", posting.Draft, false, ls("Draft", "草稿")), st(2, "pending_review", posting.Pending, false, ls("Pending review", "待审核")), st(3, "reviewed", posting.Pending, false, ls("Reviewed", "已审核")), st(4, "approved", posting.Pending, false, ls("Approved", "已批准")), st(5, "category_verified", posting.Accrual, false, ls("Category verified", "类别已核实")), st(6, "completed", posting.Payout, false, ls("Paid", "已付款")), st(7, "received", posting.Acknowledged, true, ls("Received", "已收款")), st(8, "rejected", posting.Rejected, false, ls("Rejected", "已拒绝")), st(9, "disputed", posting.Dispute, false, ls("Disputed", "有争议")), }, Edges: []EdgeSpec{ ed(1, "draft", "pending_review", GateReporter, "submit", ls("Submit", "提交")), ed(2, "pending_review", "reviewed", GateTarget, "review", ls("Mark reviewed", "审核通过")), ed(3, "reviewed", "approved", GateTarget, "approve", ls("Approve", "批准")), ed(4, "approved", "category_verified", GateTarget, "verify_category", ls("Verify category", "核实类别")), ed(5, "category_verified", "completed", GateTarget, "pay", ls("Mark paid", "标记已付款")), ed(6, "completed", "received", GateReporter, "acknowledge", ls("Confirm receipt", "确认收款")), why(ed(7, "pending_review", "rejected", GateAdvance, "reject", reject), false), why(ed(8, "reviewed", "rejected", GateAdvance, "reject", reject), false), why(ed(9, "approved", "rejected", GateAdvance, "reject", reject), false), why(ed(10, "category_verified", "rejected", GateAdvance, "reject", reject), true), ed(11, "rejected", "draft", GateReporter, "revise", ls("Revise", "修改")), why(ed(12, "completed", "disputed", GateReporter, "dispute", ls("Raise dispute", "提出争议")), false), why(ed(13, "received", "disputed", GateReporter, "dispute", ls("Raise dispute", "提出争议")), false), ed(14, "disputed", "received", GateSource, "dismiss", ls("Dismiss dispute", "驳回争议")), why(ed(15, "disputed", "rejected", GateSource, "uphold", ls("Uphold dispute", "支持争议")), true), }, Grants: []GrantSpec{ {StateKey: "reviewed", GranteeKind: GranteeSuperior}, {StateKey: "approved", GranteeKind: GranteeUnitHead}, owner("category_verified"), owner("completed"), owner("disputed"), }, }.Normalized()}
// SeededGraphs is every graph provisioning seeds. Plans 7 and 8 append// theirs here, so the validator drift test covers them without editing it.func SeededGraphs(ownerSeat uuid.UUID) []Graph { return []Graph{DefaultExpenseClaim(ownerSeat)}}
func ownerSeatOf(ctx context.Context, tx pgx.Tx) (uuid.UUID, error) { ps, err := orgstruct.ListPositions(ctx, tx) if err != nil { return uuid.Nil, err } for _, p := range ps { if p.Code == "OWNER" && p.ReportsToPositionID == nil && p.IsActive { return p.ID, nil } } return uuid.Nil, ErrNoOwnerSeat}
// SeedDefaults publishes each seeded graph as v1 for a kind that has never// been published. A kind with any version is left alone — including one the// organization has since edited (spec 20.2, first property).func SeedDefaults(ctx context.Context, tx pgx.Tx, c clock.Clock) error { org, err := orgOf(ctx) if err != nil { return err } seat, err := ownerSeatOf(ctx, tx) if err != nil { return err } for _, g := range SeededGraphs(seat) { v, err := latestVersion(ctx, tx, org, g.Kind) if err != nil { return err } if v > 0 { continue } if _, err := PublishGraph(ctx, tx, c, g, nil); err != nil { return fmt.Errorf("workflow: seed %s: %w", g.Kind, err) } } return nil}
// CheckDefinitions is the step's Check: done when every seeded kind has a// version; warnings name latest-version grants that can no longer resolve// (spec 20.2: readiness surfaces the 5.4 failure modes that appear after// provisioning).func CheckDefinitions(ctx context.Context, tx pgx.Tx, c clock.Clock) (bool, []provisioning.Warning, error) { org, err := orgOf(ctx) if err != nil { return false, nil, err } for _, g := range SeededGraphs(uuid.Nil) { v, err := latestVersion(ctx, tx, org, g.Kind) if err != nil { return false, nil, err } if v == 0 { return false, nil, nil } } rows, err := tx.Query(ctx, ` SELECT g.id, d.kind, g.grantee_kind, g.grantee_position_id, g.on_vacancy, COALESCE(m.active, true), COALESCE(p.is_active, true) FROM workflow_gate_grant g JOIN workflow_definition d ON d.org_id = g.org_id AND d.id = g.definition_id LEFT JOIN organization_member m ON m.org_id = g.org_id AND m.id = g.grantee_member_id LEFT JOIN org_position p ON p.org_id = g.org_id AND p.id = g.grantee_position_id WHERE g.org_id = $1 AND d.version = (SELECT max(x.version) FROM workflow_definition x WHERE x.org_id = $1 AND x.kind = d.kind) ORDER BY d.kind, g.id`, org) if err != nil { return false, nil, fmt.Errorf("workflow: readiness: %w", err) } type row struct { id uuid.UUID kind, grantee, vacancy string pos *uuid.UUID memberOK, seatOK bool } var rs []row for rows.Next() { var r row if err := rows.Scan(&r.id, &r.kind, &r.grantee, &r.pos, &r.vacancy, &r.memberOK, &r.seatOK); err != nil { rows.Close() return false, nil, err } rs = append(rs, r) } rows.Close() if err := rows.Err(); err != nil { return false, nil, err } var ws []provisioning.Warning today := clock.TodayUTC(c) for _, r := range rs { switch { case !r.memberOK: ws = append(ws, provisioning.Warning{Code: WarningGrantMemberInactive, Ref: r.id, Detail: fmt.Sprintf("a %s grant names a member who has left", r.kind)}) case !r.seatOK: ws = append(ws, provisioning.Warning{Code: WarningGrantPositionRetired, Ref: r.id, Detail: fmt.Sprintf("a %s grant names a retired position", r.kind)}) case r.pos != nil && OnVacancy(r.vacancy) == VacancyRefuse: h, err := orgstruct.HoldersOf(ctx, tx, *r.pos, today) if err != nil { return false, nil, err } if len(h) == 0 { ws = append(ws, provisioning.Warning{Code: WarningGrantVacant, Ref: r.id, Detail: fmt.Sprintf("a %s grant names a vacant position and will not escalate", r.kind)}) } } } return true, ws, nil}
type definitionsStep struct{}
func (definitionsStep) Key() string { return StepKey }func (definitionsStep) Order() int { return StepOrder }func (definitionsStep) Check(ctx context.Context, tx pgx.Tx, c clock.Clock) (bool, []provisioning.Warning, error) { return CheckDefinitions(ctx, tx, c)}func (definitionsStep) Run(ctx context.Context, tx pgx.Tx, c clock.Clock, _ provisioning.Input) error { return SeedDefaults(ctx, tx, c)}TestTheSeededClaimWalksEndToEnd reuses Task 7’s inEnv.move/mustMove by building an inEnv around this world — rsChart{w: s.w} sets only the world, which is all move reads.
- Step 4: Run the package and the provisioning package
Run: go test ./internal/platform/workflow/ ./internal/platform/provisioning/ -count=1
Expected: PASS. The provisioning package does not import workflow, so its own tests do not see step 60 — correct: registration follows the import graph.
- Step 5: Watch the drift test fail once
In DefaultExpenseClaim, change edge 10’s why(…, true) to why(…, false). TestEverySeededDefinitionPassesTheValidator goes RED with reversal_missing. Restore.
- Step 6: Commit
git add internal/platform/workflow/seed.go internal/platform/workflow/seed_test.gogit commit -m "Seed the korre expense-claim flow as v1 through provisioning step 60
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>"Task 9: The workflow module — permissions, definition reads, and wiring into every binary
Section titled “Task 9: The workflow module — permissions, definition reads, and wiring into every binary”Files:
- Create:
internal/modules/workflow/permissions.go,routes.go,decode.go,definitions.go,env_test.go,definitions_test.go - Modify (append-only):
constants.yaml; regenerateinternal/kernel/vocab/vocab.gen.goandpackages/contracts/vocab.ts - Modify (append-only):
api/openapi.yaml; regenerateinternal/gen/api/api.gen.go - Modify:
internal/app/app.go,internal/app/routes_test.go(composedModules),cmd/api/main.go,cmd/migrate/main.go,cmd/orgadmin/main.go,cmd/migrate/wiring_test.go,cmd/orgadmin/wiring_test.go
Why wiring is in this task and not a later one. The moment internal/modules/workflow exists, internal/app’s TestEveryBinaryLinksTheSameModules and TestEveryModuleContributesRoutesAndPermissions require every binary to link it and composedModules to name it; and the moment a route is registered, TestTheOpenAPIDocumentSaysWhatTheRouterEnforces requires it documented. Plan 2b’s lesson (its Task 9): a route and its declaration land in one commit.
Interfaces:
-
Consumes: Task 3 only (
ListKinds,Latest,LoadVersion,PublishGraph,Definition,InvalidError,ErrNoDefinition,ErrNoDraft,ErrRevisionConflict,ErrVersionConflict);httpx.Protected,httpx.IdentityFromContext,member.RegisterPermission,member.Permissions. -
Produces (Tasks 10 and 11 extend these files):
package workflow // internal/modules/workflow; imports the engine as `engine`const (PermissionDefinitionRead = "workflow.definition.read"PermissionDefinitionManage = "workflow.definition.manage"PermissionDefinitionPublish = "workflow.definition.publish"PermissionDelegationCreate = "workflow.delegation.create"PermissionDecisionRead = "workflow.decision.read")type Deps struct { DB *db.DB; Clock clock.Clock }func Register(g *gin.RouterGroup, deps Deps)type handlers struct{ deps Deps }func decodeStrict(c *gin.Context, v any) errorfunc actorOf(c *gin.Context) (uuid.UUID, error)func mapErr(err error) errortype definitionDTO struct { ID uuid.UUID; Kind string; Version int; PublishedAt time.Time; PublishedBy *uuid.UUID; Graph engine.Graph } -
Routes added here:
GET /v1/workflow/definitions(read),GET /v1/workflow/definitions/:kind(read, latest),GET /v1/workflow/definitions/:kind/versions/:version(read). -
Step 1: Append the permissions to
constants.yamland regenerate
Append under permissions: (after org.profile.manage; workflow.definition.publish is already present — do not duplicate it):
- workflow.definition.read - workflow.definition.manage - workflow.delegation.create - workflow.decision.readRun: go run ./tools/vocabgen && go test ./internal/kernel/vocab/ -count=1
Expected: PASS; vocab.gen.go and packages/contracts/vocab.ts gain the four keys.
- Step 2: Write the failing module test
internal/modules/workflow/env_test.go — the harness Tasks 10 and 11 reuse:
package workflow_test
import ( "bytes" "context" "encoding/json" "log/slog" "net/http" "net/http/httptest" "testing"
"github.com/gin-gonic/gin" "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/stretchr/testify/require"
"github.com/mcom/erp/internal/kernel/authn" "github.com/mcom/erp/internal/kernel/clock" "github.com/mcom/erp/internal/kernel/httpx" wfmod "github.com/mcom/erp/internal/modules/workflow" "github.com/mcom/erp/internal/platform/identity/token" engine "github.com/mcom/erp/internal/platform/workflow" "github.com/mcom/erp/internal/platform/workflow/workflowtest")
type alwaysLive struct{}
func (alwaysLive) Check(context.Context, authn.Claims) error { return nil }
type discard struct{}
func (discard) Write(p []byte) (int, error) { return len(p), nil }
type env struct { w *workflowtest.World r *gin.Engine signer *token.Signer}
// newEnv mounts this module alone on a router over the workflow test world.func newEnv(t *testing.T) *env { t.Helper() gin.SetMode(gin.TestMode) w := workflowtest.NewWorld(t) signer, err := token.NewSigner(map[string][]byte{"k1": []byte("01234567890123456789012345678901")}, "k1", clock.Fixed(workflowtest.T0)) require.NoError(t, err) r, api := httpx.NewRouter(httpx.Deps{DB: w.DB, Logger: slog.New(slog.NewTextHandler(discard{}, nil)), Version: "test", TokenVerifier: signer, Liveness: alwaysLive{}}) wfmod.Register(api, wfmod.Deps{DB: w.DB, Clock: w.Clock}) return &env{w: w, r: r, signer: signer}}
// member seeds a member holding exactly these permissions and returns a token.func (e *env) member(t *testing.T, name string, perms ...string) (uuid.UUID, string) { t.Helper() ctx := context.Background() m := e.w.Member(name) var role uuid.UUID require.NoError(t, e.w.Admin.QueryRow(ctx, `INSERT INTO role (org_id, name, system) VALUES ($1, $2, true) RETURNING id`, e.w.Org, "R-"+uuid.NewString()).Scan(&role)) for _, p := range perms { _, err := e.w.Admin.Exec(ctx, `INSERT INTO role_permission (org_id, role_id, permission_key) VALUES ($1,$2,$3)`, e.w.Org, role, p) require.NoError(t, err) } _, err := e.w.Admin.Exec(ctx, `INSERT INTO member_role (org_id, member_id, role_id) VALUES ($1,$2,$3)`, e.w.Org, m, role) require.NoError(t, err) tok, err := e.signer.Mint(authn.Claims{UserID: uuid.New(), OrgID: e.w.Org, MemberID: m, SessionID: uuid.New()}) require.NoError(t, err) return m, tok}
func (e *env) do(t *testing.T, method, path, tok string, body any) *httptest.ResponseRecorder { t.Helper() var buf bytes.Buffer if body != nil { require.NoError(t, json.NewEncoder(&buf).Encode(body)) } req := httptest.NewRequest(method, "/v1"+path, &buf) req.Header.Set("Authorization", "Bearer "+tok) req.Header.Set("Content-Type", "application/json") rec := httptest.NewRecorder() e.r.ServeHTTP(rec, req) return rec}
// smallGraph is the smallest flow the module tests need: draft -> pending// -> approved (accrual) -> done, rejection from pending, approved <- the// claimant's superior. Built here rather than borrowed from the engine's// seed so the module's tests do not depend on Task 8.func smallGraph() engine.Graph { l := func(s string) engine.Labels { return engine.Labels{"en": s} } return engine.Graph{ Kind: "expense_claim", PostingOrder: engine.AccrualFirst, States: []engine.StateSpec{ {Key: "draft", Semantic: "draft", DisplayOrder: 1, Labels: l("Draft")}, {Key: "pending", Semantic: "pending", DisplayOrder: 2, Labels: l("Pending")}, {Key: "approved", Semantic: "accrual", ControlKind: "ap", DisplayOrder: 3, Labels: l("Approved")}, {Key: "done", Semantic: "acknowledged", IsTerminal: true, DisplayOrder: 4, Labels: l("Done")}, {Key: "rejected", Semantic: "rejected", DisplayOrder: 5, Labels: l("Rejected")}, }, Edges: []engine.EdgeSpec{ {From: "draft", To: "pending", Gate: engine.GateReporter, ActionKey: "submit", Labels: l("Submit")}, {From: "pending", To: "approved", Gate: engine.GateTarget, ActionKey: "approve", Labels: l("Approve")}, {From: "approved", To: "done", Gate: engine.GateReporter, ActionKey: "acknowledge", Labels: l("Done")}, {From: "pending", To: "rejected", Gate: engine.GateAdvance, ActionKey: "reject", RequiresReason: true, Labels: l("Reject")}, {From: "rejected", To: "draft", Gate: engine.GateReporter, ActionKey: "revise", Labels: l("Revise")}, }, Grants: []engine.GrantSpec{{StateKey: "approved", GranteeKind: engine.GranteeSuperior}}, }}
func (e *env) publish(t *testing.T, g engine.Graph) engine.Definition { t.Helper() var d engine.Definition e.w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var err error d, err = engine.PublishGraph(ctx, tx, e.w.Clock, g, nil) return err }) return d}internal/modules/workflow/definitions_test.go:
package workflow_test
import ( "encoding/json" "net/http" "testing"
"github.com/stretchr/testify/require"
wfmod "github.com/mcom/erp/internal/modules/workflow" engine "github.com/mcom/erp/internal/platform/workflow")
func TestDefinitionsAreReadableWithTheReadPermissionOnly(t *testing.T) { e := newEnv(t) e.publish(t, smallGraph()) _, reader := e.member(t, "Reader", wfmod.PermissionDefinitionRead) _, nobody := e.member(t, "Nobody")
rec := e.do(t, http.MethodGet, "/workflow/definitions", reader, nil) require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) var list struct { Kinds []struct { Kind string `json:"kind"` LatestVersion int `json:"latest_version"` } `json:"kinds"` } require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &list)) require.Equal(t, "expense_claim", list.Kinds[0].Kind) require.Equal(t, 1, list.Kinds[0].LatestVersion)
rec = e.do(t, http.MethodGet, "/workflow/definitions/expense_claim", reader, nil) require.Equal(t, http.StatusOK, rec.Code) var def struct { Version int `json:"version"` Graph engine.Graph `json:"graph"` } require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &def)) require.Len(t, def.Graph.States, 5)
require.Equal(t, http.StatusOK, e.do(t, http.MethodGet, "/workflow/definitions/expense_claim/versions/1", reader, nil).Code) require.Equal(t, http.StatusNotFound, e.do(t, http.MethodGet, "/workflow/definitions/expense_claim/versions/2", reader, nil).Code) require.Equal(t, http.StatusNotFound, e.do(t, http.MethodGet, "/workflow/definitions/advance", reader, nil).Code)
// ABSENCE, after the presence above: without the permission, 403. require.Equal(t, http.StatusForbidden, e.do(t, http.MethodGet, "/workflow/definitions", nobody, nil).Code)}- Step 3: Run it and watch it fail
Run: go test ./internal/modules/workflow/ -count=1
Expected: FAIL — the package does not exist.
- Step 4: Write
permissions.go
package workflow
import "github.com/mcom/erp/internal/platform/identity/member"
// Five permissions, split along who needs them. Reading a flow is for every// approver and auditor; editing a draft and publishing it are separate// because publishing changes who may approve money (the "a claimant never// picks their own approver" invariant is only as strong as the set of people// who can publish grants); delegating one's own authority is for every// approver; reading a decision log is for auditors and finance.const ( PermissionDefinitionRead = "workflow.definition.read" PermissionDefinitionManage = "workflow.definition.manage" PermissionDefinitionPublish = "workflow.definition.publish" PermissionDelegationCreate = "workflow.delegation.create" PermissionDecisionRead = "workflow.decision.read")
func init() { member.RegisterPermission(PermissionDefinitionRead, "View workflow definitions and who would approve what") member.RegisterPermission(PermissionDefinitionManage, "Edit and discard draft workflow definitions") member.RegisterPermission(PermissionDefinitionPublish, "Publish a workflow definition as a new version") member.RegisterPermission(PermissionDelegationCreate, "Lend your own approval authority to a colleague for a period") member.RegisterPermission(PermissionDecisionRead, "Read a document's workflow state and decision log")}- Step 5: Write
routes.go
// Package workflow is the HTTP surface of the workflow engine: the// definition editor's API (Plan 10 builds the screen), delegations, and the// decision log. It is a thin shell over internal/platform/workflow, which// owns every rule. It exposes no transition endpoint: moving a document is// its owning module's route (Ruling 14).package workflow
import ( "context" "fmt"
"github.com/gin-gonic/gin" "github.com/google/uuid" "github.com/jackc/pgx/v5"
"github.com/mcom/erp/internal/kernel/clock" "github.com/mcom/erp/internal/kernel/db" apperr "github.com/mcom/erp/internal/kernel/errors" "github.com/mcom/erp/internal/kernel/httpx" "github.com/mcom/erp/internal/platform/identity/member")
type Deps struct { DB *db.DB Clock clock.Clock}
type handlers struct{ deps Deps }
func Register(g *gin.RouterGroup, deps Deps) { h := handlers{deps: deps} p := permissionLoader{db: deps.DB}
httpx.Protected(g, p, "GET", "/workflow/definitions", PermissionDefinitionRead, h.listDefinitions) httpx.Protected(g, p, "GET", "/workflow/definitions/:kind", PermissionDefinitionRead, h.latestDefinition) httpx.Protected(g, p, "GET", "/workflow/definitions/:kind/versions/:version", PermissionDefinitionRead, h.definitionVersion)}
// permissionLoader adapts member.Permissions to httpx.PermissionLoader.// Identical in shape to orgstructure's; not shared, because sharing it would// need a cross-module import for a few lines of glue.type permissionLoader struct{ db *db.DB }
var _ httpx.PermissionLoader = permissionLoader{}
func (p permissionLoader) Permissions(ctx context.Context, memberID uuid.UUID) (map[string]struct{}, error) { if p.db == nil { return nil, apperr.New(apperr.KindInternal, "authorization is not configured") } var out map[string]struct{} err := p.db.InTx(ctx, func(ctx context.Context, tx pgx.Tx) error { var err error out, err = member.Permissions(ctx, tx, memberID) return err }) if err != nil { return nil, fmt.Errorf("workflow: load permissions: %w", err) } return out, nil}- Step 6: Write
decode.go
package workflow
import ( "bytes" "encoding/json" "errors" "io" "strings"
"github.com/gin-gonic/gin" "github.com/google/uuid"
apperr "github.com/mcom/erp/internal/kernel/errors" "github.com/mcom/erp/internal/kernel/httpx" engine "github.com/mcom/erp/internal/platform/workflow")
// decodeStrict reads exactly one JSON object and refuses unknown fields: on// an editor save an ignored field is a silent no-op reported as success. A// local copy of orgstructure's rule, not an import of it (module boundary).func decodeStrict(c *gin.Context, v any) error { if c.Request.Body == nil { return apperr.New(apperr.KindBadRequest, "a JSON request body is required") } raw, err := io.ReadAll(c.Request.Body) if err != nil { return apperr.New(apperr.KindBadRequest, "the request body could not be read") } dec := json.NewDecoder(bytes.NewReader(raw)) dec.DisallowUnknownFields() if err := dec.Decode(v); err != nil { if errors.Is(err, io.EOF) { return apperr.New(apperr.KindBadRequest, "a JSON request body is required") } if msg := err.Error(); strings.HasPrefix(msg, "json: unknown field ") { return apperr.New(apperr.KindBadRequest, "field "+strings.TrimPrefix(msg, "json: unknown field ")+" is not accepted by this endpoint") } // money.Amount refuses a JSON number with its own message. return apperr.New(apperr.KindBadRequest, "the request body is not valid for this endpoint: "+err.Error()) } if _, err := dec.Token(); !errors.Is(err, io.EOF) { return apperr.New(apperr.KindBadRequest, "the request body must be a single JSON object") } return nil}
func actorOf(c *gin.Context) (uuid.UUID, error) { id, ok := httpx.IdentityFromContext(c.Request.Context()) if !ok { return uuid.Nil, apperr.New(apperr.KindUnauthorized, "no authenticated member") } return id.MemberID, nil}
// mapErr turns the engine's refusals into problems. Anything unlisted passes// to httpx.Errors, which maps Postgres errors and hides the rest as 500.// Task 11 appends the delegation and instance cases.func mapErr(err error) error { var inv *engine.InvalidError switch { case errors.As(err, &inv): return apperr.New(apperr.KindValidation, inv.Error()) case errors.Is(err, engine.ErrNoDefinition), errors.Is(err, engine.ErrNoDraft): return apperr.New(apperr.KindNotFound, "no such record in this organization") case errors.Is(err, engine.ErrRevisionConflict), errors.Is(err, engine.ErrVersionConflict): return apperr.New(apperr.KindConflict, err.Error()) } return err}- Step 7: Write
definitions.go
package workflow
import ( "context" "net/http" "strconv" "time"
"github.com/gin-gonic/gin" "github.com/google/uuid" "github.com/jackc/pgx/v5"
apperr "github.com/mcom/erp/internal/kernel/errors" engine "github.com/mcom/erp/internal/platform/workflow")
type definitionDTO struct { ID uuid.UUID `json:"id"` Kind string `json:"kind"` Version int `json:"version"` PublishedAt time.Time `json:"published_at"` PublishedBy *uuid.UUID `json:"published_by"` Graph engine.Graph `json:"graph"`}
func toDefinitionDTO(d engine.Definition) definitionDTO { return definitionDTO{ID: d.ID, Kind: d.Kind, Version: d.Version, PublishedAt: d.PublishedAt, PublishedBy: d.PublishedBy, Graph: d.Graph}}
type kindDTO struct { Kind string `json:"kind"` LatestVersion int `json:"latest_version"` DraftRevision int `json:"draft_revision"`}
func (h handlers) listDefinitions(c *gin.Context) { var kinds []engine.KindSummary if err := h.deps.DB.InTx(c.Request.Context(), func(ctx context.Context, tx pgx.Tx) error { var err error kinds, err = engine.ListKinds(ctx, tx) return err }); err != nil { _ = c.Error(mapErr(err)) return } out := make([]kindDTO, 0, len(kinds)) for _, k := range kinds { out = append(out, kindDTO{Kind: k.Kind, LatestVersion: k.LatestVersion, DraftRevision: k.DraftRevision}) } c.JSON(http.StatusOK, gin.H{"kinds": out})}
func (h handlers) latestDefinition(c *gin.Context) { h.serveDefinition(c, func(ctx context.Context, tx pgx.Tx) (engine.Definition, error) { return engine.Latest(ctx, tx, c.Param("kind")) })}
func (h handlers) definitionVersion(c *gin.Context) { v, err := strconv.Atoi(c.Param("version")) if err != nil || v < 1 { _ = c.Error(apperr.New(apperr.KindValidation, "version must be a positive integer")) return } h.serveDefinition(c, func(ctx context.Context, tx pgx.Tx) (engine.Definition, error) { return engine.LoadVersion(ctx, tx, c.Param("kind"), v) })}
func (h handlers) serveDefinition(c *gin.Context, load func(context.Context, pgx.Tx) (engine.Definition, error)) { var d engine.Definition if err := h.deps.DB.InTx(c.Request.Context(), func(ctx context.Context, tx pgx.Tx) error { var err error d, err = load(ctx, tx) return err }); err != nil { _ = c.Error(mapErr(err)) return } c.JSON(http.StatusOK, toDefinitionDTO(d))}- Step 8: Run the module test
Run: go test ./internal/modules/workflow/ -count=1
Expected: PASS.
- Step 9: Document the three routes in
api/openapi.yaml, then regenerate
Append these paths at the end of paths: (just before components:), and these schemas at the end of components.schemas. Tasks 10 and 11 append further entries after these. bearerAuth is inherited from the document default.
/workflow/definitions: get: operationId: listWorkflowDefinitions summary: Every workflow kind, its latest published version and its draft revision description: Requires workflow.definition.read. responses: '200': description: The kinds content: application/json: schema: $ref: '#/components/schemas/WorkflowKindList' '401': $ref: '#/components/responses/Unauthenticated' '403': $ref: '#/components/responses/WorkflowForbidden' /workflow/definitions/{kind}: get: operationId: getLatestWorkflowDefinition summary: The latest published version of a kind description: Requires workflow.definition.read. parameters: - $ref: '#/components/parameters/WorkflowKind' responses: '200': description: The definition content: application/json: schema: $ref: '#/components/schemas/WorkflowDefinition' '401': $ref: '#/components/responses/Unauthenticated' '403': $ref: '#/components/responses/WorkflowForbidden' '404': $ref: '#/components/responses/WorkflowNotFound' /workflow/definitions/{kind}/versions/{version}: get: operationId: getWorkflowDefinitionVersion summary: One published version of a kind description: > Published versions never change: an instance pinned to this version is decided by exactly this graph. Requires workflow.definition.read. parameters: - $ref: '#/components/parameters/WorkflowKind' - name: version in: path required: true schema: { type: integer, minimum: 1 } responses: '200': description: The definition content: application/json: schema: $ref: '#/components/schemas/WorkflowDefinition' '401': $ref: '#/components/responses/Unauthenticated' '403': $ref: '#/components/responses/WorkflowForbidden' '404': $ref: '#/components/responses/WorkflowNotFound' '422': description: version is not a positive integer. content: application/problem+json: schema: $ref: '#/components/schemas/Problem'Under components: add a parameters: map if the document has none (it has none today), and append to responses: and schemas::
parameters: WorkflowKind: name: kind in: path required: true schema: { type: string, pattern: '^[a-z][a-z0-9_]{0,62}$' } WorkflowForbidden: description: The caller does not hold the workflow permission this operation requires. content: application/problem+json: schema: $ref: '#/components/schemas/Problem' WorkflowNotFound: description: No such workflow record in this organization. content: application/problem+json: schema: $ref: '#/components/schemas/Problem' WorkflowKindList: type: object required: [kinds] properties: kinds: type: array items: type: object required: [kind, latest_version, draft_revision] properties: kind: { type: string } latest_version: { type: integer, description: 0 when never published } draft_revision: { type: integer, description: 0 when there is no draft } WorkflowLabels: type: object description: Locale to text. "en" is required; keys are supported UI locales. additionalProperties: { type: string, maxLength: 200 } WorkflowSemantic: type: string enum: [draft, pending, accrual, payout, acknowledged, dispute, rejected, cancelled, custom] WorkflowGate: type: string enum: [reporter, target, source, advance] WorkflowGranteeKind: type: string enum: [member, superior, position, unit_head, levels_up] WorkflowState: type: object required: [key, semantic, display_order, is_terminal, labels] properties: key: { type: string, pattern: '^[a-z][a-z0-9_]{0,62}$' } semantic: { $ref: '#/components/schemas/WorkflowSemantic' } control_kind: type: string enum: [ap, employee_advances, petty_cash_ap, ar, bank, fx_gain_loss] description: Required on, and only on, an accrual state. display_order: { type: integer } is_terminal: { type: boolean } escalate_after_days: { type: integer, minimum: 1, maximum: 365 } labels: { $ref: '#/components/schemas/WorkflowLabels' } WorkflowEdge: type: object required: [from, to, gate, action_key, requires_reason, reverses_accrual, display_order, labels] properties: from: { type: string } to: { type: string } gate: { $ref: '#/components/schemas/WorkflowGate' } action_key: { type: string, pattern: '^[a-z][a-z0-9_]{0,62}$' } requires_reason: { type: boolean } reverses_accrual: { type: boolean } display_order: { type: integer } labels: { $ref: '#/components/schemas/WorkflowLabels' } WorkflowGrant: type: object required: [all_states, grantee_kind] properties: state_key: { type: string, description: Absent exactly when all_states is true. } all_states: { type: boolean } grantee_kind: { $ref: '#/components/schemas/WorkflowGranteeKind' } member_id: { type: string, format: uuid } position_id: { type: string, format: uuid } levels_up: { type: integer, minimum: 1, maximum: 9 } min_amount: type: string pattern: '^[0-9]+(\.[0-9]{1,4})?$' description: The floor, as a decimal string. Absent means every amount. Floors only; a band's ceiling is the next larger floor on the same state. min_amount_currency: { type: string, pattern: '^[A-Z]{3}$', description: Must be the organization's base currency. } on_vacancy: { type: string, enum: [escalate, refuse] } max_escalation_levels: { type: integer, minimum: 1, maximum: 10 } WorkflowGraph: type: object required: [kind, posting_order, states, edges, grants] properties: kind: { type: string } posting_order: { type: string, enum: [accrual_first, payout_first] } states: { type: array, maxItems: 64, items: { $ref: '#/components/schemas/WorkflowState' } } edges: { type: array, maxItems: 256, items: { $ref: '#/components/schemas/WorkflowEdge' } } grants: { type: array, maxItems: 256, items: { $ref: '#/components/schemas/WorkflowGrant' } } WorkflowDefinition: type: object required: [id, kind, version, published_at, graph] properties: id: { type: string, format: uuid } kind: { type: string } version: { type: integer } published_at: { type: string, format: date-time } published_by: { type: string, format: uuid, nullable: true } graph: { $ref: '#/components/schemas/WorkflowGraph' }Run: go tool oapi-codegen -config oapi-codegen.yaml api/openapi.yaml && go test ./internal/gen/ -count=1
Expected: PASS.
- Step 10: Wire the module into the composition root and every binary
internal/app/app.go — append the import, the Deps field and the Register call after the existing ones:
"github.com/mcom/erp/internal/modules/workflow" Workflow workflow.Deps workflow.Register(api, deps.Workflow)internal/app/routes_test.go — append the import and the composedModules entry:
"workflow": {workflow.PermissionDefinitionRead, workflow.PermissionDefinitionManage, workflow.PermissionDefinitionPublish, workflow.PermissionDelegationCreate, workflow.PermissionDecisionRead},cmd/api/main.go — append the import and, inside app.Deps{…} after Activity:
Workflow: workflow.Deps{ DB: database, Clock: clock.System(), },cmd/migrate/main.go and cmd/orgadmin/main.go — append to the blank-import block:
_ "github.com/mcom/erp/internal/modules/workflow"cmd/migrate/wiring_test.go and cmd/orgadmin/wiring_test.go — append to wantModules and wantPermissions (literals, per those files’ own rule):
"github.com/mcom/erp/internal/modules/workflow", "workflow.definition.read", // modules/workflow "workflow.definition.publish", // modules/workflowRun: go build ./... && go test ./internal/app/ ./cmd/... ./internal/modules/workflow/ -count=1
Expected: PASS. TestTheOpenAPIDocumentSaysWhatTheRouterEnforces sees the three documented routes; TestEveryBinaryLinksTheSameModules sees four modules in every binary.
- Step 11: Watch the wiring guard fail once
Remove the _ ".../modules/workflow" line from cmd/migrate/main.go; go test ./internal/app/ -run TestEveryBinaryLinksTheSameModules -count=1 goes RED naming workflow. Restore.
- Step 12: Commit
git add constants.yaml internal/kernel/vocab/vocab.gen.go packages/contracts/vocab.ts \ api/openapi.yaml internal/gen/api/api.gen.go internal/modules/workflow/ \ internal/app/app.go internal/app/routes_test.go cmd/api/main.go cmd/migrate/ cmd/orgadmin/git commit -m "Add the workflow module with definition reads, wired into every binary
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>"Task 10: The editor’s API — drafts, live validation, publish, and the routing preview
Section titled “Task 10: The editor’s API — drafts, live validation, publish, and the routing preview”Files:
- Create:
internal/modules/workflow/drafts.go,internal/modules/workflow/drafts_test.go - Modify:
internal/modules/workflow/routes.go(append fivehttpx.Protectedlines),api/openapi.yaml(append), regenerateinternal/gen/api/api.gen.go
Interfaces:
- Consumes: Task 3 (
SaveDraft,GetDraft,DiscardDraft,Publish,Validate,Latest), Task 6 (Route,StepRouting), Task 9 (handlers,decodeStrict,actorOf,mapErr,toDefinitionDTO, the testenv,smallGraph). - Routes:
| Route | Permission | Result |
|---|---|---|
GET /v1/workflow/drafts/:kind |
workflow.definition.read |
200 draft + its current violations; 404 if none |
PUT /v1/workflow/drafts/:kind |
workflow.definition.manage |
body {revision, graph}; 200 draft + violations (an invalid draft saves); 409 stale revision |
DELETE /v1/workflow/drafts/:kind?revision=n |
workflow.definition.manage |
204; 409 stale; 404 none |
POST /v1/workflow/drafts/:kind/publish |
workflow.definition.publish |
body {revision}; 201 definition; 422 invalid; 409 stale |
GET /v1/workflow/definitions/:kind/routing?member_id=&amount= |
workflow.definition.read |
200 {version, steps: [StepRouting]} — “who would approve this member’s claim of this amount, and what would strand it” |
- Step 1: Write the failing tests
internal/modules/workflow/drafts_test.go:
package workflow_test
import ( "encoding/json" "net/http" "testing"
"github.com/stretchr/testify/require"
wfmod "github.com/mcom/erp/internal/modules/workflow" engine "github.com/mcom/erp/internal/platform/workflow")
type draftResp struct { Revision int `json:"revision"` Violations []engine.Violation `json:"violations"`}
func TestTheEditorSavesSeesViolationsLiveAndPublishes(t *testing.T) { e := newEnv(t) _, admin := e.member(t, "Admin", wfmod.PermissionDefinitionRead, wfmod.PermissionDefinitionManage, wfmod.PermissionDefinitionPublish)
broken := smallGraph() broken.Grants = nil rec := e.do(t, http.MethodPut, "/workflow/drafts/expense_claim", admin, map[string]any{"revision": 0, "graph": broken}) require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) var d draftResp require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &d)) require.Equal(t, 1, d.Revision) require.NotEmpty(t, d.Violations, "an invalid draft saves, and says why it cannot be published")
rec = e.do(t, http.MethodPost, "/workflow/drafts/expense_claim/publish", admin, map[string]any{"revision": 1}) require.Equal(t, http.StatusUnprocessableEntity, rec.Code, "an invalid draft does not publish")
rec = e.do(t, http.MethodPut, "/workflow/drafts/expense_claim", admin, map[string]any{"revision": 0, "graph": smallGraph()}) require.Equal(t, http.StatusConflict, rec.Code, "a stale revision is refused")
rec = e.do(t, http.MethodPut, "/workflow/drafts/expense_claim", admin, map[string]any{"revision": 1, "graph": smallGraph()}) require.Equal(t, http.StatusOK, rec.Code) require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &d)) require.Empty(t, d.Violations)
rec = e.do(t, http.MethodPost, "/workflow/drafts/expense_claim/publish", admin, map[string]any{"revision": 2}) require.Equal(t, http.StatusCreated, rec.Code, rec.Body.String()) require.Equal(t, http.StatusNotFound, e.do(t, http.MethodGet, "/workflow/drafts/expense_claim", admin, nil).Code, "publishing consumes the draft")}
func TestPublishingNeedsItsOwnPermission(t *testing.T) { e := newEnv(t) _, editor := e.member(t, "Editor", wfmod.PermissionDefinitionManage) _, publisher := e.member(t, "Publisher", wfmod.PermissionDefinitionPublish) require.Equal(t, http.StatusOK, e.do(t, http.MethodPut, "/workflow/drafts/expense_claim", editor, map[string]any{"revision": 0, "graph": smallGraph()}).Code) require.Equal(t, http.StatusForbidden, e.do(t, http.MethodPost, "/workflow/drafts/expense_claim/publish", editor, map[string]any{"revision": 1}).Code) require.Equal(t, http.StatusCreated, e.do(t, http.MethodPost, "/workflow/drafts/expense_claim/publish", publisher, map[string]any{"revision": 1}).Code, "PRESENCE: the same request succeeds with the permission")}
func TestAMoneyFloorSentAsANumberIsRefused(t *testing.T) { e := newEnv(t) _, admin := e.member(t, "Admin", wfmod.PermissionDefinitionManage) body := `{"revision":0,"graph":{"kind":"expense_claim","posting_order":"accrual_first","states":[],"edges":[],` + `"grants":[{"state_key":"approved","all_states":false,"grantee_kind":"superior","min_amount":2000}]}}` rec := e.doRaw(t, http.MethodPut, "/workflow/drafts/expense_claim", admin, body) require.Equal(t, http.StatusBadRequest, rec.Code, "money is a string on the wire (spec 20.1.9)")}
func TestTheRoutingPreviewNamesWhoWouldApproveAndWhatWouldStrand(t *testing.T) { e := newEnv(t) boss, staff, loner := e.w.Member("Boss"), e.w.Member("Staff"), e.w.Member("Loner") top := e.w.Seat("BOSS", nil) under := e.w.Seat("STAFF", &top) e.w.Sit(boss, top) e.w.Sit(staff, under) e.publish(t, smallGraph()) _, reader := e.member(t, "Reader", wfmod.PermissionDefinitionRead)
var resp struct { Steps []engine.StepRouting `json:"steps"` } rec := e.do(t, http.MethodGet, "/workflow/definitions/expense_claim/routing?member_id="+staff.String()+"&amount=100.00", reader, nil) require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &resp)) require.True(t, resp.Steps[0].Routable) require.Equal(t, boss, resp.Steps[0].Eligible[0])
rec = e.do(t, http.MethodGet, "/workflow/definitions/expense_claim/routing?member_id="+loner.String()+"&amount=100.00", reader, nil) require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &resp)) require.False(t, resp.Steps[0].Routable) require.Equal(t, "no_primary_position", resp.Steps[0].Gaps[0].Reason)
require.Equal(t, http.StatusUnprocessableEntity, e.do(t, http.MethodGet, "/workflow/definitions/expense_claim/routing?member_id="+staff.String()+"&amount=1e3", reader, nil).Code)}Append to env_test.go (Task 9’s harness) the raw-body helper this task needs:
func (e *env) doRaw(t *testing.T, method, path, tok, body string) *httptest.ResponseRecorder { t.Helper() req := httptest.NewRequest(method, "/v1"+path, strings.NewReader(body)) req.Header.Set("Authorization", "Bearer "+tok) req.Header.Set("Content-Type", "application/json") rec := httptest.NewRecorder() e.r.ServeHTTP(rec, req) return rec}(add "strings" to its imports).
- Step 2: Run them and watch them fail
Run: go test ./internal/modules/workflow/ -count=1
Expected: FAIL — 404s: the routes do not exist.
- Step 3: Write
drafts.go
package workflow
import ( "context" "net/http" "regexp" "strconv" "time"
"github.com/gin-gonic/gin" "github.com/google/uuid" "github.com/jackc/pgx/v5"
"github.com/mcom/erp/internal/kernel/clock" apperr "github.com/mcom/erp/internal/kernel/errors" "github.com/mcom/erp/internal/kernel/money" engine "github.com/mcom/erp/internal/platform/workflow")
type draftDTO struct { Kind string `json:"kind"` Revision int `json:"revision"` BaseVersion *int `json:"base_version"` Graph engine.Graph `json:"graph"` UpdatedAt time.Time `json:"updated_at"` UpdatedBy uuid.UUID `json:"updated_by"` Violations []engine.Violation `json:"violations"`}
func toDraftDTO(d engine.Draft, vs []engine.Violation) draftDTO { if vs == nil { vs = []engine.Violation{} } return draftDTO{Kind: d.Kind, Revision: d.Revision, BaseVersion: d.BaseVersion, Graph: d.Graph, UpdatedAt: d.UpdatedAt, UpdatedBy: d.UpdatedBy, Violations: vs}}
func (h handlers) getDraft(c *gin.Context) { var out draftDTO if err := h.deps.DB.InTx(c.Request.Context(), func(ctx context.Context, tx pgx.Tx) error { d, err := engine.GetDraft(ctx, tx, c.Param("kind")) if err != nil { return err } vs, err := engine.Validate(ctx, tx, d.Graph) out = toDraftDTO(d, vs) return err }); err != nil { _ = c.Error(mapErr(err)) return } c.JSON(http.StatusOK, out)}
type saveDraftBody struct { Revision int `json:"revision"` Graph engine.Graph `json:"graph"`}
func (h handlers) saveDraft(c *gin.Context) { var body saveDraftBody if err := decodeStrict(c, &body); err != nil { _ = c.Error(err) return } actor, err := actorOf(c) if err != nil { _ = c.Error(err) return } var out draftDTO if err := h.deps.DB.InTx(c.Request.Context(), func(ctx context.Context, tx pgx.Tx) error { d, vs, err := engine.SaveDraft(ctx, tx, h.deps.Clock, c.Param("kind"), body.Revision, body.Graph, actor) out = toDraftDTO(d, vs) return err }); err != nil { _ = c.Error(mapErr(err)) return } c.JSON(http.StatusOK, out)}
func (h handlers) discardDraft(c *gin.Context) { rev, err := strconv.Atoi(c.Query("revision")) if err != nil || rev < 1 { _ = c.Error(apperr.New(apperr.KindValidation, "revision must be the draft's current revision")) return } if err := h.deps.DB.InTx(c.Request.Context(), func(ctx context.Context, tx pgx.Tx) error { return engine.DiscardDraft(ctx, tx, h.deps.Clock, c.Param("kind"), rev) }); err != nil { _ = c.Error(mapErr(err)) return } c.Status(http.StatusNoContent)}
type publishBody struct { Revision int `json:"revision"`}
func (h handlers) publishDraft(c *gin.Context) { var body publishBody if err := decodeStrict(c, &body); err != nil { _ = c.Error(err) return } actor, err := actorOf(c) if err != nil { _ = c.Error(err) return } var d engine.Definition if err := h.deps.DB.InTx(c.Request.Context(), func(ctx context.Context, tx pgx.Tx) error { var err error d, err = engine.Publish(ctx, tx, h.deps.Clock, c.Param("kind"), body.Revision, actor) return err }); err != nil { _ = c.Error(mapErr(err)) return } c.JSON(http.StatusCreated, toDefinitionDTO(d))}
// decimalPattern is the wire form of an amount in a query string: a// non-negative decimal with at most four places. money.Parse alone would// accept "1e3", which is not a form any client should send.var decimalPattern = regexp.MustCompile(`^[0-9]+(\.[0-9]{1,4})?$`)
// routing serves the editor's and readiness's question: for this member and// amount, today, who decides each approval step, and what would strand it.func (h handlers) routing(c *gin.Context) { member, err := uuid.Parse(c.Query("member_id")) if err != nil { _ = c.Error(apperr.New(apperr.KindValidation, "member_id must be a UUID")) return } raw := c.Query("amount") if !decimalPattern.MatchString(raw) { _ = c.Error(apperr.New(apperr.KindValidation, "amount must be a non-negative decimal string")) return } amount := money.MustParse(raw) var def engine.Definition var steps []engine.StepRouting if err := h.deps.DB.InTx(c.Request.Context(), func(ctx context.Context, tx pgx.Tx) error { var err error if def, err = engine.Latest(ctx, tx, c.Param("kind")); err != nil { return err } steps, err = engine.Route(ctx, tx, def, member, amount, clock.TodayUTC(h.deps.Clock)) return err }); err != nil { _ = c.Error(mapErr(err)) return } c.JSON(http.StatusOK, gin.H{"version": def.Version, "steps": steps})}- Step 4: Register the routes
Append to Register in routes.go:
httpx.Protected(g, p, "GET", "/workflow/drafts/:kind", PermissionDefinitionRead, h.getDraft) httpx.Protected(g, p, "PUT", "/workflow/drafts/:kind", PermissionDefinitionManage, h.saveDraft) httpx.Protected(g, p, "DELETE", "/workflow/drafts/:kind", PermissionDefinitionManage, h.discardDraft) httpx.Protected(g, p, "POST", "/workflow/drafts/:kind/publish", PermissionDefinitionPublish, h.publishDraft) httpx.Protected(g, p, "GET", "/workflow/definitions/:kind/routing", PermissionDefinitionRead, h.routing)- Step 5: Run the module tests
Run: go test ./internal/modules/workflow/ -count=1
Expected: PASS.
- Step 6: Document the five routes, regenerate, run the app guards
Append to paths: in api/openapi.yaml:
/workflow/drafts/{kind}: parameters: - $ref: '#/components/parameters/WorkflowKind' get: operationId: getWorkflowDraft summary: The kind's draft and its current violations description: Requires workflow.definition.read. responses: '200': description: The draft content: application/json: schema: { $ref: '#/components/schemas/WorkflowDraft' } '401': { $ref: '#/components/responses/Unauthenticated' } '403': { $ref: '#/components/responses/WorkflowForbidden' } '404': { $ref: '#/components/responses/WorkflowNotFound' } put: operationId: saveWorkflowDraft summary: Replace the kind's draft description: > Saves even an invalid graph and returns its violations, so the editor can show them live (spec 5.7). revision is the draft's current revision, or 0 when there is none. Requires workflow.definition.manage. requestBody: required: true content: application/json: schema: type: object required: [revision, graph] additionalProperties: false properties: revision: { type: integer, minimum: 0 } graph: { $ref: '#/components/schemas/WorkflowGraph' } responses: '200': description: The saved draft content: application/json: schema: { $ref: '#/components/schemas/WorkflowDraft' } '400': { $ref: '#/components/responses/BadRequestBody' } '401': { $ref: '#/components/responses/Unauthenticated' } '403': { $ref: '#/components/responses/WorkflowForbidden' } '409': { $ref: '#/components/responses/WorkflowConflict' } delete: operationId: discardWorkflowDraft summary: Discard the kind's draft description: Requires workflow.definition.manage. parameters: - name: revision in: query required: true schema: { type: integer, minimum: 1 } responses: '204': { description: Discarded } '401': { $ref: '#/components/responses/Unauthenticated' } '403': { $ref: '#/components/responses/WorkflowForbidden' } '404': { $ref: '#/components/responses/WorkflowNotFound' } '409': { $ref: '#/components/responses/WorkflowConflict' } '422': { $ref: '#/components/responses/WorkflowInvalid' } /workflow/drafts/{kind}/publish: parameters: - $ref: '#/components/parameters/WorkflowKind' post: operationId: publishWorkflowDraft summary: Publish the draft as the kind's next version description: > Refused unless the draft passes every publish-time rule. In-flight documents stay on the version they started on. Requires workflow.definition.publish. requestBody: required: true content: application/json: schema: type: object required: [revision] additionalProperties: false properties: revision: { type: integer, minimum: 1 } responses: '201': description: The published version content: application/json: schema: { $ref: '#/components/schemas/WorkflowDefinition' } '400': { $ref: '#/components/responses/BadRequestBody' } '401': { $ref: '#/components/responses/Unauthenticated' } '403': { $ref: '#/components/responses/WorkflowForbidden' } '404': { $ref: '#/components/responses/WorkflowNotFound' } '409': { $ref: '#/components/responses/WorkflowConflict' } '422': { $ref: '#/components/responses/WorkflowInvalid' } /workflow/definitions/{kind}/routing: parameters: - $ref: '#/components/parameters/WorkflowKind' get: operationId: previewWorkflowRouting summary: Who would decide each approval step for this member and amount, today description: Requires workflow.definition.read. parameters: - { name: member_id, in: query, required: true, schema: { type: string, format: uuid } } - { name: amount, in: query, required: true, schema: { type: string, pattern: '^[0-9]+(\.[0-9]{1,4})?$' } } responses: '200': description: One entry per approval step content: application/json: schema: type: object required: [version, steps] properties: version: { type: integer } steps: { type: array, items: { $ref: '#/components/schemas/WorkflowStepRouting' } } '401': { $ref: '#/components/responses/Unauthenticated' } '403': { $ref: '#/components/responses/WorkflowForbidden' } '404': { $ref: '#/components/responses/WorkflowNotFound' } '422': { $ref: '#/components/responses/WorkflowInvalid' }Append to components.responses:
WorkflowConflict: description: The draft or version changed since the caller read it. content: application/problem+json: schema: { $ref: '#/components/schemas/Problem' } WorkflowInvalid: description: The request is well-formed but refused by a workflow rule. content: application/problem+json: schema: { $ref: '#/components/schemas/Problem' }Append to components.schemas:
WorkflowViolation: type: object required: [rule, code, detail] properties: rule: { type: integer, minimum: 0, maximum: 7, description: 0 is well-formedness; 1-7 are the publish-time rules. } code: { type: string } detail: { type: string } ref: { type: string } WorkflowDraft: type: object required: [kind, revision, graph, updated_at, updated_by, violations] properties: kind: { type: string } revision: { type: integer } base_version: { type: integer, nullable: true } graph: { $ref: '#/components/schemas/WorkflowGraph' } updated_at: { type: string, format: date-time } updated_by: { type: string, format: uuid } violations: { type: array, items: { $ref: '#/components/schemas/WorkflowViolation' } } WorkflowGap: type: object required: [grant_id, reason] properties: grant_id: { type: string, format: uuid } reason: { type: string } position_id: { type: string, format: uuid } member_id: { type: string, format: uuid } WorkflowResolvedGrant: type: object required: [grant_id, grantee_kind, floor, members, escalated] properties: grant_id: { type: string, format: uuid } grantee_kind: { $ref: '#/components/schemas/WorkflowGranteeKind' } floor: { type: string } members: { type: array, items: { type: string, format: uuid } } escalated: { type: boolean } gap: { $ref: '#/components/schemas/WorkflowGap' } WorkflowStepRouting: type: object required: [state, resolution, eligible, routable] properties: state: { type: string } resolution: { type: array, items: { $ref: '#/components/schemas/WorkflowResolvedGrant' } } winning_floor: { type: string } eligible: { type: array, items: { type: string, format: uuid } } gaps: { type: array, items: { $ref: '#/components/schemas/WorkflowGap' } } routable: { type: boolean }Run: go tool oapi-codegen -config oapi-codegen.yaml api/openapi.yaml && go test ./internal/gen/ ./internal/app/ ./internal/modules/workflow/ -count=1
Expected: PASS.
- Step 7: Commit
git add internal/modules/workflow/ api/openapi.yaml internal/gen/api/api.gen.gogit commit -m "Add the workflow editor API: drafts, publish and the routing preview
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>"Task 11: Delegations and the decision log over HTTP
Section titled “Task 11: Delegations and the decision log over HTTP”Files:
- Create:
internal/modules/workflow/delegations.go,internal/modules/workflow/instances.go,internal/modules/workflow/delegations_test.go,internal/modules/workflow/instances_test.go - Modify:
internal/modules/workflow/routes.go(append four lines),internal/modules/workflow/decode.go(append cases tomapErr),api/openapi.yaml(append), regenerateinternal/gen/api/api.gen.go
Interfaces:
- Consumes: Task 5 (
CreateDelegation,RevokeDelegation,ListDelegations, delegation errors), Task 7 (Get,AvailableTransitions,Decisions,Start,Transition,ErrNoInstance,RefusedError), Task 9/10 module files and test harness. - Routes:
| Route | Permission | Result |
|---|---|---|
GET /v1/workflow/delegations |
workflow.delegation.create |
200 the caller’s own, as delegator or delegate |
POST /v1/workflow/delegations |
workflow.delegation.create |
body {delegate_member_id, kind?, valid_from, valid_to?, reason}; delegator is the caller; 201 |
DELETE /v1/workflow/delegations/:id |
workflow.delegation.create |
204; 404 unless the caller is its delegator; 409 already revoked |
GET /v1/workflow/instances/:id |
workflow.decision.read |
200 {instance, available_transitions, decisions} |
- Step 1: Write the failing tests
internal/modules/workflow/delegations_test.go:
package workflow_test
import ( "encoding/json" "net/http" "testing"
"github.com/stretchr/testify/require"
wfmod "github.com/mcom/erp/internal/modules/workflow")
func TestAMemberLendsAndRevokesTheirOwnAuthority(t *testing.T) { e := newEnv(t) _, boss := e.member(t, "Boss", wfmod.PermissionDelegationCreate) deputy, deputyTok := e.member(t, "Deputy", wfmod.PermissionDelegationCreate) _, nobody := e.member(t, "Nobody")
rec := e.do(t, http.MethodPost, "/workflow/delegations", boss, map[string]any{ "delegate_member_id": deputy, "valid_from": "2026-09-21", "valid_to": "2026-09-28", "reason": "annual leave"}) require.Equal(t, http.StatusCreated, rec.Code, rec.Body.String()) var created struct { ID string `json:"id"` } require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &created))
var list struct { Delegations []map[string]any `json:"delegations"` } require.NoError(t, json.Unmarshal(e.do(t, http.MethodGet, "/workflow/delegations", deputyTok, nil).Body.Bytes(), &list)) require.Len(t, list.Delegations, 1, "the delegate sees what was lent to them")
require.Equal(t, http.StatusNotFound, e.do(t, http.MethodDelete, "/workflow/delegations/"+created.ID, deputyTok, nil).Code, "only the delegator revokes") require.Equal(t, http.StatusNoContent, e.do(t, http.MethodDelete, "/workflow/delegations/"+created.ID, boss, nil).Code) require.Equal(t, http.StatusConflict, e.do(t, http.MethodDelete, "/workflow/delegations/"+created.ID, boss, nil).Code)
require.Equal(t, http.StatusUnprocessableEntity, e.do(t, http.MethodPost, "/workflow/delegations", boss, map[string]any{ "delegate_member_id": deputy, "valid_from": "2026-09-01", "reason": "backdated"}).Code) require.Equal(t, http.StatusForbidden, e.do(t, http.MethodGet, "/workflow/delegations", nobody, nil).Code)}internal/modules/workflow/instances_test.go:
package workflow_test
import ( "context" "encoding/json" "net/http" "testing"
"github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/stretchr/testify/require"
"github.com/mcom/erp/internal/kernel/money" wfmod "github.com/mcom/erp/internal/modules/workflow" engine "github.com/mcom/erp/internal/platform/workflow")
func TestAnAuditorReadsTheStateTheLogAndWhyTheyCannotAct(t *testing.T) { e := newEnv(t) boss, staff := e.w.Member("Boss"), e.w.Member("Staff") top := e.w.Seat("BOSS", nil) under := e.w.Seat("STAFF", &top) e.w.Sit(boss, top) e.w.Sit(staff, under) e.publish(t, smallGraph()) _, auditor := e.member(t, "Auditor", wfmod.PermissionDecisionRead) _, nobody := e.member(t, "Nobody")
var inst engine.Instance e.w.MustTx(func(ctx context.Context, tx pgx.Tx) error { var err error if inst, err = engine.Start(ctx, tx, e.w.Clock, engine.StartInput{Kind: "expense_claim", SourceID: uuid.New(), Reporter: staff, Amount: money.MustParse("75.00")}); err != nil { return err } _, err = engine.Transition(ctx, tx, e.w.Clock, engine.TransitionInput{InstanceID: inst.ID, From: "draft", To: "pending", Actor: staff}) return err })
rec := e.do(t, http.MethodGet, "/workflow/instances/"+inst.ID.String(), auditor, nil) require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) var got struct { Instance struct { State string `json:"state"` Amount string `json:"amount"` } `json:"instance"` Available []engine.Available `json:"available_transitions"` Decisions []struct { To string `json:"to"` Eligible []uuid.UUID `json:"eligible_member_ids"` } `json:"decisions"` } require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &got)) require.Equal(t, "pending", got.Instance.State) require.Equal(t, "75.0000", got.Instance.Amount, "money is a string on the wire") require.Len(t, got.Decisions, 1) require.Equal(t, []uuid.UUID{staff}, got.Decisions[0].Eligible) require.NotEmpty(t, got.Available) for _, a := range got.Available { require.False(t, a.Allowed, "an auditor reads; they are in no band") require.NotNil(t, a.Refusal) }
require.Equal(t, http.StatusNotFound, e.do(t, http.MethodGet, "/workflow/instances/"+uuid.NewString(), auditor, nil).Code) require.Equal(t, http.StatusForbidden, e.do(t, http.MethodGet, "/workflow/instances/"+inst.ID.String(), nobody, nil).Code)}These tests dispatch no ledger event (submit and read only), so no fake Poster is needed.
- Step 2: Run them and watch them fail
Run: go test ./internal/modules/workflow/ -count=1
Expected: FAIL — 404s.
- Step 3: Write
delegations.go
package workflow
import ( "context" "net/http" "time"
"github.com/gin-gonic/gin" "github.com/google/uuid" "github.com/jackc/pgx/v5"
apperr "github.com/mcom/erp/internal/kernel/errors" engine "github.com/mcom/erp/internal/platform/workflow")
const dateLayout = "2006-01-02"
type delegationDTO struct { ID uuid.UUID `json:"id"` DelegatorID uuid.UUID `json:"delegator_member_id"` DelegateID uuid.UUID `json:"delegate_member_id"` Kind *string `json:"kind"` ValidFrom string `json:"valid_from"` ValidTo *string `json:"valid_to"` Reason string `json:"reason"` CreatedAt time.Time `json:"created_at"` RevokedAt *time.Time `json:"revoked_at"`}
func toDelegationDTO(d engine.Delegation) delegationDTO { out := delegationDTO{ID: d.ID, DelegatorID: d.DelegatorMemberID, DelegateID: d.DelegateMemberID, Kind: d.Kind, ValidFrom: d.ValidFrom.Format(dateLayout), Reason: d.Reason, CreatedAt: d.CreatedAt, RevokedAt: d.RevokedAt} if d.ValidTo != nil { s := d.ValidTo.Format(dateLayout) out.ValidTo = &s } return out}
func (h handlers) listDelegations(c *gin.Context) { actor, err := actorOf(c) if err != nil { _ = c.Error(err) return } var ds []engine.Delegation if err := h.deps.DB.InTx(c.Request.Context(), func(ctx context.Context, tx pgx.Tx) error { var err error ds, err = engine.ListDelegations(ctx, tx, actor) return err }); err != nil { _ = c.Error(mapErr(err)) return } out := make([]delegationDTO, 0, len(ds)) for _, d := range ds { out = append(out, toDelegationDTO(d)) } c.JSON(http.StatusOK, gin.H{"delegations": out})}
type createDelegationBody struct { DelegateMemberID uuid.UUID `json:"delegate_member_id"` Kind *string `json:"kind"` ValidFrom string `json:"valid_from"` ValidTo *string `json:"valid_to"` Reason string `json:"reason"`}
func (h handlers) createDelegation(c *gin.Context) { var body createDelegationBody if err := decodeStrict(c, &body); err != nil { _ = c.Error(err) return } actor, err := actorOf(c) if err != nil { _ = c.Error(err) return } from, err := time.Parse(dateLayout, body.ValidFrom) if err != nil { _ = c.Error(apperr.New(apperr.KindValidation, "valid_from must be a date in YYYY-MM-DD form")) return } in := engine.DelegationInput{Delegator: actor, Delegate: body.DelegateMemberID, Kind: body.Kind, ValidFrom: from, Reason: body.Reason} if body.ValidTo != nil { to, err := time.Parse(dateLayout, *body.ValidTo) if err != nil { _ = c.Error(apperr.New(apperr.KindValidation, "valid_to must be a date in YYYY-MM-DD form")) return } in.ValidTo = &to } var d engine.Delegation if err := h.deps.DB.InTx(c.Request.Context(), func(ctx context.Context, tx pgx.Tx) error { var err error d, err = engine.CreateDelegation(ctx, tx, h.deps.Clock, in) return err }); err != nil { _ = c.Error(mapErr(err)) return } c.JSON(http.StatusCreated, toDelegationDTO(d))}
func (h handlers) revokeDelegation(c *gin.Context) { id, err := uuid.Parse(c.Param("id")) if err != nil { _ = c.Error(apperr.New(apperr.KindValidation, "the delegation id must be a UUID")) return } actor, err := actorOf(c) if err != nil { _ = c.Error(err) return } if err := h.deps.DB.InTx(c.Request.Context(), func(ctx context.Context, tx pgx.Tx) error { return engine.RevokeDelegation(ctx, tx, h.deps.Clock, id, actor) }); err != nil { _ = c.Error(mapErr(err)) return } c.Status(http.StatusNoContent)}- Step 4: Write
instances.go
package workflow
import ( "context" "encoding/json" "net/http" "time"
"github.com/gin-gonic/gin" "github.com/google/uuid" "github.com/jackc/pgx/v5"
apperr "github.com/mcom/erp/internal/kernel/errors" engine "github.com/mcom/erp/internal/platform/workflow")
type instanceDTO struct { ID uuid.UUID `json:"id"` Kind string `json:"kind"` Version int `json:"version"` SourceID uuid.UUID `json:"source_id"` Reporter uuid.UUID `json:"reporter_member_id"` State string `json:"state"` Amount string `json:"amount"` Currency string `json:"currency"` EnteredStateAt time.Time `json:"entered_state_at"`}
type decisionDTO struct { ID uuid.UUID `json:"id"` From string `json:"from"` To string `json:"to"` Gate engine.Gate `json:"gate"` ActionKey string `json:"action_key"` Actor uuid.UUID `json:"actor_member_id"` AuthorityMember uuid.UUID `json:"authority_member_id"` Via engine.Via `json:"authority_via"` DelegationID *uuid.UUID `json:"delegation_id"` GrantID *uuid.UUID `json:"grant_id"` Amount string `json:"amount"` Currency string `json:"currency"` WinningFloor *string `json:"winning_floor"` ResolvedFor string `json:"resolved_for"` Eligible []uuid.UUID `json:"eligible_member_ids"` Resolution json.RawMessage `json:"resolution"` Reason string `json:"reason,omitempty"` DecidedAt time.Time `json:"decided_at"`}
func (h handlers) getInstance(c *gin.Context) { id, err := uuid.Parse(c.Param("id")) if err != nil { _ = c.Error(apperr.New(apperr.KindValidation, "the instance id must be a UUID")) return } actor, err := actorOf(c) if err != nil { _ = c.Error(err) return } var inst engine.Instance var av []engine.Available var ds []engine.Decision if err := h.deps.DB.InTx(c.Request.Context(), func(ctx context.Context, tx pgx.Tx) error { var err error if inst, err = engine.Get(ctx, tx, id); err != nil { return err } if av, err = engine.AvailableTransitions(ctx, tx, h.deps.Clock, id, actor); err != nil { return err } ds, err = engine.Decisions(ctx, tx, id) return err }); err != nil { _ = c.Error(mapErr(err)) return } out := make([]decisionDTO, 0, len(ds)) for _, d := range ds { dto := decisionDTO{ID: d.ID, From: d.From, To: d.To, Gate: d.Gate, ActionKey: d.ActionKey, Actor: d.Actor, AuthorityMember: d.AuthorityMember, Via: d.Via, DelegationID: d.DelegationID, GrantID: d.GrantID, Amount: d.Amount.String(), Currency: d.Currency, ResolvedFor: d.ResolvedFor.Format(dateLayout), Eligible: d.Eligible, Resolution: d.Resolution, Reason: d.Reason, DecidedAt: d.DecidedAt} if d.WinningFloor != nil { s := d.WinningFloor.String() dto.WinningFloor = &s } out = append(out, dto) } c.JSON(http.StatusOK, gin.H{ "instance": instanceDTO{ID: inst.ID, Kind: inst.Kind, Version: inst.Version, SourceID: inst.SourceID, Reporter: inst.Reporter, State: inst.StateKey, Amount: inst.Amount.String(), Currency: inst.Currency, EnteredStateAt: inst.EnteredStateAt}, "available_transitions": av, "decisions": out, })}- Step 5: Register the routes and extend
mapErr
Append to Register:
httpx.Protected(g, p, "GET", "/workflow/delegations", PermissionDelegationCreate, h.listDelegations) httpx.Protected(g, p, "POST", "/workflow/delegations", PermissionDelegationCreate, h.createDelegation) httpx.Protected(g, p, "DELETE", "/workflow/delegations/:id", PermissionDelegationCreate, h.revokeDelegation) httpx.Protected(g, p, "GET", "/workflow/instances/:id", PermissionDecisionRead, h.getInstance)In decode.go, add these cases to mapErr’s switch, before its closing brace:
case errors.Is(err, engine.ErrNoInstance), errors.Is(err, engine.ErrNoDelegation): return apperr.New(apperr.KindNotFound, "no such record in this organization") case errors.Is(err, engine.ErrAlreadyRevoked): return apperr.New(apperr.KindConflict, err.Error()) case errors.Is(err, engine.ErrSelfDelegation), errors.Is(err, engine.ErrBackdated), errors.Is(err, engine.ErrBadWindow), errors.Is(err, engine.ErrDelegateInactive): return apperr.New(apperr.KindValidation, err.Error())- Step 6: Document the four routes, regenerate, run the guards
Append to paths::
/workflow/delegations: get: operationId: listWorkflowDelegations summary: Delegations the caller gave or received description: Requires workflow.delegation.create. responses: '200': description: The delegations, newest first content: application/json: schema: type: object required: [delegations] properties: delegations: { type: array, items: { $ref: '#/components/schemas/WorkflowDelegation' } } '401': { $ref: '#/components/responses/Unauthenticated' } '403': { $ref: '#/components/responses/WorkflowForbidden' } post: operationId: createWorkflowDelegation summary: Lend the caller's own approval authority for a period description: > The delegate acts on the caller's thresholds, and every decision taken under the delegation names both (spec 17.4). Dated from today onward; never retroactive. Requires workflow.delegation.create. requestBody: required: true content: application/json: schema: type: object required: [delegate_member_id, valid_from, reason] additionalProperties: false properties: delegate_member_id: { type: string, format: uuid } kind: { type: string, nullable: true } valid_from: { type: string, format: date } valid_to: { type: string, format: date, nullable: true } reason: { type: string, minLength: 1 } responses: '201': description: The delegation content: application/json: schema: { $ref: '#/components/schemas/WorkflowDelegation' } '400': { $ref: '#/components/responses/BadRequestBody' } '401': { $ref: '#/components/responses/Unauthenticated' } '403': { $ref: '#/components/responses/WorkflowForbidden' } '422': { $ref: '#/components/responses/WorkflowInvalid' } /workflow/delegations/{id}: delete: operationId: revokeWorkflowDelegation summary: Revoke one of the caller's delegations description: Stamps revoked_at; nothing is deleted. Requires workflow.delegation.create. parameters: - { name: id, in: path, required: true, schema: { type: string, format: uuid } } responses: '204': { description: Revoked } '401': { $ref: '#/components/responses/Unauthenticated' } '403': { $ref: '#/components/responses/WorkflowForbidden' } '404': { $ref: '#/components/responses/WorkflowNotFound' } '409': { $ref: '#/components/responses/WorkflowConflict' } /workflow/instances/{id}: get: operationId: getWorkflowInstance summary: A document's workflow state, the caller's available transitions, and its decision log description: > The decision log is returned exactly as written at decision time, including the approver set resolved then; it is never re-resolved. Requires workflow.decision.read. parameters: - { name: id, in: path, required: true, schema: { type: string, format: uuid } } responses: '200': description: The instance content: application/json: schema: { $ref: '#/components/schemas/WorkflowInstanceView' } '401': { $ref: '#/components/responses/Unauthenticated' } '403': { $ref: '#/components/responses/WorkflowForbidden' } '404': { $ref: '#/components/responses/WorkflowNotFound' }Append to components.schemas:
WorkflowDelegation: type: object required: [id, delegator_member_id, delegate_member_id, valid_from, reason, created_at] properties: id: { type: string, format: uuid } delegator_member_id: { type: string, format: uuid } delegate_member_id: { type: string, format: uuid } kind: { type: string, nullable: true } valid_from: { type: string, format: date } valid_to: { type: string, format: date, nullable: true } reason: { type: string } created_at: { type: string, format: date-time } revoked_at: { type: string, format: date-time, nullable: true } WorkflowRefusal: type: object required: [code, detail] properties: code: { type: string } detail: { type: string } gaps: { type: array, items: { $ref: '#/components/schemas/WorkflowGap' } } WorkflowAvailableTransition: type: object required: [to, action_key, gate, requires_reason, allowed, labels] properties: to: { type: string } action_key: { type: string } gate: { $ref: '#/components/schemas/WorkflowGate' } requires_reason: { type: boolean } allowed: { type: boolean } labels: { $ref: '#/components/schemas/WorkflowLabels' } refusal: { $ref: '#/components/schemas/WorkflowRefusal' } WorkflowDecision: type: object required: [id, from, to, gate, action_key, actor_member_id, authority_member_id, authority_via, amount, currency, resolved_for, eligible_member_ids, resolution, decided_at] properties: id: { type: string, format: uuid } from: { type: string } to: { type: string } gate: { $ref: '#/components/schemas/WorkflowGate' } action_key: { type: string } actor_member_id: { type: string, format: uuid } authority_member_id: { type: string, format: uuid } authority_via: { type: string, enum: [reporter, grant, delegation, escalation] } delegation_id: { type: string, format: uuid, nullable: true } grant_id: { type: string, format: uuid, nullable: true } amount: { type: string } currency: { type: string } winning_floor: { type: string, nullable: true } resolved_for: { type: string, format: date } eligible_member_ids: { type: array, items: { type: string, format: uuid } } resolution: { type: array, items: { $ref: '#/components/schemas/WorkflowResolvedGrant' } } reason: { type: string } decided_at: { type: string, format: date-time } WorkflowInstanceView: type: object required: [instance, available_transitions, decisions] properties: instance: type: object required: [id, kind, version, source_id, reporter_member_id, state, amount, currency, entered_state_at] properties: id: { type: string, format: uuid } kind: { type: string } version: { type: integer } source_id: { type: string, format: uuid } reporter_member_id: { type: string, format: uuid } state: { type: string } amount: { type: string } currency: { type: string } entered_state_at: { type: string, format: date-time } available_transitions: { type: array, items: { $ref: '#/components/schemas/WorkflowAvailableTransition' } } decisions: { type: array, items: { $ref: '#/components/schemas/WorkflowDecision' } }Run: go tool oapi-codegen -config oapi-codegen.yaml api/openapi.yaml && go test ./internal/gen/ ./internal/app/ ./internal/modules/workflow/ -count=1
Expected: PASS.
- Step 7: Commit
git add internal/modules/workflow/ api/openapi.yaml internal/gen/api/api.gen.gogit commit -m "Add delegation and decision-log endpoints to the workflow module
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>"Task 12: Final guards, roadmap, and the one full run
Section titled “Task 12: Final guards, roadmap, and the one full run”Files:
- Modify:
cmd/migrate/wiring_test.go,cmd/orgadmin/wiring_test.go(one test each),docs/ROADMAP.md(Plan 3’s lines only)
Interfaces: consumes everything; produces nothing new.
- Step 1: Assert every provisioning binary sees step 60
A registry filled by init() is only as complete as the binary’s import graph (spec 2b §2.3). Append to each of cmd/migrate/wiring_test.go and cmd/orgadmin/wiring_test.go (both package main; importing internal/platform/provisioning is allowed — it is not the package that registers step 60, so the test cannot satisfy itself):
// TestProvisioningSeesTheWorkflowStep: step 60 is registered by// internal/platform/workflow's init(), which this binary reaches only// through its blank import of internal/modules/workflow. Without it,// provisioning would record an organization complete with no workflow// definition, and its first claim would fail at Start.func TestProvisioningSeesTheWorkflowStep(t *testing.T) { var keys []string for _, s := range provisioning.Steps() { keys = append(keys, s.Key()) } require.Contains(t, keys, "org_structure", "PRESENCE: the registry is populated at all") require.Contains(t, keys, "workflow_definitions")}Add "github.com/mcom/erp/internal/platform/provisioning" to each file’s imports.
Run: go test ./cmd/migrate/ ./cmd/orgadmin/ -run TestProvisioningSeesTheWorkflowStep -count=1
Expected: PASS. Then remove _ ".../modules/workflow" from cmd/migrate/main.go, re-run, confirm RED, restore.
- Step 2: Tick Plan 3 in
docs/ROADMAP.md
Under - [ ] **Plan 3 — Workflow engine**, add the plan link as its first sub-line and tick the eight items (touch no other plan’s lines):
Plan: [`2026-09-21-plan-3-workflow-engine.md`](superpowers/plans/2026-09-21-plan-3-workflow-engine.md) (12 tasks) Existing organizations get the seeded flow by re-running provisioning (`POST /v1/org/provision`); step 60 is new. - [x] Definitions, states, edges, grants, version pinning (§5.2) - [x] Gates: `reporter` / `target` / `source` / `advance` (§5.3) - [x] Thresholds on grants, floors only, tiling (§5.4) - [x] Grantee kinds incl. `superior` and its resolution chain (§5.4) - [x] Semantic roles as the ledger contract (§5.5) - [x] Publish-time validator, all seven rules (§5.7) - [x] Delegation, escalation, dispute, immutable decision log (§5.8) - [x] Seed the korre default flow as v1 (§6.3) - [ ] **Seam follow-up (Ruling 17):** `posting.Event` cannot say "past payout", so an upheld dispute after payment reaches the Poster as `rejected` + `PastAccrual`. Raise on `phase1-wave1`; not changed by Plan 3. - [ ] Role sets and quorum (§5.8) — deferred by Ruling 9.Leave the **Plan 3 — Workflow engine** checkbox itself unticked until the user merges.
- Step 3: Format, vet, and the one full run
Run: gofmt -l . && go vet ./... && go test ./... -count=1
Expected: no gofmt output; vet clean; every package PASS. This is the only full run in the plan (house rule). If a package outside Plan 3’s paths fails, stop and report it — do not fix another plan’s code.
- Step 4: Commit
git add cmd/migrate/wiring_test.go cmd/orgadmin/wiring_test.go docs/ROADMAP.mdgit commit -m "Assert every provisioning binary sees the workflow step; tick Plan 3
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>"Parallel groups
Section titled “Parallel groups”Tasks in the same group touch disjoint files and may run at the same time, each in its own worktree; groups run in order. Within internal/platform/workflow, parallel tasks add separate files to one package and prefix their test helpers (st, rs, dl, az, in, sd) so the merged test binary has no collisions.
| Group | Tasks | Needs merged first | Files (disjoint within the group) |
|---|---|---|---|
| A | 1, 2 | — | 1: db/migrations/0003[0-3]_*, internal/kernel/db/retention_test.go, workflow/workflowtest/. 2: workflow/model.go, validate.go, sqlhelp.go, validate_test.go |
| B | 3, 4, 5 | A | 3: store.go, store_test.go. 4: resolve.go, resolve_test.go. 5: delegation.go, delegation_test.go |
| C | 6, 9 | B | 6: authorize.go, ladder_test.go, authorize_test.go. 9: internal/modules/workflow/{permissions,routes,decode,definitions}.go + tests, constants.yaml + vocab gen, api/openapi.yaml + api gen, internal/app/{app.go,routes_test.go}, cmd/{api,migrate,orgadmin}/ |
| D | 7, 10 | C | 7: instance.go, instance_test.go. 10: internal/modules/workflow/drafts.go, drafts_test.go, appends to routes.go, env_test.go, api/openapi.yaml |
| E | 8, 11 | D | 8: seed.go, seed_test.go. 11: internal/modules/workflow/{delegations,instances}.go + tests, appends to routes.go, decode.go, api/openapi.yaml |
| F | 12 | E | cmd/*/wiring_test.go, docs/ROADMAP.md; full suite |
Dependencies behind the grouping: 6 uses 4’s rsNewChart/Resolve and 5’s dlCreate/ActiveDelegationsTo; 7 uses 3’s PublishGraph/stPublish and 6’s Authorize/Route; 8 uses 7’s inEnv/inPoster; 10 uses 6’s Route and 9’s harness; 11 uses 5, 7 and 10’s routes.go. Tasks 9, 10 and 11 each edit api/openapi.yaml and regenerate api.gen.go, which is why they sit in different groups.
Self-Review Notes
Section titled “Self-Review Notes”- Spec coverage. §5.2 data model and version pinning → Tasks 1, 3, 7 (
TestAnInstanceStaysPinnedToTheVersionItStartedOn). §5.3 gates → Tasks 2 (GateState), 6. §5.4 thresholds, floors only, half-open, no fallback, resolve-before-floor → Task 6 ladder tests; grantee kinds and the superior chain, vacancy, self, primary position → Task 4; refusal-not-silence at submission → Tasks 6 (Route), 7; convert-once-at-save → Ruling 8, Task 2currency_mismatch. §5.5 semantics as the ledger contract → Task 7 (posting.Dispatchevents), Task 1 CHECKs. §5.6 tiers → the validator enforces tier 2; tier 3 is “support-assisted” and needs no code (an admin with publish rights can publish a valid semantic change; the UI block is Plan 10’s). §5.7 rules 1–7 → Task 2, one test each. §5.8 delegation → 5, 6; escalation → 6 (Ruling 10); return-for-correction →rejected → draftin the seed; dispute with mandatory reason → Rulings 7, 16, Task 8 e2e; immutable decision log → Tasks 1, 7; multi-level chains and amount bands → 6; role sets and quorum → deferred, Ruling 9, carried on the roadmap. §6.3 seed → Task 8. §17.4 → Task 6 delegation test asserts actor ≠ authority. §20.1.1 item 5 →TestEverySeededDefinitionPassesTheValidator. §20.2 step 7, idempotence, readiness warnings → Task 8. Wave-1 contract: ranges, paths, append-only shared files,posting.Dispatchin every transition, no edit toposting→ Global Constraints, Tasks 7, 9–12. Org-structure spec §5.5 (persist the resolved set) → Task 7TestTheDecisionLogKeepsTheApproverSetWhenTheChartLaterChanges. - Placeholder scan. Every code step carries its code; no step says “similar to” another task. Task-local helpers that a later task reuses (
valid,rsNewChart,dlCreate,stPublish,inEnv, the moduleenv) are named in that later task’s Interfaces block. - Type consistency.
Graph/StateSpec/EdgeSpec/GrantSpec/Definition/Grant/Violation(Task 2) are used unchanged by 3–11.Resolved/Gap(4) →Ladder/Route/Authorize(6) →Decision.Resolution(7) and the OpenAPIWorkflowResolvedGrant(10).Viavaluesreporter|grant|delegation|escalationmatchworkflow_decision.authority_via’s CHECK (Task 1) and the OpenAPI enum (11).orgOf/idSetlive in Task 2’ssqlhelp.goand nowhere else.workflowtest.Todayis theonevery test passes;T0is only the clock’s instant. - Known sharp edge for implementers.
postingis process-global: Task 7’s and Task 8’s tests register a fakePosterand reset it on cleanup; no test in the package may callt.Parallel().