22-P1 — IU Native Create Helper Functions (rev8 — final)

Date: 2026-05-06 | Status: PROMPT rev8 — chờ final yes/no Controlling: GPT rev7 hidden-schema review (5 fixes) Principle: helper đọc cột nào → preflight kiểm cột đó. Không đọc cột không kiểm.

§0. Mission

Lắp 5 helper functions. Transaction-wrapped. Preflight kiểm MỌI schema field helpers read.

Allowed: helper CREATE FUNCTION + REVOKE only. Forbidden: table/index/constraint DDL, row DML, DOT/script, adapter, seed data.

§1. Pre-read

P0 report
All GPT P1 reviews
Pack 22 rev6

§2. Baseline Counts (audit only)

SELECT 'information_unit' AS tbl, count(*) FROM public.information_unit
UNION ALL SELECT 'unit_version', count(*) FROM public.unit_version
UNION ALL SELECT 'birth_iu', count(*) FROM public.birth_registry WHERE collection_name='information_unit'
UNION ALL SELECT 'birth_total', count(*) FROM public.birth_registry;

Drift → WARNING. NOT rollback trigger.

§3. Preflight + Samples

-- Name conflict (exact list)
SELECT proname FROM pg_proc p JOIN pg_namespace n ON n.oid=p.pronamespace
WHERE n.nspname='public' AND proname IN ('fn_content_hash','fn_iu_resolve_default','fn_iu_classify_existing','fn_iu_create_preflight','fn_iu_verify_invariants');
-- 0 rows. If any → STOP. Future overloads need separate design pack. Agent must NOT workaround.

SELECT to_regprocedure('digest(text,text)') IS NOT NULL AS digest_ok,
       to_regprocedure('gen_random_uuid()') IS NOT NULL AS uuid_ok;

-- FK count + detail
SELECT count(*) AS fk_count FROM pg_constraint
WHERE conrelid='public.information_unit'::regclass AND confrelid='public.unit_version'::regclass AND contype='f' AND condeferrable;
-- Must = 1.
SELECT conname,condeferrable,condeferred FROM pg_constraint
WHERE conrelid='public.information_unit'::regclass AND confrelid='public.unit_version'::regclass AND contype='f' AND condeferrable;

-- Samples
SELECT substring(key from length('vocab.unit_kind.')+1) AS sample_unit_kind
FROM public.dot_config WHERE left(key,length('vocab.unit_kind.'))='vocab.unit_kind.' ORDER BY key LIMIT 1;

SELECT canonical_address AS sample_iu_addr
FROM public.information_unit ORDER BY canonical_address LIMIT 1;

§4. Hash Inspect (report only, no reuse)

SELECT proname,pronargs,provolatile,pg_get_functiondef(oid) FROM pg_proc WHERE proname='fn_sbx_compute_content_hash';

§5. Transaction

BEGIN;

5.1 fn_content_hash

CREATE FUNCTION public.fn_content_hash(p_body text)
RETURNS text LANGUAGE plpgsql IMMUTABLE
-- IMMUTABLE: algorithm constant (sha256), no config lookup.
-- If future hash policy config-driven → redesign volatility.
SET search_path=pg_catalog,public AS $$
BEGIN RETURN encode(digest(p_body,'sha256'),'hex'); END;$$;

5.2 fn_iu_resolve_default

