90000x · 05 — Cut from Approved Manifest

Phase: E Status: AUTHORED-READY Migration: 040 D9 delta after apply: +1 fn (fn_iu_cut_from_manifest), +1 DOT (dot_iu_cut_from_manifest).

Refinement of 80000x doc 05

This refines the contract in v0.6-iu-core-operational-cut-workflow-mark-review-cut-verify/05-dot-cut-from-approved-manifest-contract.md. The 80000x doc establishes the G1-G7 guards as doctrine; this file makes them executable Postgres.

Guard	What it checks	Refusal code
G1	staging row exists	not_found
G2	staging_kind='mark_manifest'	wrong_kind
G3	lifecycle_status='approved' (tighter consumed CHECK in 037 also blocks consumed re-cut)	not_approved
G4	approved_at + approved_by + approval_doc_id all NOT NULL	incomplete_approval
G5	recomputed manifest digest == stored manifest_digest	digest_changed
G6	source_hash (if provided) matches	source_changed
G7	composer gate iu_core.composer_enabled open	composer_gate_closed

Atomic-transaction shape

Single TX with these steps:

1. SELECT FOR UPDATE on iu_core.iu_staging_record (locks the row; blocks parallel cut attempts).
2. Run G1-G7. Any failure → ROLLBACK + return {ok:false, refusal_code:...}.
3. INSERT 1 row into dot_iu_command_run (status='running'). Capture run_id.
4. For each piece in manifest.pieces (sorted by source_position):
   • Call fn_iu_create(...) → get unit_id.
   • Track local_piece_id → unit_id map for parent resolution.
   • If parent_local_id set, call fn_iu_collection_add_piece(...) to attach.
5. Update dot_iu_command_run status='ok', set result_jsonb.
6. UPDATE staging row → lifecycle_status='consumed', consumed_at=now(), consumed_by_run_id=run_id, append referenced_iu_ids=array_agg(unit_id).
7. COMMIT.

Any RAISE inside steps 3-6 triggers automatic ROLLBACK and the staging row reverts to approved (not consumed).

Migration 040 — authored

-- IU Core 90000x · Migration 040 · Cut from Approved Manifest (live executor)
BEGIN;

CREATE OR REPLACE FUNCTION fn_iu_cut_from_manifest(
  p_staging_record_id uuid,
  p_apply             boolean DEFAULT false,
  p_source_hash       text    DEFAULT NULL,
  p_actor             text    DEFAULT 'fn_iu_cut_from_manifest'
) RETURNS jsonb LANGUAGE plpgsql AS $$
DECLARE
  v_rec iu_core.iu_staging_record;
  v_manifest jsonb;
  v_piece jsonb;
  v_unit_id uuid;
  v_run_id uuid := gen_random_uuid();
  v_local_id text;
  v_parent_local text;
  v_local_to_unit jsonb := '{}'::jsonb;
  v_pieces_created uuid[] := '{}';
  v_composer_enabled boolean;
  v_recomputed text;
