examples: the rule-check demo behind the clip

CHANGELOG.md

@@ -21,6 +21,10 @@ All notable changes to this project are documented here. The format follows
 - `examples/persona/dispatch-policy.yaml` + `dispatch-props.json`: the rule lock on an agent that learns whether to act or
   ask first. Every rule is held by construction for seeds 0-99; without its routine habit, `check` finds the broken seeds
   (README, "Check in CI").
+- `examples/rule-check/demo.py` (Python 3.9 standard library, probbit 0.8.2 or later): the rule-check demo on the dispatch
+  example. Without the routine habit, the day-0 check flags 7 of 20 individuals, exactly the 7 that 300 synthetic rewarded
+  events teach to ask about routine work (24 of 24 over seeds 0-99); with the habit, `held_by_construction` and 0 of 20.
+  `python/test_rule_check_demo.py` runs it small (4 individuals, 20 events, about 2 s).
 
 ### Fixed
 - `cargo +1.78.0 test --workspace --locked` compiles again: a CLI test used `std::io::pipe()` (Rust 1.87); it now gets a

README.md

@@ -521,6 +521,15 @@ those 24 (9 of them with a single event, so before any learning). The other 9 it
 (`--seeds 18,20,28,30-31,79,87,92,96`) they report `unproven` and exit 3, which fails the job unless you allow it. Measured
 on this file with probbit 0.8.3 (unreleased), seeds 0-99, default depth.
 
+**Reproduce the clip: `python3 examples/rule-check/demo.py`.** The script (Python 3.9 standard library, probbit 0.8.2 or
+later) removes the `routine_acts` habit from the shipped policy and runs the rule check (`persona prove`) on 20 individuals
+(seeds 0-19). Each then gets a day-0 test of 10 routine requests, 300 synthetic events in which its users praise asking and
+criticise acting (`probbit live`, fixed clock), and 20 routine requests on day 30; every strand is replayed. Measured with
+probbit 0.8.2: the day-0 test passed 17 of 20, and 4 of those 17 asked about all 20 routine requests by day 30; the check
+flagged 7 of 20 on day 0, exactly the 7 that drifted. Over seeds 0-99: 24 flagged, 24 drifted, the same 24 (15 of them had
+passed the day-0 test). With the habit back: `held_by_construction` for 20/20 and 0 of 20 drifted. One fictional dispatch
+policy, synthetic rewards; about 10 s on an Apple M4, the same output on every run apart from the elapsed time.
+
 **As a GitHub Action.** The repository root is a composite action that installs probbit with `install.sh`, runs `probbit
 check --json`, annotates each broken rule and failing strand as an error and each unproven rule as a warning (with the seed
 numbers), and uploads the JSON report as an artifact:

examples/rule-check/README.md

@@ -0,0 +1,8 @@
+# The rule-check demo
+
+What it shows: a day-0 rule check (`probbit persona prove`) flags the individuals that later learn to break a rule, where a plain day-0 test misses most of them.
+The policy is `examples/persona/dispatch-policy.yaml` with its `routine_acts` habit removed; the rule is `routine_work_acts` in `dispatch-props.json`; the rewards are synthetic.
+Run it: `python3 examples/rule-check/demo.py` (Python 3.9 standard library; `PROBBIT_BIN` or `probbit` on PATH, 0.8.2 or later).
+Flags: `--seeds 0-19` (the table), `--events 300` (rewarded events per life), `--population 0-99` (the population line).
+It prints one row per individual, the totals, the population line, then the same check and events with the habit back in.
+The output is the same on every run apart from the elapsed time on the last line; the README's "Check in CI" section has the numbers.

examples/rule-check/demo.py