CREATE FUNCTION public.fn_iu_resolve_default(p_explicit text,p_config_key text,p_vocab_prefix text)
RETURNS jsonb LANGUAGE plpgsql STABLE SET search_path=pg_catalog,public AS $$
DECLARE v_configured text;v_config_count int;v_values text[];v_count int;v_prefix_len int;v_explicit_trimmed text;
BEGIN
  IF p_config_key IS NULL OR trim(p_config_key)='' THEN RETURN jsonb_build_object('status','invalid_input','message','p_config_key NULL/empty'); END IF;
  IF p_vocab_prefix IS NULL OR trim(p_vocab_prefix)='' THEN RETURN jsonb_build_object('status','invalid_input','message','p_vocab_prefix NULL/empty'); END IF;
  v_prefix_len:=length(p_vocab_prefix); v_explicit_trimmed:=btrim(COALESCE(p_explicit,''));
  IF v_explicit_trimmed!='' THEN
    IF NOT EXISTS(SELECT 1 FROM public.dot_config WHERE key=p_vocab_prefix||v_explicit_trimmed) THEN
      RETURN jsonb_build_object('status','invalid','value',v_explicit_trimmed,'message','Not in vocab. Available: '||COALESCE((SELECT string_agg(substring(key from v_prefix_len+1),', ' ORDER BY key) FROM public.dot_config WHERE left(key,v_prefix_len)=p_vocab_prefix),'none'));
    END IF; RETURN jsonb_build_object('status','explicit','value',v_explicit_trimmed); END IF;
  SELECT count(*) INTO v_config_count FROM public.dot_config WHERE key=p_config_key;
  IF v_config_count>1 THEN RETURN jsonb_build_object('status','invalid_config_duplicate','message','Duplicate "'||p_config_key||'" ('||v_config_count||' rows)'); END IF;
  IF v_config_count=1 THEN
    SELECT value INTO v_configured FROM public.dot_config WHERE key=p_config_key;
    IF EXISTS(SELECT 1 FROM public.dot_config WHERE key=p_vocab_prefix||v_configured) THEN RETURN jsonb_build_object('status','default','value',v_configured);
    ELSE RETURN jsonb_build_object('status','invalid_config','value',v_configured,'message','Default "'||v_configured||'" removed. Fix "'||p_config_key||'".'); END IF; END IF;
  SELECT array_agg(substring(key from v_prefix_len+1) ORDER BY key) INTO v_values FROM public.dot_config WHERE left(key,v_prefix_len)=p_vocab_prefix;
  v_count:=COALESCE(array_length(v_values,1),0);
  IF v_count=1 THEN RETURN jsonb_build_object('status','auto_single','value',v_values[1]); END IF;
  RETURN jsonb_build_object('status','unresolved','value',NULL,'message',CASE WHEN v_count=0 THEN 'No vocab under '||p_vocab_prefix ELSE 'Multiple, choose: '||array_to_string(v_values,', ') END);
END;$$;

5.3 fn_iu_classify_existing (no lifecycle_status — reduce coupling)

CREATE FUNCTION public.fn_iu_classify_existing(p_addr text)
RETURNS jsonb LANGUAGE plpgsql STABLE SET search_path=pg_catalog,public AS $$
DECLARE v_iu_id uuid;v_uv record;v_uv_count int;v_has_birth boolean;v_anchor_exact boolean;v_issues text[]:='{}';
  v_var uuid; v_car text;
BEGIN
  IF p_addr IS NULL OR btrim(p_addr)='' THEN RETURN jsonb_build_object('status','invalid_input','message','canonical_address NULL/empty'); END IF;
  SELECT id,version_anchor_ref,content_anchor_ref INTO v_iu_id,v_var,v_car FROM public.information_unit WHERE canonical_address=btrim(p_addr);
  IF v_iu_id IS NULL THEN RETURN jsonb_build_object('status','not_found','canonical_address',btrim(p_addr),'issues','[]'::jsonb); END IF;
  SELECT count(*) INTO v_uv_count FROM public.unit_version WHERE unit_id=v_iu_id AND version_seq=1;
  IF v_uv_count=0 THEN v_issues:=array_append(v_issues,'missing_version');
  ELSIF v_uv_count>1 THEN v_issues:=array_append(v_issues,'duplicate_version_v1'); END IF;
  v_has_birth:=COALESCE((SELECT EXISTS(SELECT 1 FROM public.birth_registry WHERE collection_name='information_unit' AND entity_code='information_unit::'||v_iu_id::text)),false);
  IF NOT v_has_birth THEN v_issues:=array_append(v_issues,'missing_birth'); END IF;
  v_anchor_exact:=false;
  IF v_uv_count=1 THEN
    SELECT * INTO v_uv FROM public.unit_version WHERE unit_id=v_iu_id AND version_seq=1;
    v_anchor_exact:=COALESCE(v_var=v_uv.id AND v_car=v_uv.id::text AND v_uv.unit_id=v_iu_id,false);
    IF NOT v_anchor_exact THEN v_issues:=array_append(v_issues,'anchor_invalid'); END IF; END IF;
  RETURN jsonb_build_object('status',CASE WHEN array_length(v_issues,1) IS NULL THEN 'exists_complete'
    WHEN 'missing_version'=ANY(v_issues) THEN 'exists_missing_version'
    WHEN 'duplicate_version_v1'=ANY(v_issues) THEN 'exists_duplicate_version'
    WHEN 'missing_birth'=ANY(v_issues) AND NOT('anchor_invalid'=ANY(v_issues)) THEN 'exists_missing_birth'
    WHEN 'anchor_invalid'=ANY(v_issues) THEN 'exists_anchor_invalid'
    ELSE 'exists_unknown_state' END,
    'iu_id',v_iu_id,'canonical_address',btrim(p_addr),'issues',to_jsonb(v_issues));
