06 — PRIMITIVES P4, P5, P6

All three: STAGED (authored, not deployed/registered/executed). Each runs only inside an admitted, active sandbox via the admitted exec channel; each carries the same 8-facet lifecycle (reuse-first, birth, admission, registration, dot_manage_ledger, rollback/retire, readback, orphan check) recorded in sbx_meta. Below are the load-bearing payloads.

---

P4: dot-c1-staging-verify (DOT-C1-STG-VERIFY)

Verify required fields / integrity of the loaded C1 vocab. Read-mostly (sets status=validated).

CREATE OR REPLACE FUNCTION c1_verify(p_code text) RETURNS jsonb LANGUAGE plpgsql AS $f$
DECLARE r canonical_operation; o jsonb; BEGIN
  SELECT * INTO r FROM canonical_operation WHERE operation_code=p_code;
  IF NOT FOUND THEN RETURN jsonb_build_object('ok',false,'reason','not_found','code',p_code); END IF;
  o := jsonb_build_object('ok',true,'code',r.operation_code,'has_title',length(btrim(r.operation_title))>0,
    'group_valid',EXISTS(SELECT 1 FROM c1_operation_group g WHERE g.group_code=r.operation_group),
    'required_inputs_count',jsonb_array_length(r.required_inputs),
    'expected_outputs_count',jsonb_array_length(r.expected_outputs),'allowed_modes',r.allowed_modes,'status',r.status);
  UPDATE canonical_operation SET status='validated', evidence=evidence||jsonb_build_object('verify',o) WHERE operation_code=p_code;
  RETURN o; END;$f$;
-- verify all + integrity invariants (orphan/field check)
SELECT c1_verify(operation_code) FROM canonical_operation;
SELECT 'INVARIANT_FAIL' WHERE EXISTS (SELECT 1 FROM canonical_operation
  WHERE jsonb_array_length(required_inputs)<1 OR jsonb_array_length(expected_outputs)<1
     OR operation_group NOT IN (SELECT group_code FROM c1_operation_group));

PASS = every row validates, zero INVARIANT_FAIL rows.

---

P5: dot-c1-staging-bad-input-harness (DOT-C1-STG-BAD-HARNESS)

Run ≥9 bad cases; each MUST fail closed; assert NO residue (canonical_operation stays 3 rows). Each bad statement runs in a sub-transaction (PL/pgSQL EXCEPTION block = automatic savepoint rollback) so a failed case leaves no partial state.

CREATE TABLE IF NOT EXISTS c1_test_results(case_no int, phase text, label text, expect text,
  outcome text, reject_code text, sqlstate text, err_message text, pass boolean, ts timestamptz DEFAULT now());
CREATE OR REPLACE FUNCTION c1_run(c int,ph text,lb text,ex text,q text) RETURNS void LANGUAGE plpgsql AS $f$
DECLARE st text; ms text; BEGIN
  BEGIN EXECUTE q;
    INSERT INTO c1_test_results VALUES (c,ph,lb,ex,'accepted',NULL,NULL,NULL,(ex='accept'),now());
  EXCEPTION WHEN OTHERS THEN GET STACKED DIAGNOSTICS st=RETURNED_SQLSTATE, ms=MESSAGE_TEXT;
    INSERT INTO c1_test_results VALUES (c,ph,lb,ex,'rejected',split_part(ms,' |',1),st,ms,(ex='reject'),now());
  END; END;$f$;
