Verify exact task metadata baseline and readback

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786

tests/test_task_execution_readback.py

@@ -0,0 +1,127 @@
+"""Exact disposable native acceptance, partial effects and restart refusal."""
+import json
+from dataclasses import replace
+import pytest
+from test_task_execution_native_fixture import owner,bao_dev,compiler,admit
+from task_execution_native_fixture import NativeFixtureAdapter
+from task_execution_journal import Refused,Journal
+from task_execution_cleanup import run_cleanup
+
+
+def start(native,journal,binding):
+    admit(native,journal,binding);native.materialize_grant()
+    journal.transition(binding,'grant-issued','applying')
+
+
+def test_expectation_digest_pin_rejected_before_reserve(owner):
+    n,j,b,c,compiler,*_=owner
+    with pytest.raises(Refused,match='expected-state digest'):
+        NativeFixtureAdapter(compiler,n.plan,j,b,replace(n.admission,expected_after=b'{}'))
+    assert c('GET','sys/policies/acl/'+n.policy_name)[0]==404
+
+
+@pytest.mark.parametrize('target',[0,3])
+def test_actual_conflicting_baseline_gates_its_target_write(owner,target):
+    n,j,b,c,compiler,*_=owner
+    call=compiler.before_call(n.plan,target+1)
+    existing={'policy':'path "unrelated" { capabilities = ["read"] }'} if target==0 else json.loads(call.payload)
+    assert c('PUT',call.path,existing)[0] in {200,204}
+    with j.locked():
+        start(n,j,b)
+        for i in range(target): n.metadata_call(i)
+        with pytest.raises(Refused,match='baseline or readback'):n.metadata_call(target)
+        assert j.observe(b)['state']=='unknown'
+        if target==3:
+            # Policy effect already happened: this is deliberately no aggregate CAS.
+            assert c('GET',compiler.before_call(n.plan,1).path)[1]['data']['policy']==json.loads(compiler.before_call(n.plan,1).payload)['policy']
+        assert run_cleanup(j,b,n)['confirmed']
+        assert j.observe(b)['state']=='unknown'
+
+
+@pytest.mark.parametrize('target',[2,5])
+def test_actual_readback_drift_never_verifies(owner,target):
+    n,j,b,c,compiler,*_=owner
+    with j.locked():
+        start(n,j,b)
+        for i in range(target):n.metadata_call(i)
+        call=compiler.before_call(n.plan,target-1)
+        body=json.loads(call.payload)
+        if target==2:body['policy']+='\npath "sibling/*" { capabilities = ["read"] }'
+        else:body['policies']+=',unexpected-policy'
+        assert c('PUT',call.path,body)[0] in {200,204}
+        with pytest.raises(Refused,match='baseline or readback'):n.metadata_call(target)
+        with pytest.raises(Refused):n.verify_metadata()
+        assert run_cleanup(j,b,n)['confirmed']
+        assert j.observe(b)['state']=='unknown'
+
+
+@pytest.mark.parametrize('status,response',[(403,{'errors':[]}), (404,{}),(404,{'errors':['permission denied']}), (200,{'data':None})])
+def test_only_exact_backend_absence_counts(status,response):
+    with pytest.raises(Refused):NativeFixtureAdapter._metadata_snapshot('auth/kubernetes/role/x',status,response)
+
+
+def test_json_types_and_unknown_role_fields_remain_significant(owner):
+    n,j,b,c,compiler,*_=owner
+    original=n._http
+    def changed(method,path,token,body=b''):
+        status,response=original(method,path,token,body)
+        if method=='GET' and path==compiler.before_call(n.plan,5).path and status==200:
+            response['data']['token_num_uses']=False # False == 0 in Python; never canonical JSON.
+        return status,response
+    n._http=changed
+    with j.locked():
+        start(n,j,b)
+        for i in range(5):n.metadata_call(i)
+        with pytest.raises(Refused):n.metadata_call(5)
+        assert run_cleanup(j,b,n)['confirmed']
+        assert j.observe(b)['state']=='unknown'
+
+
+def test_reopened_sqlite_cannot_resume_or_recover_memory_handles(owner):
+    n,j,b,c,compiler,*_=owner
+    with j.locked():
+        start(n,j,b)
+        n.metadata_call(0);n.metadata_call(1)
+    # Model loss of the live admission; handles deliberately are not reconstructed.
+    # The still-running disposable server proves native effect survives reopening.
+    location=j.directory
+    assert c('GET',compiler.before_call(n.plan,1).path)[0]==200
+    j.db.close()
+    # Owner journal retains the exact DB path; this is a real reopened SQLite.
+    reopened=Journal(location)
+    try:
+        with reopened.locked():
+            assert reopened.observe(b)['dispatch_allowed'] is False
+            with pytest.raises(Refused):reopened.reserve(b,n.cleanup)
+            with pytest.raises(Refused):reopened.next_call_ordinal(b)
+    finally:reopened.close()
+    # No recovery/cleanup claim: only same-process fixture teardown retains handles.
+
+
+def test_verified_readback_with_real_cleanup_denial_is_not_complete(owner):
+    n,j,b,c,compiler,*_=owner
+    with j.locked():
+        start(n,j,b)
+        for i in range(6):n.metadata_call(i)
+        n.verify_metadata()
+        # Actual backend removal denies the task's self-revoke capability.
+        assert c('DELETE','sys/policies/acl/'+n.policy_name)[0] in {200,204}
+        result=run_cleanup(j,b,n)
+        assert result['confirmed'] is False
+        assert j.observe(b)['state']=='verified'
+        with pytest.raises(Refused):j.transition(b,'verified','cleanup-confirmed','0'*64)
+        assert c('POST','auth/token/lookup-accessor',{'accessor':n.task.accessor})[0]==200
+
+
+def test_actual_denied_baseline_cannot_be_caught_then_write_retried(owner):
+    n,j,b,c,compiler,*_=owner
+    with j.locked():
+        start(n,j,b)
+        # Task still has update but loses read: real 403 is not absent baseline.
+        denied=n.hcl.replace('["read", "create", "update"]','["create", "update"]')
+        assert c('PUT','sys/policies/acl/'+n.policy_name,{'policy':denied})[0] in {200,204}
+        with pytest.raises(Refused):n.metadata_call(0)
+        assert j.observe(b)['state']=='unknown'
+        with pytest.raises(Refused):n.metadata_call(1)
+        assert c('GET',compiler.before_call(n.plan,1).path)[0]==404
+        assert run_cleanup(j,b,n)['confirmed']