END;$$;

5.4 fn_iu_create_preflight (full column coverage + fixed unique index check)

CREATE FUNCTION public.fn_iu_create_preflight()
RETURNS jsonb LANGUAGE plpgsql STABLE SET search_path=pg_catalog,public AS $$
DECLARE
  v_missing text[]:='{}';v_col text;
  -- Contract columns: every column any helper reads or writes
  v_req_iu text[]:=ARRAY['id','canonical_address','unit_kind','owner_ref','created_by','updated_by','identity_profile','parent_or_container_ref','version_anchor_ref','content_anchor_ref'];
  v_req_uv text[]:=ARRAY['id','unit_id','body','content_hash','version_seq','created_by'];
  v_req_br text[]:=ARRAY['collection_name','entity_code'];
  v_req_cr text[]:=ARRAY['collection_name','birth_code_strategy'];
  v_req_dc text[]:=ARRAY['key','value'];
  v_fk_name text;v_fk_deferred boolean;v_fk_count int;
  v_birth_count int;v_l1_count int;v_l2_count int;
  v_birth_def text;v_l1_def text;
  v_l2_is_constraint boolean;v_l2_is_deferrable boolean;
  v_has_unique boolean;
BEGIN
  -- Required columns: IU
  FOREACH v_col IN ARRAY v_req_iu LOOP
    IF NOT EXISTS(SELECT 1 FROM information_schema.columns WHERE table_schema='public' AND table_name='information_unit' AND column_name=v_col) THEN
      v_missing:=array_append(v_missing,'information_unit.'||v_col); END IF; END LOOP;
  -- UV
  FOREACH v_col IN ARRAY v_req_uv LOOP
    IF NOT EXISTS(SELECT 1 FROM information_schema.columns WHERE table_schema='public' AND table_name='unit_version' AND column_name=v_col) THEN
      v_missing:=array_append(v_missing,'unit_version.'||v_col); END IF; END LOOP;
  -- birth_registry
  FOREACH v_col IN ARRAY v_req_br LOOP
    IF NOT EXISTS(SELECT 1 FROM information_schema.columns WHERE table_schema='public' AND table_name='birth_registry' AND column_name=v_col) THEN
      v_missing:=array_append(v_missing,'birth_registry.'||v_col); END IF; END LOOP;
  -- collection_registry
  FOREACH v_col IN ARRAY v_req_cr LOOP
    IF NOT EXISTS(SELECT 1 FROM information_schema.columns WHERE table_schema='public' AND table_name='collection_registry' AND column_name=v_col) THEN
      v_missing:=array_append(v_missing,'collection_registry.'||v_col); END IF; END LOOP;
  -- dot_config
  FOREACH v_col IN ARRAY v_req_dc LOOP
    IF NOT EXISTS(SELECT 1 FROM information_schema.columns WHERE table_schema='public' AND table_name='dot_config' AND column_name=v_col) THEN
      v_missing:=array_append(v_missing,'dot_config.'||v_col); END IF; END LOOP;
  IF array_length(v_missing,1)>0 THEN RAISE EXCEPTION 'preflight: missing columns: %',array_to_string(v_missing,', '); END IF;

  -- Unique on canonical_address: sole key only, no composite, no partial
  v_has_unique:=EXISTS(
    SELECT 1 FROM pg_constraint c WHERE c.conrelid='public.information_unit'::regclass AND c.contype='u'
    AND array_length(c.conkey,1)=1
    AND EXISTS(SELECT 1 FROM unnest(c.conkey) k JOIN pg_attribute a ON a.attrelid=c.conrelid AND a.attnum=k WHERE a.attname='canonical_address'));
  IF NOT v_has_unique THEN
    -- Fallback: unique index, sole key, not partial (use indnkeyatts)
    v_has_unique:=EXISTS(
      SELECT 1 FROM pg_index i
      JOIN pg_attribute a ON a.attrelid=i.indrelid AND a.attnum=i.indkey[0]
      WHERE i.indrelid='public.information_unit'::regclass AND i.indisunique
        AND i.indpred IS NULL AND i.indnkeyatts=1 AND a.attname='canonical_address');
  END IF;
  IF NOT v_has_unique THEN RAISE EXCEPTION 'preflight: no sole-column unique on canonical_address'; END IF;

  -- Birth: exactly one, enabled O/A, AFTER INSERT
  SELECT count(*) INTO v_birth_count FROM pg_trigger t JOIN pg_proc p ON p.oid=t.tgfoid
  WHERE t.tgrelid='public.information_unit'::regclass AND NOT t.tgisinternal AND t.tgenabled IN ('O','A') AND p.proname='fn_birth_registry_auto';
  IF v_birth_count!=1 THEN RAISE EXCEPTION 'preflight: birth(O/A)=% (need 1)',v_birth_count; END IF;
  SELECT pg_get_triggerdef(t.oid) INTO v_birth_def FROM pg_trigger t JOIN pg_proc p ON p.oid=t.tgfoid
  WHERE t.tgrelid='public.information_unit'::regclass AND NOT t.tgisinternal AND t.tgenabled IN ('O','A') AND p.proname='fn_birth_registry_auto';
  IF v_birth_def NOT ILIKE '%AFTER INSERT%' THEN RAISE EXCEPTION 'preflight: birth not AFTER INSERT'; END IF;

  -- L1: exactly one, enabled O/A, BEFORE INSERT
  SELECT count(*) INTO v_l1_count FROM pg_trigger t JOIN pg_proc p ON p.oid=t.tgfoid
  WHERE t.tgrelid='public.information_unit'::regclass AND NOT t.tgisinternal AND t.tgenabled IN ('O','A') AND p.proname='fn_iu_birth_gate_layer1';
  IF v_l1_count!=1 THEN RAISE EXCEPTION 'preflight: L1(O/A)=% (need 1)',v_l1_count; END IF;
  SELECT pg_get_triggerdef(t.oid) INTO v_l1_def FROM pg_trigger t JOIN pg_proc p ON p.oid=t.tgfoid
  WHERE t.tgrelid='public.information_unit'::regclass AND NOT t.tgisinternal AND t.tgenabled IN ('O','A') AND p.proname='fn_iu_birth_gate_layer1';
  IF v_l1_def NOT ILIKE '%BEFORE INSERT%' THEN RAISE EXCEPTION 'preflight: L1 not BEFORE INSERT'; END IF;

  -- L2: exactly one, enabled O/A, constraint + deferrable
  SELECT count(*) INTO v_l2_count FROM pg_trigger t JOIN pg_proc p ON p.oid=t.tgfoid
  WHERE t.tgrelid='public.information_unit'::regclass AND NOT t.tgisinternal AND t.tgenabled IN ('O','A') AND p.proname='fn_iu_birth_gate_layer2';
  IF v_l2_count!=1 THEN RAISE EXCEPTION 'preflight: L2(O/A)=% (need 1)',v_l2_count; END IF;
  SELECT t.tgconstraint!=0,t.tgdeferrable INTO v_l2_is_constraint,v_l2_is_deferrable
  FROM pg_trigger t JOIN pg_proc p ON p.oid=t.tgfoid
  WHERE t.tgrelid='public.information_unit'::regclass AND NOT t.tgisinternal AND t.tgenabled IN ('O','A') AND p.proname='fn_iu_birth_gate_layer2';
  IF NOT COALESCE(v_l2_is_constraint,false) THEN RAISE EXCEPTION 'preflight: L2 NOT constraint trigger'; END IF;
  IF NOT COALESCE(v_l2_is_deferrable,false) THEN RAISE EXCEPTION 'preflight: L2 NOT deferrable'; END IF;

  IF to_regprocedure('digest(text,text)') IS NULL THEN RAISE EXCEPTION 'preflight: digest unavailable'; END IF;
  IF to_regprocedure('gen_random_uuid()') IS NULL THEN RAISE EXCEPTION 'preflight: gen_random_uuid unavailable'; END IF;

  SELECT count(*) INTO v_fk_count FROM pg_constraint
  WHERE conrelid='public.information_unit'::regclass AND confrelid='public.unit_version'::regclass AND contype='f' AND condeferrable;
  IF v_fk_count!=1 THEN RAISE EXCEPTION 'preflight: deferrable FK=%  (need 1)',v_fk_count; END IF;
  SELECT conname,condeferred INTO v_fk_name,v_fk_deferred FROM pg_constraint
  WHERE conrelid='public.information_unit'::regclass AND confrelid='public.unit_version'::regclass AND contype='f' AND condeferrable;

  RETURN jsonb_build_object('status','pass','fk_name',v_fk_name,'fk_initially_deferred',v_fk_deferred);
