Skip to content

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):

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/… and internal/modules/workflow/…. Migrations only 00030–00049, and this plan uses exactly 00030–00033. Provisioning step orders only 60–69; this plan uses 60. Permission keys only workflow.*. HTTP paths only /v1/workflow/…. Never edit internal/platform/posting. Never edit internal/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 in cmd/api, cmd/migrate, cmd/orgadmin, the lists in internal/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, ENABLE and FORCE ROW LEVEL SECURITY, policy named org_isolation with USING and WITH CHECK on current_setting('app.current_org', true)::uuid. Every SQL statement naming an org-scoped table carries an explicit org_id = $n predicate on each org-scoped table it names (directly or through a.org_id = b.org_id join equalities); every INSERT names org_id. internal/kernel/layering/tenantpredicate_test.go fails the build otherwise.
  • Composite tenant FKs. Every FK between two org-scoped tables is (org_id, …) REFERENCES t (org_id, …), org_id at the same position on both sides. internal/kernel/db/tenantfk_test.go enforces it.
  • Retention. Every new table appears in exactly one list in internal/kernel/db/retention_test.go. Published definition tables and the decision log are appendOnly (no UPDATE, no DELETE granted to erp_app). Instances and delegations are amendable (no DELETE). Drafts are operational.
  • Every state change writes audit.Write on the same pgx.Tx as the change.
  • Lock order. This plan adds no advisory lock class. Serialization is by SELECT … FOR UPDATE on the row being changed and by unique constraints. Inside Transition, posting.Dispatch runs before the engine’s audit.Write, so any advisory lock a Poster takes is acquired before class 1 (the rule on lockClassAuditChain in internal/kernel/audit/audit.go).
  • No nested db.InTx. Platform functions take (ctx, tx pgx.Tx, …) and never open a transaction. Only handlers call db.InTx, once per request, reads included.
  • Money is money.Amount end to end, NUMERIC(19,4) in Postgres, a string on the wire. Read a numeric column as col::text into a *string and money.Parse it; never scan into a float.
  • Dates come from kernel/clock. The business date is clock.TodayUTC(c); never time.Now() in code under test. Every function that stamps a time or decides “today” takes a clock.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.Protected with a workflow.* permission registered by member.RegisterPermission in the module’s init(). api/openapi.yaml documents every route in the same commit that registers it (internal/app/routes_test.go fails 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.
  • posting is a process-global registry. Tests that register a fake Poster call posting.ResetForTest() before registering and in t.Cleanup, and never use t.Parallel().
  • Tests, during development: only the changed packages (go test ./internal/platform/workflow/... -count=1, etc.). Never go test ./... until Task 12, which runs the full suite once.
  • Record-bearing free text is English or Bahasa Malaysia (§11A.3). Seeded labels ship en and zh (the supported UI locales, kernel/locale.Supported).
  • Commits end with the line Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>.

Decisions taken where the specs are silent or ambiguous. Each is Ruling: decision — why — cost if wrong.

  1. 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 no status column, and “current” means max(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.
  2. Ruling: the spec’s columns are typed rather than polymorphic — semantic_params becomes control_kind (CHECKed against vocab.ControlKinds), grantee_ref becomes grantee_member_id / grantee_position_id / levels_up, the undefined edge guard is omitted, and the instance pins definition_id (whose row carries the version) instead of duplicating definition_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.
  3. Ruling: a terminal state may have exits, but only into dispute-semantic states — §6.3’s flow has received (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.
  4. Ruling: rule 3 is evaluated on the graph with rejected/cancelled states removed (they reverse, so they reset the count): no cycle there may contain an accrual or payout state; from draft and from every exit of a reversing state, no path crosses two accruals or two payouts; accrual_first means no path reaches a payout without first crossing an accrual, payout_first means 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.
  5. Ruling: rule 5’s “reversal action” is a reverses_accrual flag on the edge; at runtime the reversal is driven by workflow_instance.accrual_open, passed as posting.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.
  6. Ruling: rule 6’s “forward edge” is the single target-gated edge leaving the same state, and an advance edge 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 (use source or target gating there).
  7. Ruling: a reason is mandatory on every edge into a dispute, rejected or cancelled state (validator rule 0), and the member who entered a dispute state can never take a source-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.
  8. 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.
  9. Ruling: role sets and quorum are deferred — no role_set_id column 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.
  10. Ruling: escalation is evaluated lazily at authorization time, never auto-approves, and needs no worker: once today − entered_state_on ≥ escalate_after_days of the state being left, the holders of NextFilledAbove(primary position of each winning-band approver, grant.max_escalation_levels) may also decide, acting on the escalated-from approver’s band, recorded as authority_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.
  11. 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.
  12. Ruling: all_states grants (a) join every state’s ladder, (b) exempt their holder from the self-approval refusal only on edges into accrual or payout states, (c) give standing on reporter-gated edges except those into acknowledged states — the three properties §5.4 lists, read literally — cost: none beyond the literal reading.
  13. Ruling: submission (any transition out of a draft-semantic state into a non-draft, non-reversing state) resolves every approval step (every target edge’s destination) for this claimant and amount and is refused, naming each gap, when (a) a superior grant 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 (NextFilledAbove from 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 a draft-semantic state”, which made withdrawing a draft into cancelled/rejected need a routable chain, so a superior-less claimant could never withdraw. Withdrawal asks nobody to decide, so it is not a submission.)
  14. Ruling: Plan 3 exposes no generic instance-transition HTTP endpoint; consumers (Plan 5+) own POST /v1/<resource>/{id}/transitions, its role permission and its Idempotency-Key. The engine’s TransitionInput requires the From state the caller saw and refuses a stale one — §4.3 keeps “can this role do this verb” with the owning module; From makes 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.
  15. Ruling: the instance stores its amount; SetAmount is allowed only while the instance is in a draft-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.
  16. Ruling: the seeded v1 grants are reviewed ← superior, approved ← unit_head, category_verified, completed and disputed ← position OWNER, all floors NULL; dispute exits are disputed → received (dismiss) and disputed → rejected (uphold, reverses accrual); completed is never re-entered; only expense_claim is seeded, by step 60 workflow_definitions, registered from internal/platform/workflow’s init() — the only seat provisioning guarantees is OWNER; re-entering completed would 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).
  17. Ruling (flag, not a change): posting.Event cannot say “past payout”, so an upheld dispute after payout reaches the Poster only as rejected with PastAccrual = 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.
  18. Ruling: labels are translation rows keyed to the published row ids (workflow_state/label, workflow_edge/action_label), seeded in en and zh; 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.
  19. Ruling: gate kinds, grantee kinds, posting orders and vacancy policies are Go constants plus OpenAPI enums, not constants.yaml entries; 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.
  20. Ruling: the engine writes no activity_log rows 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.
  21. Ruling: platform SQL is raw pgx, like orgstruct; no db/queries/workflow*.sql files — 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.
  22. 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 at Start with “no published definition”.

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).
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.
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; new workflowtest/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.T
    H *testdb.Handles
    DB *db.DB
    Admin *pgxpool.Pool
    Ctx context.Context // tenancy.WithOrg(ctx, Org)
    Org uuid.UUID
    Unit uuid.UUID // root unit "HQ"
    Clock clock.Clock // clock.Fixed(T0)
    }
    func NewWorld(t *testing.T) *World
    func (w *World) SecondOrg() *World // same database, another tenant
    func (w *World) Member(name string) uuid.UUID
    func (w *World) Seat(code string, reportsTo *uuid.UUID) uuid.UUID
    func (w *World) Sit(member, seat uuid.UUID) // primary, from Jan1, open-ended
    func (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) error
    func (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 Down
DROP 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 Down
DROP 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 Down
DROP 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 Down
DROP 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
Terminal window
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.Semantic and its constants; vocab.ControlKinds; locale.Supported; money.Amount.

  • Produces (every later task uses these names exactly):

    type Gate string // GateReporter, GateTarget, GateSource, GateAdvance
    type GranteeKind string // GranteeMember, GranteeSuperior, GranteePosition, GranteeUnitHead, GranteeLevelsUp
    type PostingOrder string // AccrualFirst, PayoutFirst
    type OnVacancy string // VacancyEscalate, VacancyRefuse
    const DefaultMaxEscalationLevels = 2
    const MaxStates, MaxEdges, MaxGrants = 64, 256, 256
    type Labels map[string]string
    type 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() Graph
    func (g Graph) State(key string) (StateSpec, bool)
    func (g Graph) Edge(from, to string) (EdgeSpec, bool)
    func (g Graph) EdgesFrom(key string) []EdgeSpec
    func (g Graph) GateState(e EdgeSpec) (string, bool)
    func (g Graph) ApprovalStates() []string
    type 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 grant
    type Violation struct { Rule int; Code, Detail, Ref string }
    func ValidateGraph(g Graph, baseCurrency string) []Violation
    var 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
Terminal window
git add internal/platform/workflow/model.go internal/platform/workflow/validate.go internal/platform/workflow/sqlhelp.go internal/platform/workflow/validate_test.go
git 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 prefixed st)