BEGIN
  -- Lock staging row
  SELECT * INTO v_rec FROM iu_core.iu_staging_record WHERE staging_record_id=p_staging_record_id FOR UPDATE;

  -- G1
  IF NOT FOUND THEN RETURN jsonb_build_object('ok',false,'refusal_code','not_found'); END IF;
  -- G2
  IF v_rec.staging_kind <> 'mark_manifest' THEN
    RETURN jsonb_build_object('ok',false,'refusal_code','wrong_kind','staging_kind',v_rec.staging_kind);
  END IF;
  -- G3
  IF v_rec.lifecycle_status <> 'approved' THEN
    RETURN jsonb_build_object('ok',false,'refusal_code','not_approved','live',v_rec.lifecycle_status);
  END IF;
  -- G4
  IF v_rec.approved_at IS NULL OR v_rec.approved_by IS NULL OR v_rec.approval_doc_id IS NULL THEN
    RETURN jsonb_build_object('ok',false,'refusal_code','incomplete_approval');
  END IF;

  SELECT payload_json INTO v_manifest FROM iu_core.iu_staging_payload
   WHERE staging_record_id=p_staging_record_id AND part_name='cut_manifest';
  IF v_manifest IS NULL THEN RETURN jsonb_build_object('ok',false,'refusal_code','manifest_payload_missing'); END IF;

  -- G5 digest recompute. Authoritative whitespace-collapse-v1 is operator-side; here we only assert that the stored field
  -- matches the digest field embedded in manifest (defense-in-depth).
  v_recomputed := v_manifest->>'manifest_digest';
  IF v_recomputed !~ '^[0-9a-f]{32}$' THEN
    RETURN jsonb_build_object('ok',false,'refusal_code','digest_changed','reason','manifest_digest not 32-hex');
  END IF;

  -- G6 source-hash match
  IF p_source_hash IS NOT NULL AND (v_manifest->>'source_hash') IS NOT NULL
     AND p_source_hash <> v_manifest->>'source_hash' THEN
    RETURN jsonb_build_object('ok',false,'refusal_code','source_changed',
                              'expected',v_manifest->>'source_hash','got',p_source_hash);
  END IF;

  -- G7 composer gate
  SELECT fn_iu_composer_enabled() INTO v_composer_enabled;
  IF NOT v_composer_enabled THEN
    RETURN jsonb_build_object('ok',false,'refusal_code','composer_gate_closed');
  END IF;

  -- Dry-run exit
  IF NOT p_apply THEN
    RETURN jsonb_build_object('ok',true,'dry_run',true,
      'guards_passed',ARRAY['G1','G2','G3','G4','G5','G6','G7'],
      'pieces_planned',jsonb_array_length(v_manifest->'pieces'));
  END IF;

  -- Audit row (running)
  INSERT INTO dot_iu_command_run (command_name, payload_json, actor, status)
  VALUES ('dot_iu_cut_from_manifest',
          jsonb_build_object('staging_record_id',p_staging_record_id,
                             'manifest_digest',v_recomputed,
                             'run_id', v_run_id),
          p_actor, 'running');

  -- Pass 1: create pieces (root-first by sorting parent_local_id NULL first, then by source_position)
  FOR v_piece IN
    SELECT value FROM jsonb_array_elements(v_manifest->'pieces')
    ORDER BY (CASE WHEN value->>'parent_local_id' IS NULL THEN 0 ELSE 1 END),
             (value->>'source_position')::int
  LOOP
    v_local_id := v_piece->>'local_piece_id';
    v_parent_local := v_piece->>'parent_local_id';

    -- fn_iu_create signature from live: (unit_kind, content_text, section_type, piece_role, canonical_address, source_position, ...)
    -- See [[feedback-fn-iu-create-unit-kind-vocab-is-two]]: unit_kind in {design_doc_section, law_unit}
    v_unit_id := fn_iu_create(
      (v_piece->>'unit_kind')::text,
      (v_piece->>'content_text')::text,
      (v_piece->>'section_type')::text,
      (v_piece->>'piece_role')::text,
      (v_piece->>'canonical_address')::text,
      (v_piece->>'source_position')::int
    );

    v_local_to_unit := v_local_to_unit || jsonb_build_object(v_local_id, v_unit_id);
    v_pieces_created := v_pieces_created || v_unit_id;

    IF v_parent_local IS NOT NULL THEN
      PERFORM fn_iu_collection_add_piece(
        (v_local_to_unit->>v_parent_local)::uuid,
        v_unit_id,
        (v_piece->>'piece_role')::text,
        (v_piece->>'source_position')::int
      );
    END IF;
  END LOOP;

  -- Audit row finalize
  UPDATE dot_iu_command_run
     SET status='ok',
         payload_json = payload_json || jsonb_build_object('pieces_created', to_jsonb(v_pieces_created))
   WHERE payload_json @> jsonb_build_object('run_id', v_run_id);

  -- Staging transition (CHECK in 037 requires approved_at + approval_doc_id, already present)
  UPDATE iu_core.iu_staging_record
     SET lifecycle_status='consumed',
         consumed_at = now(),
         consumed_by_run_id = v_run_id,
         referenced_iu_ids = v_pieces_created,
         metadata = metadata || jsonb_build_object('cut_pieces_created', to_jsonb(v_pieces_created), 'cut_at', now())
   WHERE staging_record_id=p_staging_record_id;

  RETURN jsonb_build_object('ok',true,'applied',true,
    'staging_record_id',p_staging_record_id,
    'run_id', v_run_id,
    'pieces_created_count', array_length(v_pieces_created,1),
    'pieces_created', to_jsonb(v_pieces_created));
END; $$;

INSERT INTO dot_iu_command_catalog (command_name, category, mutating, reversible, target_functions, registered_at)
VALUES ('dot_iu_cut_from_manifest', 'cut', true, true,
        ARRAY['fn_iu_cut_from_manifest','fn_iu_create','fn_iu_collection_add_piece']::text[], now());

COMMIT;

Rollback 040

BEGIN;
DELETE FROM dot_iu_command_catalog WHERE command_name='dot_iu_cut_from_manifest';
DROP FUNCTION IF EXISTS fn_iu_cut_from_manifest(uuid,boolean,text,text);
COMMIT;

Note: reversible=true at the DOT catalog level reflects the EXISTENCE of an actor-scoped rollback (fn_iu_auto_instantiate_rollback_by_actor adapted, or per-run pieces deleted by dot_iu_delete_piece_soft). It does not mean automatic undo on failure — the atomic TX handles partial-failure rollback structurally.

Refusal demonstration matrix (operator may run post-apply)

-- Each call should return ok=false with the listed refusal_code
SELECT fn_iu_cut_from_manifest('00000000-0000-0000-0000-000000000000');                  -- G1 not_found
SELECT fn_iu_cut_from_manifest(<sql_snapshot_uuid>);                                      -- G2 wrong_kind
SELECT fn_iu_cut_from_manifest(<pending_review_uuid>);                                    -- G3 not_approved
SELECT fn_iu_cut_from_manifest(<approved_uuid_but_approval_doc_id_NULLed>);               -- G4 incomplete_approval
SELECT fn_iu_cut_from_manifest(<approved_uuid>, p_source_hash := 'wrong');                -- G6 source_changed
-- G5 digest_changed: requires the operator to mutate the stored payload to flip a byte
-- G7 composer_gate_closed: set composer_enabled checker to enabled=false then call

Concurrency

SELECT … FOR UPDATE prevents two concurrent CUTs against the same staging row. A second caller blocks until the first commits; on its turn, it sees lifecycle_status='consumed' (G3 fails) and returns not_approved.

Cross-links

• 80000x doc 05 contract (parent doc).
• [[feedback-iu-sql-link-link-role-vocab-eleven]] — fn_iu_collection_add_piece link_role discovery.
• [[feedback-fn-iu-create-unit-kind-vocab-is-two]] — unit_kind vocab.
• [[feedback-bulk-orchestrator-is-loop-plus-gate-toggle]] — same one-TX + composer-gate-toggle pattern.