END;$$;

5.5 fn_iu_verify_invariants (no lifecycle_status)

CREATE FUNCTION public.fn_iu_verify_invariants(p_addr text)
RETURNS jsonb LANGUAGE plpgsql STABLE SET search_path=pg_catalog,public AS $$
DECLARE v_iu_id uuid;v_var uuid;v_car text;v_uv record;v_uv_count int;v_uv_strategy text;
  v_has_birth boolean;v_anchor_exact boolean;v_uv_birth_ok boolean;
BEGIN
  IF p_addr IS NULL OR btrim(p_addr)='' THEN RETURN jsonb_build_object('status','invalid_input','message','canonical_address NULL/empty'); END IF;
  SELECT id,version_anchor_ref,content_anchor_ref INTO v_iu_id,v_var,v_car FROM public.information_unit WHERE canonical_address=btrim(p_addr);
  IF v_iu_id IS NULL THEN RETURN jsonb_build_object('status','not_found','canonical_address',btrim(p_addr)); END IF;
  SELECT count(*) INTO v_uv_count FROM public.unit_version WHERE unit_id=v_iu_id AND version_seq=1;
  IF v_uv_count=0 THEN RETURN jsonb_build_object('i1_iu_exists',true,'i2_uv_linked',false,'i3_anchors_exact',false,'i4_birth_exists',false,'i5_uv_birth_ok',true,'all_pass',false,'issue','missing_version'); END IF;
  IF v_uv_count>1 THEN RETURN jsonb_build_object('i1_iu_exists',true,'i2_uv_linked',true,'i3_anchors_exact',false,'i4_birth_exists',false,'i5_uv_birth_ok',true,'all_pass',false,'issue','duplicate_version'); END IF;
  SELECT * INTO v_uv FROM public.unit_version WHERE unit_id=v_iu_id AND version_seq=1;
  v_has_birth:=COALESCE((SELECT EXISTS(SELECT 1 FROM public.birth_registry WHERE collection_name='information_unit' AND entity_code='information_unit::'||v_iu_id::text)),false);
  v_anchor_exact:=COALESCE(v_var=v_uv.id AND v_car=v_uv.id::text AND v_uv.unit_id=v_iu_id,false);
  SELECT birth_code_strategy INTO v_uv_strategy FROM public.collection_registry WHERE collection_name='unit_version';
  IF v_uv_strategy='subordinate' OR v_uv_strategy IS NULL THEN
    v_uv_birth_ok:=NOT EXISTS(SELECT 1 FROM public.birth_registry WHERE collection_name='unit_version' AND entity_code='unit_version::'||v_uv.id::text);
  ELSE
    RETURN jsonb_build_object('i1_iu_exists',true,'i2_uv_linked',true,'i3_anchors_exact',v_anchor_exact,'i4_birth_exists',v_has_birth,'i5_uv_birth_ok',false,'all_pass',false,'issue','uv_strategy_changed','uv_strategy',v_uv_strategy);
  END IF;
  RETURN jsonb_build_object('i1_iu_exists',true,'i2_uv_linked',true,'i3_anchors_exact',v_anchor_exact,'i4_birth_exists',v_has_birth,'i5_uv_birth_ok',v_uv_birth_ok,'all_pass',(v_anchor_exact AND v_has_birth AND v_uv_birth_ok));
