05 — PRIMITIVE P3: dot-c1-staging-vocab-build

Status: STAGED (authored, not deployed/registered/executed). DOT code (staging-lite): DOT-C1-STG-VOCAB-BUILD Purpose: Inside an admitted, active sandbox, build the C1 canonical_operation vocabulary contract objects (table + reference vocab + validation trigger + insert API) and load the valid record set. Idempotent ensure. Registers every object it creates in sbx_meta.object_registry.

C1 = canonical_operation vocabulary contract carrier. This primitive proves facets 1–4 of the macro's C1 definition (create vocab record set; validate required fields; the rejection rules; evidence hooks). Verify / bad-input / digest are P4 / P5 / P6.

8-facet lifecycle

1. reuse-first — CREATE … IF NOT EXISTS; if canonical_operation already present, ensure (no duplicate build); records reused=true in the ledger.
2. birth — ledger op c1_vocab_build_begin with sandbox_id + payload hash.
3. admission — requires sandbox status='active' in sbx_meta.sandbox_registry; fail-closed.
4. registration / staging-registry — inserts each object into sbx_meta.object_registry (kind=table/function/trigger, name).
5. DOT manage ledger — dot_manage_ledger op c1_vocab_build, detail = rows_loaded.
6. rollback / retire — whole build wrapped in ONE transaction; failure → automatic rollback (no partial object set). Full teardown = P2 sandbox-drop.
7. readback — selects the loaded vocab + object_registry.
8. orphan check — asserts object_registry count == actual sbx objects; flags drift.

Governed payload (the C1 contract objects + validation; runs only via the admitted runner)

BEGIN;
-- reference vocab
CREATE TABLE IF NOT EXISTS c1_operation_group(group_code text PRIMARY KEY, description text NOT NULL);
INSERT INTO c1_operation_group VALUES
 ('READ','read-only'),('WRITE','governed write'),('VERIFY','verification'),('REPORT','reporting')
 ON CONFLICT DO NOTHING;
CREATE TABLE IF NOT EXISTS c1_allowed_mode(mode_code text PRIMARY KEY);
INSERT INTO c1_allowed_mode VALUES ('dry_run'),('verify'),('readback'),('plan') ON CONFLICT DO NOTHING;
-- contract carrier
CREATE TABLE IF NOT EXISTS canonical_operation(
  id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  operation_code text NOT NULL UNIQUE,
  operation_title text NOT NULL,
  operation_group text NOT NULL REFERENCES c1_operation_group(group_code),
  required_inputs jsonb NOT NULL, expected_outputs jsonb NOT NULL, allowed_modes jsonb NOT NULL,
  reject_rules jsonb NOT NULL DEFAULT '[]', status text NOT NULL DEFAULT 'staged',
  evidence jsonb NOT NULL DEFAULT '{}', created_at timestamptz NOT NULL DEFAULT now(),
  CONSTRAINT chk_code_format CHECK (operation_code ~ '^C1\.[A-Z0-9_]+$'),
  CONSTRAINT chk_title_nonempty CHECK (length(btrim(operation_title))>0),
  CONSTRAINT chk_status_staging CHECK (status IN ('staged','validated','rejected')));