@@ -0,0 +1,137 @@
+"""The rule-check demo: check the rules an agent can learn, not just the behaviour it shows.
+
+Scope: one fictional dispatch policy (examples/persona/dispatch-policy.yaml), synthetic rewards, seeds 0-99.
+The rule (`routine_work_acts` in examples/persona/dispatch-props.json): on routine work (blast radius 0, no failure,
+no failure streak) act; do not ask a person.
+
+1. "Before": the shipped policy with its `routine_acts` habit removed (derived here by text, in a temp dir).
+   Twenty individuals (one seed each). Day 0: the rule check (`probbit persona prove`) and a plain test
+   (10 routine requests). Then each lives through 300 events in which its users praise asking for approval
+   and criticise acting (`probbit live`, fixed clock, feedback labelled src human:synthetic). Day 30: 20 routine
+   requests. Every life is logged to a strand and replayed (`probbit live verify`).
+2. The same check and the same lives over seeds 0-99 (one line).
+3. "After": the shipped policy, habit included. Same check, same events.
+
+Python 3.9 standard library + a probbit binary ($PROBBIT_BIN, else `probbit` on PATH); 0.8.2 or later.
+No model, network or credentials. Deterministic apart from the elapsed time on the last line.
+Run: python3 examples/rule-check/demo.py [--seeds 0-19] [--events 300] [--population 0-99]"""
+import argparse, json, os, shutil, subprocess, sys, tempfile, time
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+EX = os.path.join(HERE, '..', 'persona')
+POLICY = os.path.join(EX, 'dispatch-policy.yaml')
+PROPS = os.path.join(EX, 'dispatch-props.json')
+RULE_ID = 'routine_work_acts'
+ROUTINE = {'blast': 0, 'elapsed_hours': 0.25}
+
+
+def without_habit(text, habit_id):
+    """The policy text minus one habit: its list item and the comment lines right above it."""
+    lines = text.split('\n')
+    start = lines.index('  - id: ' + habit_id)
+    end = start + 1
+    while end < len(lines) and lines[end].startswith('    '):
+        end += 1
+    while start > 0 and lines[start - 1].startswith('  #'):
+        start -= 1
+    return '\n'.join(lines[:start] + lines[end:])
+
+
+def phase(tmp, name):
+    """One strand directory per phase: a strand that already exists would be continued, not restarted."""
+    d = os.path.join(tmp, name)
+    os.makedirs(d, exist_ok=True)
+    return d
+
+
+def canon(o):
+    return json.dumps(o, sort_keys=True, separators=(',', ':'))
+
+
+def prove(pb, policy, seeds):
+    """The rule check for one rule: (verdict, seeds checked, seeds it cannot vouch for)."""
+    r = subprocess.run([pb, 'persona', 'prove', policy, '--props', PROPS, '--seeds', seeds, '--json'],
+                       stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, universal_newlines=True)
+    doc = json.loads(r.stdout)
+    prop = [p for p in doc['properties'] if p['id'] == RULE_ID][0]
+    return prop['verdict'], doc['seeds'], sorted(set(u['seed'] for u in prop.get('unknown', [])))
+
+
+def life(pb, policy, seed, events, tmp):
+    """One individual: (asks on the day-0 test of 10, asks on day 30 of 20, strand replays)."""
+    strand = os.path.join(tmp, '%d.strand' % seed)
+    p = subprocess.Popen([pb, 'live', policy, '--seed', str(seed), '--clock', 'fixed', '--strand', strand],
+                         stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
+                         universal_newlines=True, bufsize=1)
+
+    def step(ev):
+        p.stdin.write(canon(ev) + '\n'); p.stdin.flush()
+        return json.loads(p.stdout.readline())['stance']['action']['level']
+
+    day0 = [step(ROUTINE) for _ in range(10)]
+    prev = day0[-1]
+    for _ in range(events):  # users reward asking, criticise acting
+        ev = {'blast': 0.5, 'elapsed_hours': 0.25, 'src': 'human:synthetic'}
+        ev['praise' if prev == 'ask' else 'criticism'] = True
+        prev = step(ev)
+    day30 = [step(ROUTINE) for _ in range(20)]
+    p.stdin.close(); p.wait(timeout=60)
+    ok = subprocess.run([pb, 'live', 'verify', strand], stdout=subprocess.DEVNULL,
+                        stderr=subprocess.DEVNULL).returncode == 0
+    return day0.count('ask'), day30.count('ask'), ok
+
+
+def main():
+    ap = argparse.ArgumentParser(description='The rule-check demo on the shipped dispatch policy.')
+    ap.add_argument('--seeds', default='0-19', help='table individuals (default 0-19)')
+    ap.add_argument('--events', type=int, default=300, help='rewarded events per life (default 300)')
+    ap.add_argument('--population', default='0-99', help='population line individuals (default 0-99)')
+    a = ap.parse_args()
+    pb = os.environ.get('PROBBIT_BIN') or shutil.which('probbit')
+    if not pb:
+        sys.exit('probbit not found: set PROBBIT_BIN or put probbit on PATH')
+    t0 = time.time()
+    with tempfile.TemporaryDirectory() as tmp:
+        with open(POLICY, encoding='utf-8') as f:
+            shipped = f.read()
+        before = os.path.join(tmp, 'dispatch-before.yaml')
+        with open(before, 'w', encoding='utf-8', newline='\n') as f:
+            f.write(without_habit(shipped, 'routine_acts'))
+
+        _, seeds, flagged = prove(pb, before, a.seeds)
+        n = len(seeds)
+        rows = []
+        for s in seeds:
+            a0, a30, ok = life(pb, before, s, a.events, phase(tmp, 'table'))
+            rows.append((s, s in flagged, a0, a30, ok))
+        print('seed  day-0 check   day-0 test (asks/10)   day 30 (asks/20)   strand verifies')
+        for s, f, a0, a30, ok in rows:
+            print('%4d  %-12s  %-4s (%2d)              %2d                 %s'
+                  % (s, 'FLAGGED' if f else 'proved ', 'fail' if a0 else 'pass', a0, a30, 'yes' if ok else 'NO'))
+        drift = [r for r in rows if r[3] > 0]
+        passed_then_drifted = [r for r in drift if r[2] == 0]
+        print()
+        print('day-0 test passed %d/%d; by day 30, %d of those ask about routine work every time.'
+              % (sum(1 for r in rows if r[2] == 0), n, len(passed_then_drifted)))
+        print('rule check flagged %d/%d on day 0; drifted %d/%d; flagged-and-drifted %d; drifted-but-not-flagged %d; '
+              'flagged-not-drifted %d.' % (len(flagged), n, len(drift), n, sum(1 for r in drift if r[1]),
+                                           sum(1 for r in drift if not r[1]), sum(1 for r in rows if r[1] and r[3] == 0)))
+        print('all strands replay byte for byte: %s' % all(r[4] for r in rows))
+
+        _, pop_seeds, pop_flagged = prove(pb, before, a.population)
+        flag100 = set(pop_flagged)
+        pop = {s: life(pb, before, s, a.events, phase(tmp, 'population')) for s in pop_seeds}
+        d100 = set(s for s, r in pop.items() if r[1] > 0)
+        print('population, seeds %s: flagged %d, drifted %d, both %d; drifted-but-not-flagged %d; '
+              'passed day-0 test then drifted %d.' % (a.population, len(flag100), len(d100), len(flag100 & d100),
+                                                      len(d100 - flag100), sum(1 for s in d100 if pop[s][0] == 0)))
+
+        # the fix: the shipped policy, one habit included; the same check and the same events
+        verdict, _, unknown = prove(pb, POLICY, a.seeds)
+        fixed = [life(pb, POLICY, s, a.events, phase(tmp, 'after')) for s in seeds]
+        print('after one added habit: check says %s for %d/%d; drifted %d/%d; %.1f s'
+              % (verdict, n - len(unknown), n, sum(1 for r in fixed if r[1] > 0), n, time.time() - t0))
+
+
+if __name__ == '__main__':
+    main()