END;$$;

5.6 REVOKE PUBLIC

REVOKE ALL ON FUNCTION public.fn_content_hash(text) FROM PUBLIC;
REVOKE ALL ON FUNCTION public.fn_iu_resolve_default(text,text,text) FROM PUBLIC;
REVOKE ALL ON FUNCTION public.fn_iu_classify_existing(text) FROM PUBLIC;
REVOKE ALL ON FUNCTION public.fn_iu_create_preflight() FROM PUBLIC;
REVOKE ALL ON FUNCTION public.fn_iu_verify_invariants(text) FROM PUBLIC;

§6. Tests + Expected Statuses

Test	Acceptable	FAIL if
hash_test	non-null 64-char hex	NULL/wrong length
resolve_null	default/auto_single/unresolved/invalid_config/invalid_config_duplicate	exception
resolve_valid	explicit / SKIPPED	invalid when sample in vocab
resolve_invalid	invalid	else
resolve_null_config	invalid_input	else
resolve_null_prefix	invalid_input	else
classify_existing	exists_* / SKIPPED	exception
classify_absent	not_found	else
classify_null/empty	invalid_input	else
preflight	pass	exception/non-pass
verify_existing	any valid JSON / SKIPPED	exception. If all_pass=false → P2 WARNING, not P1 rollback
verify_absent	not_found	else
verify_null	invalid_input	else