-- validation trigger (explicit reject codes; fail-closed)
CREATE OR REPLACE FUNCTION c1_validate_operation() RETURNS trigger LANGUAGE plpgsql AS $f$
DECLARE m text; BEGIN
  IF NEW.operation_code IS NULL OR btrim(NEW.operation_code)='' THEN RAISE EXCEPTION 'C1_REJ_MISSING_CODE | required'; END IF;
  IF NEW.operation_code !~ '^C1\.[A-Z0-9_]+$' THEN RAISE EXCEPTION 'C1_REJ_BAD_CODE_FORMAT | %', NEW.operation_code; END IF;
  IF NEW.operation_title IS NULL OR btrim(NEW.operation_title)='' THEN RAISE EXCEPTION 'C1_REJ_EMPTY_TITLE | required'; END IF;
  IF NEW.operation_group IS NULL OR NOT EXISTS(SELECT 1 FROM c1_operation_group g WHERE g.group_code=NEW.operation_group) THEN RAISE EXCEPTION 'C1_REJ_INVALID_GROUP | %', COALESCE(NEW.operation_group,'<null>'); END IF;
  IF NEW.required_inputs IS NULL OR jsonb_typeof(NEW.required_inputs)<>'array' OR jsonb_array_length(NEW.required_inputs)<1 THEN RAISE EXCEPTION 'C1_REJ_MISSING_REQUIRED_INPUTS | non-empty array'; END IF;
  IF NEW.expected_outputs IS NULL OR jsonb_typeof(NEW.expected_outputs)<>'array' OR jsonb_array_length(NEW.expected_outputs)<1 THEN RAISE EXCEPTION 'C1_REJ_BAD_EXPECTED_OUTPUTS | non-empty array'; END IF;
  IF NEW.allowed_modes IS NULL OR jsonb_typeof(NEW.allowed_modes)<>'array' OR jsonb_array_length(NEW.allowed_modes)<1 THEN RAISE EXCEPTION 'C1_REJ_UNKNOWN_MODE | non-empty array'; END IF;
  SELECT string_agg(e,',') INTO m FROM (SELECT jsonb_array_elements_text(NEW.allowed_modes) e) s WHERE e NOT IN (SELECT mode_code FROM c1_allowed_mode);
  IF m IS NOT NULL THEN RAISE EXCEPTION 'C1_REJ_UNKNOWN_MODE | %', m; END IF;
  IF NEW.status NOT IN ('staged','validated','rejected') THEN RAISE EXCEPTION 'C1_REJ_PRODUCTION_READY_FORBIDDEN | %', NEW.status; END IF;
  RETURN NEW; END;$f$;
DROP TRIGGER IF EXISTS trg_c1_validate ON canonical_operation;
CREATE TRIGGER trg_c1_validate BEFORE INSERT OR UPDATE ON canonical_operation FOR EACH ROW EXECUTE FUNCTION c1_validate_operation();
-- insert API (jsonb payload)
CREATE OR REPLACE FUNCTION c1_insert_operation(p jsonb) RETURNS bigint LANGUAGE plpgsql AS $f$
DECLARE v bigint; BEGIN
  INSERT INTO canonical_operation(operation_code,operation_title,operation_group,required_inputs,expected_outputs,allowed_modes,reject_rules,status)
  VALUES (p->>'operation_code',p->>'operation_title',p->>'operation_group',
    COALESCE(p->'required_inputs','null'),COALESCE(p->'expected_outputs','null'),
    COALESCE(p->'allowed_modes','null'),COALESCE(p->'reject_rules','[]'),COALESCE(p->>'status','staged'))
  RETURNING id INTO v; RETURN v; END;$f$;
-- valid record set (the 3+ valid C1 ops)
SELECT c1_insert_operation('{"operation_code":"C1.READ_BALANCE","operation_title":"Read account balance","operation_group":"READ","required_inputs":["account_id"],"expected_outputs":["balance","as_of"],"allowed_modes":["dry_run","readback"]}');
SELECT c1_insert_operation('{"operation_code":"C1.WRITE_POSTING","operation_title":"Create governed posting","operation_group":"WRITE","required_inputs":["ledger_id","amount","memo"],"expected_outputs":["posting_id","status"],"allowed_modes":["dry_run","plan","verify"]}');
SELECT c1_insert_operation('{"operation_code":"C1.VERIFY_DIGEST","operation_title":"Verify canonical digest","operation_group":"VERIFY","required_inputs":["set_id"],"expected_outputs":["md5","row_count"],"allowed_modes":["verify","readback"]}');
-- object registration
INSERT INTO sbx_meta.object_registry(sandbox_id,object_kind,object_name) VALUES
 ('${SBX}','table','c1_operation_group'),('${SBX}','table','c1_allowed_mode'),
 ('${SBX}','table','canonical_operation'),('${SBX}','function','c1_validate_operation'),
 ('${SBX}','trigger','trg_c1_validate'),('${SBX}','function','c1_insert_operation');
INSERT INTO sbx_meta.dot_manage_ledger(sandbox_id,op,actor,detail) VALUES ('${SBX}','c1_vocab_build','${OPERATOR}', jsonb_build_object('rows_loaded',(SELECT count(*) FROM canonical_operation)));
COMMIT;

Minimal columns (matches macro §4)

operation_code, operation_title, operation_group, required_inputs jsonb, expected_outputs jsonb, allowed_modes jsonb, reject_rules jsonb, status, evidence jsonb, created_at. ✓