Interfaces:

  • Consumes: Task 1’s tables and workflowtest; Task 2’s Graph, Definition, Violation, ValidateGraph; company.Get; audit.Write.

  • Produces:

    var ErrNoDraft, ErrNoDefinition, ErrRevisionConflict, ErrVersionConflict error
    type InvalidError struct{ Violations []Violation } // *InvalidError; Error() lists codes
    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; DraftRevision int } // 0 = none
    func 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) error
    func 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 orgOf and idSet (sqlhelp.go). expectedRevision == 0 means “there is no draft yet”. Audit entity kinds: workflow_draft (actions save, 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
Terminal window
git add internal/platform/workflow/store.go internal/platform/workflow/store_test.go
git 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 prefixed rs)

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 nil
    Members []uuid.UUID `json:"members"` // active members, sorted, deduplicated
    Escalated bool `json:"escalated"` // members came from vacancy escalation or self-replacement
    Gap *Gap `json:"gap,omitempty"`
    }
    func Resolve(ctx context.Context, tx pgx.Tx, grants []Grant, reporter uuid.UUID, on time.Time) ([]Resolved, error)

    Resolved is what the decision log persists (its JSON is workflow_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
Terminal window
git add internal/platform/workflow/resolve.go internal/platform/workflow/resolve_test.go
git 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 prefixed dl)