SQL (CTE samples, schema-qualified, no substitution)

SELECT public.fn_content_hash('test body') AS hash_test;
SELECT public.fn_iu_resolve_default(NULL,'iu_create.default_unit_kind','vocab.unit_kind.') AS resolve_null;

WITH vs AS (SELECT substring(key from length('vocab.unit_kind.')+1) AS v FROM public.dot_config WHERE left(key,length('vocab.unit_kind.'))='vocab.unit_kind.' ORDER BY key LIMIT 1)
SELECT CASE WHEN v IS NOT NULL THEN public.fn_iu_resolve_default(v,'iu_create.default_unit_kind','vocab.unit_kind.') ELSE '{"status":"SKIPPED"}'::jsonb END FROM vs
UNION ALL SELECT '{"status":"SKIPPED"}'::jsonb WHERE NOT EXISTS(SELECT 1 FROM public.dot_config WHERE left(key,length('vocab.unit_kind.'))='vocab.unit_kind.');

SELECT public.fn_iu_resolve_default('__nonexistent__','iu_create.default_unit_kind','vocab.unit_kind.') AS resolve_invalid;
SELECT public.fn_iu_resolve_default(NULL,NULL,'vocab.unit_kind.') AS resolve_null_config;
SELECT public.fn_iu_resolve_default(NULL,'x',NULL) AS resolve_null_prefix;