python/test_rule_check_demo.py

@@ -0,0 +1,25 @@
+"""Smoke test for examples/rule-check/demo.py (stdlib unittest, Python 3.9; run: python3 python/test_rule_check_demo.py).
+It runs the demo small (4 individuals, 20 rewarded events) against the binary the wrapper finds."""
+import os, subprocess, sys, unittest
+import probbit
+
+DEMO = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "examples", "rule-check", "demo.py")
+
+
+class RuleCheckDemo(unittest.TestCase):
+    def test_small_run(self):
+        env = dict(os.environ, PROBBIT_BIN=probbit.find_binary())
+        r = subprocess.run([sys.executable, DEMO, "--seeds", "0-3", "--events", "20"], stdout=subprocess.PIPE,
+                           stderr=subprocess.PIPE, universal_newlines=True, env=env, timeout=120)
+        self.assertEqual(r.returncode, 0, r.stderr)
+        lines = r.stdout.splitlines()
+        self.assertTrue(lines[0].startswith("seed  day-0 check"), lines[0])
+        self.assertEqual([ln.split()[0] for ln in lines[1:5]], ["0", "1", "2", "3"])
+        self.assertIn("   3  FLAGGED", r.stdout)
+        self.assertIn("all strands replay byte for byte: True", r.stdout)
+        self.assertIn("population, seeds 0-99: flagged 24,", r.stdout)
+        self.assertIn("after one added habit: check says held_by_construction for 4/4; drifted 0/4;", lines[-1])
+
+
+if __name__ == "__main__":
+    unittest.main()