Interfaces:

  • Consumes: Task 1’s workflow_delegation; audit.Write; clock.TodayUTC.

  • Produces:

    var ErrSelfDelegation, ErrDelegateInactive, ErrBackdated, ErrBadWindow, ErrNoDelegation, ErrAlreadyRevoked error
    type Delegation struct {
    ID, DelegatorMemberID, 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 }
    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) error
    func ListDelegations(ctx context.Context, tx pgx.Tx, member uuid.UUID) ([]Delegation, error) // as delegator or delegate, newest first
    func 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
Terminal window
git add internal/platform/workflow/delegation.go internal/platform/workflow/delegation_test.go
git 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 prefixed az)

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_via
    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"` }
    type Subject struct { Kind string; Reporter uuid.UUID; Amount money.Amount; EnteredStateOn time.Time; DisputeRaiser *uuid.UUID }
    type Authority struct {
    Actor, AuthorityMember uuid.UUID
    Via Via
    GrantID *uuid.UUID
    DelegationID *uuid.UUID
    GateState string // "" for a reporter edge
    WinningFloor *money.Amount // nil for a reporter edge
    Eligible []uuid.UUID // members of the winning band (reporter edge: the reporter plus all_states holders)
    Resolution []Resolved // every grant consulted, as resolved
    ResolvedFor 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
Terminal window
git add internal/platform/workflow/authorize.go internal/platform/workflow/ladder_test.go internal/platform/workflow/authorize_test.go
git 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 prefixed in)

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 error
    const RefuseNotRoutable = "not_routable"
    type RefusedError struct{ Refusal Refusal } // 403 at the edge
    type SubmissionRefusedError struct{ Steps []StepRouting } // 422 at the edge; only the unroutable steps
    type Instance struct {
    ID, DefinitionID uuid.UUID; Kind string; Version int; SourceID, Reporter uuid.UUID
    StateKey string; Amount money.Amount; Currency string; AccrualOpen bool
    EnteredStateAt, 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 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, 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) error
    func 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 Start in the same transaction, calls SetAmount while its document is editable, exposes POST /v1/<resource>/{id}/transitions that calls Transition (owning the role permission and Idempotency-Key, Ruling 14), and embeds AvailableTransitions in 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