WITH is_ AS (SELECT canonical_address AS addr FROM public.information_unit ORDER BY canonical_address LIMIT 1)
SELECT CASE WHEN addr IS NOT NULL THEN public.fn_iu_classify_existing(addr) ELSE '{"status":"SKIPPED"}'::jsonb END FROM is_
UNION ALL SELECT '{"status":"SKIPPED"}'::jsonb WHERE NOT EXISTS(SELECT 1 FROM public.information_unit);

SELECT public.fn_iu_classify_existing('__nonexistent__') AS classify_absent;
SELECT public.fn_iu_classify_existing(NULL) AS classify_null;
SELECT public.fn_iu_classify_existing('') AS classify_empty;

SELECT public.fn_iu_create_preflight() AS preflight;

WITH is_ AS (SELECT canonical_address AS addr FROM public.information_unit ORDER BY canonical_address LIMIT 1)
SELECT CASE WHEN addr IS NOT NULL THEN public.fn_iu_verify_invariants(addr) ELSE '{"status":"SKIPPED"}'::jsonb END FROM is_
UNION ALL SELECT '{"status":"SKIPPED"}'::jsonb WHERE NOT EXISTS(SELECT 1 FROM public.information_unit);

SELECT public.fn_iu_verify_invariants('__nonexistent__') AS verify_absent;
SELECT public.fn_iu_verify_invariants(NULL) AS verify_null;

SELECT proname,pronargs,provolatile,prosecdef FROM pg_proc p JOIN pg_namespace n ON n.oid=p.pronamespace
WHERE n.nspname='public' AND proname IN ('fn_content_hash','fn_iu_resolve_default','fn_iu_classify_existing','fn_iu_create_preflight','fn_iu_verify_invariants') ORDER BY proname;

SELECT routine_schema,routine_name,grantee,privilege_type FROM information_schema.routine_privileges
WHERE routine_schema='public' AND routine_name IN ('fn_content_hash','fn_iu_resolve_default','fn_iu_classify_existing','fn_iu_create_preflight','fn_iu_verify_invariants') ORDER BY routine_name,grantee;

SKIPPED ≠ FAIL. verify_existing all_pass=false → P2 WARNING, not P1 rollback.

§7. Post-Test Counts

SELECT 'information_unit' AS tbl, count(*) FROM public.information_unit
UNION ALL SELECT 'unit_version', count(*) FROM public.unit_version
UNION ALL SELECT 'birth_iu', count(*) FROM public.birth_registry WHERE collection_name='information_unit'
UNION ALL SELECT 'birth_total', count(*) FROM public.birth_registry;

Drift → WARNING.

§8. COMMIT or ROLLBACK

Pass → COMMIT; | Fail → ROLLBACK;

If ROLLBACK:

SELECT n.nspname,p.proname FROM pg_proc p JOIN pg_namespace n ON n.oid=p.pronamespace
WHERE n.nspname='public' AND p.proname IN ('fn_content_hash','fn_iu_resolve_default','fn_iu_classify_existing','fn_iu_create_preflight','fn_iu_verify_invariants') ORDER BY p.proname;

Report: failing block, error, rollback confirmed, post-check = 0.

§9. Report

Upload: knowledge/dev/laws/dieu44-trien-khai/reports/22-p1-iu-native-create-helper-functions-report.md

Include samples found, each test run/skipped/result, REVOKE verified, counts.

## P2 readiness
- Helper install: [all 5 / rollback]
- Preflight pass: [yes/no]
- Existing sample invariant: [all_pass / diagnostic — if diagnostic: P2 BLOCKER or WARNING]
- Runtime blockers before fn_iu_create: [list or none]

§10. Hard Boundaries

❌ No row DML / No CREATE OR REPLACE / No table DDL
❌ No DOT/adapter/seed/retry/improvise
❌ No hardcode in tests (CTE only)
❌ No reuse fn_sbx_compute_content_hash
❌ No lifecycle_status in output (reduce coupling)

22-P1 Prompt rev8 | 2026-05-06 | Preflight covers all read columns. Sole-key unique. O/A triggers. indnkeyatts. No lifecycle_status. Chờ final yes/no.