-- the 9 bad cases (all expect 'reject')
SELECT c1_run(1,'bad','missing operation_code','reject',$$SELECT c1_insert_operation('{"operation_title":"x","operation_group":"READ","required_inputs":["a"],"expected_outputs":["b"],"allowed_modes":["dry_run"]}')$$);
SELECT c1_run(2,'bad','duplicate operation_code','reject',$$SELECT c1_insert_operation('{"operation_code":"C1.READ_BALANCE","operation_title":"dup","operation_group":"READ","required_inputs":["a"],"expected_outputs":["b"],"allowed_modes":["dry_run"]}')$$);
SELECT c1_run(3,'bad','empty title','reject',$$SELECT c1_insert_operation('{"operation_code":"C1.BAD_TITLE","operation_title":"   ","operation_group":"READ","required_inputs":["a"],"expected_outputs":["b"],"allowed_modes":["dry_run"]}')$$);
SELECT c1_run(4,'bad','invalid operation_group','reject',$$SELECT c1_insert_operation('{"operation_code":"C1.BAD_GROUP","operation_title":"x","operation_group":"BOGUS","required_inputs":["a"],"expected_outputs":["b"],"allowed_modes":["dry_run"]}')$$);
SELECT c1_run(5,'bad','missing required_inputs','reject',$$SELECT c1_insert_operation('{"operation_code":"C1.NO_INPUTS","operation_title":"x","operation_group":"READ","expected_outputs":["b"],"allowed_modes":["dry_run"]}')$$);
SELECT c1_run(6,'bad','bad expected_outputs (object)','reject',$$SELECT c1_insert_operation('{"operation_code":"C1.BAD_OUT","operation_title":"x","operation_group":"READ","required_inputs":["a"],"expected_outputs":{"k":"v"},"allowed_modes":["dry_run"]}')$$);
SELECT c1_run(7,'bad','unknown allowed mode','reject',$$SELECT c1_insert_operation('{"operation_code":"C1.BAD_MODE","operation_title":"x","operation_group":"READ","required_inputs":["a"],"expected_outputs":["b"],"allowed_modes":["telepathy"]}')$$);
SELECT c1_run(8,'bad','attempt to WRITE official table','reject',$$INSERT INTO dot_tools(code) VALUES ('nope')$$);  -- 42P01 in sandbox: official table unreachable
SELECT c1_run(9,'bad','attempt to mark production-ready','reject',$$SELECT c1_insert_operation('{"operation_code":"C1.PROMOTE_ME","operation_title":"x","operation_group":"READ","required_inputs":["a"],"expected_outputs":["b"],"allowed_modes":["dry_run"],"status":"production_ready"}')$$);
-- assertions
SELECT 'RESIDUE_FAIL' WHERE (SELECT count(*) FROM canonical_operation) <> 3;
SELECT 'BAD_ACCEPTED_FAIL' WHERE EXISTS (SELECT 1 FROM c1_test_results WHERE phase='bad' AND outcome='accepted');

If any bad case is accepted → harness verdict C1_STAGING_FAST_DRY_RUN_FAIL_BAD_INPUT_ACCEPTED. Expected reject signals: B1 C1_REJ_MISSING_CODE; B2 sqlstate 23505; B3 C1_REJ_EMPTY_TITLE; B4 C1_REJ_INVALID_GROUP; B5 C1_REJ_MISSING_REQUIRED_INPUTS; B6 C1_REJ_BAD_EXPECTED_OUTPUTS; B7 C1_REJ_UNKNOWN_MODE; B8 sqlstate 42P01 (official table not in sandbox = isolation proof); B9 C1_REJ_PRODUCTION_READY_FORBIDDEN.

---

P6: dot-c1-staging-evidence-readback (DOT-C1-STG-EVIDENCE-READBACK)

Digest + evidence bundle + readback; emits to KB. Final orphan sweep.

CREATE OR REPLACE FUNCTION c1_digest() RETURNS jsonb LANGUAGE plpgsql AS $f$
DECLARE b text; BEGIN
  SELECT string_agg(operation_code||'|'||operation_title||'|'||operation_group||'|'||required_inputs::text||'|'||expected_outputs::text||'|'||allowed_modes::text||'|'||status, E'\n' ORDER BY operation_code) INTO b FROM canonical_operation;
  RETURN jsonb_build_object('row_count',(SELECT count(*) FROM canonical_operation),'md5',md5(COALESCE(b,'')),'generated_at',now()); END;$f$;
SELECT jsonb_pretty(c1_digest());                                   -- digest
SELECT * FROM canonical_operation ORDER BY operation_code;          -- vocab readback
SELECT phase,count(*) total,count(*) FILTER (WHERE pass) passed,count(*) FILTER (WHERE NOT pass) failed FROM c1_test_results GROUP BY phase;  -- result readback
-- final orphan check: every sbx object registered?
SELECT 'ORPHAN_OBJECT' AS flag, c.relname FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace
  WHERE n.nspname='public' AND c.relkind IN ('r') AND c.relname NOT IN (SELECT object_name FROM sbx_meta.object_registry);
-- overall gate
SELECT CASE WHEN (SELECT count(*) FROM c1_test_results WHERE NOT pass)=0
  AND (SELECT count(*) FROM canonical_operation)=3
  AND NOT EXISTS (SELECT 1 FROM c1_test_results WHERE phase='bad' AND outcome='accepted')
  THEN 'C1_STAGING_FAST_DRY_RUN_PASS' ELSE 'C1_STAGING_FAST_DRY_RUN_FAIL' END AS dry_run_verdict;

The agent canonicalizes the digest + result tables to KB knowledge/dev/laws-new/reports/c1-staging-fast-dry-run/run-<ts>/ after execution.

---

Run order (once admitted)

P1 create → P3 vocab-build → P4 verify → P5 bad-input-harness → P6 evidence-readback → P2 drop (or TTL retire).