Terminal window
git add internal/platform/workflow/instance.go internal/platform/workflow/instance_test.go
git 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 prefixed sd)

Interfaces:

  • Consumes: Task 3 (PublishGraph, latestVersion), Task 7 (Start, Transition for 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 key
    const StepOrder = 60
    const (
    WarningGrantMemberInactive = "workflow_grant_member_inactive"
    WarningGrantPositionRetired = "workflow_grant_position_retired"
    WarningGrantVacant = "workflow_grant_vacant"
    )
    var ErrNoOwnerSeat error
    func DefaultExpenseClaim(ownerSeat uuid.UUID) Graph
    func SeededGraphs(ownerSeat uuid.UUID) []Graph // every graph provisioning seeds; the drift test iterates it
    func SeedDefaults(ctx context.Context, tx pgx.Tx, c clock.Clock) error
    func 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 links internal/platform/workflow — Task 9 makes every binary link it through modules/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
Terminal window
git add internal/platform/workflow/seed.go internal/platform/workflow/seed_test.go
git 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; regenerate internal/kernel/vocab/vocab.gen.go and packages/contracts/vocab.ts
  • Modify (append-only): api/openapi.yaml; regenerate internal/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) error
    func actorOf(c *gin.Context) (uuid.UUID, error)
    func mapErr(err error) error
    type 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.yaml and 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.read

Run: 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/workflow

Run: 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
Terminal window
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 five httpx.Protected lines), api/openapi.yaml (append), regenerate internal/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 test env, 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
Terminal window
git add internal/modules/workflow/ api/openapi.yaml internal/gen/api/api.gen.go
git 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 to mapErr), api/openapi.yaml (append), regenerate internal/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
Terminal window
git add internal/modules/workflow/ api/openapi.yaml internal/gen/api/api.gen.go
git 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
Terminal window
git add cmd/migrate/wiring_test.go cmd/orgadmin/wiring_test.go docs/ROADMAP.md
git commit -m "Assert every provisioning binary sees the workflow step; tick Plan 3
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>"

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.

  • 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 2 currency_mismatch. §5.5 semantics as the ledger contract → Task 7 (posting.Dispatch events), 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 → draft in 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.Dispatch in every transition, no edit to posting → Global Constraints, Tasks 7, 9–12. Org-structure spec §5.5 (persist the resolved set) → Task 7 TestTheDecisionLogKeepsTheApproverSetWhenTheChartLaterChanges.
  • 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 module env) 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 OpenAPI WorkflowResolvedGrant (10). Via values reporter|grant|delegation|escalation match workflow_decision.authority_via’s CHECK (Task 1) and the OpenAPI enum (11). orgOf/idSet live in Task 2’s sqlhelp.go and nowhere else. workflowtest.Today is the on every test passes; T0 is only the clock’s instant.
  • Known sharp edge for implementers. posting is process-global: Task 7’s and Task 8’s tests register a fake Poster and reset it on cleanup; no test in the package may call t.Parallel().