From 76fdc0b861ec2ee7f6512dc5023ffc0cee5940f9 Mon Sep 17 00:00:00 2001
From: codex <codex@local>
Date: Fri, 16 Oct 2026 17:15:00 +0200
Subject: [PATCH] Bound task execution by absolute call and credential
 deadlines

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786
---
 docs/task-execution-native-fixture.md         |  53 ++++++
 scripts/task_execution_mediator.py            |  40 ++++-
 scripts/task_execution_native_fixture.py      |  96 +++++++++-
 tests/test_task_execution_native_budget.py    | 167 ++++++++++++++++++
 ...-WP-0035-credential-lane-implementation.md |   9 +
 5 files changed, 356 insertions(+), 9 deletions(-)
 create mode 100644 tests/test_task_execution_native_budget.py

diff --git a/docs/task-execution-native-fixture.md b/docs/task-execution-native-fixture.md
index f4ff070..c4292db 100644
--- a/docs/task-execution-native-fixture.md
+++ b/docs/task-execution-native-fixture.md
@@ -205,3 +205,56 @@ changes during its single read. A valid older prepared capsule still cannot
 prove a task token absent when durable journal state/calls reached issuance:
 that remains unknown cleanup debt. Authentication alone is not anti-rollback.
 No business or consume admission is restored and no production scope is granted.
+
+
+## Absolute budgets, effective credential expiry and exact source bytes
+
+This source successor tightens the disposable native owner adapter; it does not
+establish actual v2 whole-stack ALLOW/native acceptance. The retained-registry
+Approval correspondence failure is preserved in the coordinator artifact. The
+new real provider fixture is retained separately as unfinished follow-up, while
+the existing ec215-based fixture evidence remains its original scoped proof.
+
+Every native business/grant method now runs inside the earlier wall/monotonic
+operation deadline minus the five-second reserve. After exact owner token lookup,
+the adapter requires aware expire_time and retains the earlier actual expiry and
+conservative mint-start + declared TTL bound, in wall and monotonic coordinates.
+Every subsequent business call uses the minimum of global and credential bounds.
+The descriptor and signed global deadline are never extended or rewritten.
+Missing/expired credential evidence refuses issuance acceptance. Cleanup continues
+under its original separate obligation after either business bound expires.
+
+The fixture owner main-thread execution profile uses absolute SIGALRM budgets,
+not socket inactivity timeouts alone. Each native transport and cleanup native
+call has a two-second execution deadline, including intent/source checking and
+send/header/body/close work. Larger compiler reconstruction remains within the
+earlier operation/credential window. A 50ms alarm backstop keeps blocked teardown
+interruptible after timeout; it cannot authorize another business effect. Nested
+timers retain the shorter outer deadline rather than refreshing it. Unsupported
+threads, invalid budgets and conflicting ambient timers refuse. This fixed Unix
+fixture profile does not select a production process supervisor or provide a
+shared arbitrary-thread execution service. Native rollout still needs exact owner
+runtime/admission, trusted clock and independent recovery/custody enforcement.
+
+The actual v2 package/version selects additional mediator checks over the exact
+independently installed descriptor source_bundle before authorization and each
+freshness gate, including pre-native grant entry. It reads bounded regular,
+files without group/world write, non-indirect paths, compares exact source hashes and stat stability,
+and applies an absolute read budget. The existing compiler's own pin set does not
+implicitly authenticate other source_bundle roles; these explicit checks fill
+that gap. No requester-selected path, approval flag or source boolean can enable
+this path. Legacy synthetic fixture proof remains separately scoped and is not
+used to satisfy v2 acceptance.
+
+Native policy/grant absence now requires exact empty-errors404 and policy HCL
+comparison admits only the already reviewed single optional terminal LF, reusing
+frozen recovery R2 helpers. Native response JSON rejects duplicate/nonfinite data.
+
+The independent slice tests cover actual header/body drips, stalled close after
+timeout, slow compiler and shorter actual credential expiry before any HTTP
+effect, nested/conflicting/foreign-thread budgets, private source hash drift,
+unsafe modes/indirection/FIFO/oversize refusal, and bounded cleanup after expiry.
+Actual native fixed grant/metadata/readback/cleanup remains a separate positive
+compatibility check. There is no successful actual normalized provider-chain
+claim in this slice; source-managed pre-approval normalization and immediate CAS
+precision must pass the follow-up full-stack proof first.
diff --git a/scripts/task_execution_mediator.py b/scripts/task_execution_mediator.py
index 6614974..ec9d235 100644
--- a/scripts/task_execution_mediator.py
+++ b/scripts/task_execution_mediator.py
@@ -10,13 +10,15 @@ from dataclasses import dataclass
 from datetime import datetime,timezone
 import importlib
 import json
+import os
+import stat
 from pathlib import Path
 from urllib.parse import urlsplit
 
 from task_execution_cleanup import run_cleanup
 from task_execution_journal import Binding,Cleanup,Journal,Refused
 from task_execution_plan import Compiler,encoded,content_hash
-from task_execution_native_fixture import FixtureAdmission,NativeFixtureAdapter
+from task_execution_native_fixture import FixtureAdmission,NativeFixtureAdapter,_absolute_budget
 
 
 @dataclass(frozen=True)
@@ -43,7 +45,36 @@ def _strict_dependency():
         raise Refused('strict Secrets task execution handoff dependency unavailable') from None
 
 
+def _trusted_v2(cfg):
+    return (getattr(cfg,'authorization_policy_package',None)=='secrets-engine.task-execution.service-pilot'
+            and getattr(cfg,'authorization_policy_version',None)=='v2')
+
+
+def _read_source_bundle(descriptor):
+    # Paths/hash expectations come from independently installed owner_descriptor,
+    # never a requester-selected file lookup. No bytes or raw OS errors escape.
+    for item in descriptor['source_bundle'].values():
+        path=Path(item['path']);fd=None
+        try:
+            if not path.is_absolute() or path.resolve()!=path:raise ValueError()
+            fd=os.open(path,os.O_RDONLY|os.O_NOFOLLOW|os.O_NONBLOCK)
+            before=os.fstat(fd)
+            if not stat.S_ISREG(before.st_mode) or before.st_nlink!=1 or before.st_mode&0o022 or not 0<before.st_size<=1048576:raise ValueError()
+            with os.fdopen(fd,'rb',closefd=False) as stream:raw=stream.read(1048577)
+            after=os.fstat(fd)
+            identity=lambda value:(value.st_dev,value.st_ino,value.st_size,value.st_mtime_ns,value.st_ctime_ns,value.st_mode,value.st_nlink)
+            if identity(before)!=identity(after) or len(raw)!=before.st_size or 'sha256:'+content_hash(raw)!=item['sha256']:raise ValueError()
+        except Exception:raise Refused('exact owner execution source bundle refused') from None
+        finally:
+            if fd is not None:os.close(fd)
+
+
+def _source_bundle(descriptor):
+    with _absolute_budget(2):_read_source_bundle(descriptor)
+
+
 def _fresh(cfg,handoff,descriptor):
+    if _trusted_v2(cfg):_source_bundle(descriptor)
     from secrets_engine.application_time import read_window,validity_bounds
     lower,upper=validity_bounds(window=read_window(cfg))
     try:
@@ -79,8 +110,9 @@ def run_native_fixture_task(*,compiler:Compiler,journal:Journal,plan_request:dic
     Strict authorization is executed here through the actual Secrets-owned module;
     an inbound TaskExecutionAuthorization or arbitrary callable is never accepted.
     This function returns closed metadata, not tokens or raw approval/decision data.
-    Positive execution requires the module's real signed PDP path; actual unadmitted
-    Access policy must still deny. Fixture policy/identity admission is independent
+    Positive execution requires the module's real signed PDP path; the retained v1
+    Access policy stays DENY-only; actual v2 successor source is separately tested
+    through its authenticated caller/fresh-source path without runtime admission. Fixture policy/identity admission is independent
     synthetic setup and cannot become production acceptance.
     """
     strict=_strict_dependency()
@@ -99,6 +131,7 @@ def run_native_fixture_task(*,compiler:Compiler,journal:Journal,plan_request:dic
             raise ValueError()
     except Exception:
         raise Refused('exact owner descriptor or provider route refused') from None
+    if _trusted_v2(secrets_cfg):_source_bundle(descriptor)
     # An independently pinned owner catalogue must match every engine field. Its
     # approval IDs are source references, never runtime authority.
     proposal=compiler.proposal()
@@ -163,6 +196,7 @@ def run_native_fixture_task(*,compiler:Compiler,journal:Journal,plan_request:dic
             # Fixture identity is independently seeded synthetic setup, not CCR38
             # provenance proof or a production service login.
             journal.transition(binding,'authenticating','grant-issuing')
+            _fresh(secrets_cfg,handoff,descriptor)
             native.materialize_grant()
             journal.transition(binding,'grant-issued','applying')
             for ordinal in range(6):
diff --git a/scripts/task_execution_native_fixture.py b/scripts/task_execution_native_fixture.py
index 31a822b..331a2d5 100644
--- a/scripts/task_execution_native_fixture.py
+++ b/scripts/task_execution_native_fixture.py
@@ -9,7 +9,12 @@ Python privacy is not credential isolation; production needs the owner runtime.
 """
 from __future__ import annotations
 from dataclasses import dataclass, field
-from datetime import datetime, timezone
+from contextlib import contextmanager
+from functools import wraps
+import signal
+import math
+import threading
+from datetime import datetime, timezone, timedelta
 import hashlib
 import http.client
 import json
@@ -19,10 +24,63 @@ from pathlib import Path
 from urllib.parse import urlsplit
 
 from task_execution_cleanup import CleanupResult
+from task_execution_recovery import _same_policy,_policy_absent,_json_unique,_invalid_constant
 from task_execution_journal import Binding, Cleanup, Journal, Refused
 from task_execution_plan import Compiler, Plan, cleanup_specs, encoded, content_hash
 
 
+def _budget_expired(_signum, _frame):
+    # Keep blocked teardown interruptible after the first execution timeout.
+    # This is unwind-only grace, never a refreshed native-effect budget.
+    signal.setitimer(signal.ITIMER_REAL,.05)
+    raise Refused('absolute native call budget exhausted')
+
+
+@contextmanager
+def _absolute_budget(seconds):
+    """Disposable owner-child bound; refuse unsupported/conflicting execution.
+
+    SIGALRM interrupts slow source reads/compilation and HTTP header/body drips.
+    Nested bounds preserve the shorter outer deadline; no timer refresh extends
+    authority. This fixture assumes its owner main-thread process, not a shared
+    arbitrary-thread production runtime.
+    """
+    if (threading.current_thread() is not threading.main_thread() or type(seconds) not in {int,float}
+            or not math.isfinite(seconds) or seconds<=0):
+        raise Refused('bounded owner main-thread execution required')
+    previous=signal.getsignal(signal.SIGALRM)
+    old,interval=signal.getitimer(signal.ITIMER_REAL)
+    if interval or (old>0 and previous is not _budget_expired):
+        raise Refused('conflicting owner execution timer')
+    started=time.monotonic()
+    try:
+        signal.signal(signal.SIGALRM,_budget_expired)
+        signal.setitimer(signal.ITIMER_REAL,min(seconds,old) if old>0 else seconds)
+        yield
+    finally:
+        signal.setitimer(signal.ITIMER_REAL,0)
+        elapsed=time.monotonic()-started
+        signal.signal(signal.SIGALRM,previous)
+        if old>0:
+            if old<=elapsed:raise Refused('outer native authority budget exhausted')
+            signal.setitimer(signal.ITIMER_REAL,old-elapsed)
+
+
+def _business_window(method):
+    @wraps(method)
+    def bounded(self,*args,**kwargs):
+        remaining=self._remaining_business()-5
+        with _absolute_budget(remaining):return method(self,*args,**kwargs)
+    return bounded
+
+
+def _call_window(method):
+    @wraps(method)
+    def bounded(self,*args,**kwargs):
+        with _absolute_budget(2):return method(self,*args,**kwargs)
+    return bounded
+
+
 @dataclass(frozen=True)
 class PrivateSession:
     token: str = field(repr=False)
@@ -101,6 +159,7 @@ class NativeFixtureAdapter:
         if self.meta['cleanup'] != cleanup_specs(binding.operation_id): raise Refused('cleanup plan mismatch')
         self.task=None; self._mint_attempted=False; self._policy_attempted=False; self._ordinal=0; self._next_metadata=0
         self._cleanup_steps={item.kind:0 for item in self.cleanup}
+        self._task_wall_deadline=None;self._task_monotonic_deadline=None
         self.recovery_custody=recovery_custody
         if recovery_custody is not None:
             from task_execution_recovery import FixtureRecoveryCustody
@@ -112,9 +171,17 @@ class NativeFixtureAdapter:
         if not self.journal._held or self.journal._admission!=self.binding or self.journal._consumed_admission!=self.binding:
             raise Refused('same live consumed owner admission required')
 
+    def _remaining_business(self):
+        bounds=[(self.deadline-datetime.now(timezone.utc)).total_seconds(),self.monotonic_deadline-time.monotonic()]
+        task_wall=getattr(self,'_task_wall_deadline',None)
+        task_mono=getattr(self,'_task_monotonic_deadline',None)
+        if task_wall is not None:bounds.append((task_wall-datetime.now(timezone.utc)).total_seconds())
+        if task_mono is not None:bounds.append(task_mono-time.monotonic())
+        return min(bounds)
+
     def _fresh(self):
         self._live()
-        if min((self.deadline-datetime.now(timezone.utc)).total_seconds(),self.monotonic_deadline-time.monotonic())<=5:
+        if self._remaining_business()<=5:
             raise Refused('business deadline exhausted')
         # The constructor independently rebuilt the complete immutable Plan and
         # grant. Recheck every pinned byte before each native effect. Metadata
@@ -122,6 +189,7 @@ class NativeFixtureAdapter:
         # creation reconstructs and compares its fixed body before issuance.
         for path in self.compiler.pins:self.compiler.read(Path(path))
 
+    @_call_window
     def _http(self, method, path, token, body=b''):
         # Fixed caller-generated paths only; http.client has no proxy or redirect
         # following. Never surface backend body/header or transport exception text.
@@ -133,13 +201,15 @@ class NativeFixtureAdapter:
                          headers={'X-Vault-Token':token,'Content-Type':'application/json'})
             response=conn.getresponse(); raw=response.read(1024*1024+1)
             if len(raw)>1024*1024: raise Refused('oversize native response')
-            data=json.loads(raw) if raw else {}
+            data=json.loads(raw,object_pairs_hook=_json_unique,parse_constant=_invalid_constant) if raw else {}
             if type(data) is not dict: raise Refused('invalid native response')
             return response.status,data
         except Exception:
             raise Refused('native transport or response unconfirmed') from None
         finally: conn.close()
 
+    @_business_window
+    @_call_window
     def _business(self, method, path, token, payload=b'', task=False):
         self._fresh()
         if task:
@@ -165,6 +235,7 @@ class NativeFixtureAdapter:
         if status not in allowed or data.get('errors'):
             raise Refused('native request refused or unverified')
 
+    @_business_window
     def materialize_grant(self):
         self._fresh()
         if self.journal.observe(self.binding)['state']!='grant-issuing' or self._policy_attempted:
@@ -173,12 +244,12 @@ class NativeFixtureAdapter:
             raise Refused('insufficient lifetime for exact fixed task TTL')
         root=self.admission.owner_token; path='sys/policies/acl/'+self.policy_name
         status,data=self._business('GET',path,root)
-        if status!=404 or data.get('data'):
+        if not _policy_absent(status,data):
             raise Refused('per-operation grant policy must be absent')
         self._policy_attempted=True
         status,data=self._business('PUT',path,root,encoded({'policy':self.hcl}));self._ok(status,data,{200,204})
         status,data=self._business('GET',path,root);self._ok(status,data,{200})
-        if data.get('data',{}).get('policy','').rstrip()!=self.hcl.rstrip(): raise Refused('grant policy readback mismatch')
+        if not _same_policy(data.get('data',{}).get('policy'),self.hcl): raise Refused('grant policy readback mismatch')
         body={'policies':[self.policy_name],'no_default_policy':True,'renewable':False,
               'ttl':str(self.ttl)+'s','explicit_max_ttl':str(self.ttl)+'s','num_uses':self.uses,
               'type':'service','display_name':'task-exec-'+self.binding.operation_id,'no_parent':False}
@@ -187,6 +258,7 @@ class NativeFixtureAdapter:
         if self.recovery_custody is not None:
             self.recovery_custody.checkpoint(self.binding,'mint-intent')
         self._mint_attempted=True
+        minted_wall=datetime.now(timezone.utc);minted_mono=time.monotonic()
         status,data=self._business('POST','auth/token/create',root,encoded(body));self._ok(status,data,{200})
         auth=data.get('auth',{})
         if type(auth.get('client_token')) is not str or not auth['client_token'] or type(auth.get('accessor')) is not str or not auth['accessor']:
@@ -202,9 +274,19 @@ class NativeFixtureAdapter:
                 or info.get('num_uses')!=self.uses or not 0<info.get('ttl',0)<=self.ttl
                 or info.get('explicit_max_ttl')!=self.ttl or info.get('orphan') is not False):
             raise Refused('native task token bounds mismatch')
+        try:
+            expiry=datetime.fromisoformat(info['expire_time'].replace('Z','+00:00'))
+            if expiry.tzinfo is None or expiry.utcoffset() is None:raise ValueError()
+            self._task_wall_deadline=min(expiry,minted_wall+timedelta(seconds=self.ttl))
+            self._task_monotonic_deadline=min(minted_mono+self.ttl,
+                time.monotonic()+(self._task_wall_deadline-datetime.now(timezone.utc)).total_seconds())
+            if self._remaining_business()<=5:raise ValueError()
+        except (KeyError,TypeError,ValueError,AttributeError):
+            raise Refused('native task expiry unconfirmed or exhausted') from None
         self.journal.transition(self.binding,'grant-issuing','grant-issued')
         return {'issued':True,'policy_name':self.policy_name,'ttl_ceiling_seconds':self.ttl,'use_ceiling':self.uses}
 
+    @_business_window
     def metadata_call(self, ordinal):
         self._fresh()
         if self.task is None or self.journal.observe(self.binding)['state']!='applying' or type(ordinal) is not int or not 0<=ordinal<6 or ordinal!=self._next_metadata:
@@ -245,6 +327,7 @@ class NativeFixtureAdapter:
         # exact canonical JSON comparison. True never equals integer 1 here.
         return value
 
+    @_business_window
     def verify_metadata(self):
         self._fresh()
         if self._next_metadata!=6 or self.journal.observe(self.binding)['state']!='applying':
@@ -267,6 +350,7 @@ class NativeFixtureAdapter:
         if digest!=expected.call_digest or actual['call_digest']!=digest or actual['outcome']!='pending':
             raise Refused('exact pending cleanup obligation required')
 
+    @_call_window
     def _cleanup_http(self,kind,method,path,token,body=b''):
         ordinal=next(i for i,item in enumerate(self.cleanup) if item.kind==kind)
         obligation=self.cleanup[ordinal];step=self._cleanup_steps[kind]
@@ -307,5 +391,5 @@ class NativeFixtureAdapter:
         if self._policy_attempted:
             status,data=self._cleanup_http('grant-policy-delete','DELETE',path,self.admission.owner_token);self._ok(status,data,{200,204})
         status,data=self._cleanup_http('grant-policy-delete','GET',path,self.admission.owner_token)
-        confirmed=status==404 and not data.get('data')
+        confirmed=_policy_absent(status,data)
         return CleanupResult(confirmed,content_hash(encoded({'kind':'grant-policy-delete','absence_confirmed':confirmed})))
diff --git a/tests/test_task_execution_native_budget.py b/tests/test_task_execution_native_budget.py
new file mode 100644
index 0000000..75cc126
--- /dev/null
+++ b/tests/test_task_execution_native_budget.py
@@ -0,0 +1,167 @@
+"""Absolute native budgets/source pins; no actual-v2 positive admission claim."""
+from datetime import datetime,timedelta,timezone
+from pathlib import Path
+import signal
+import sys
+import threading
+import time
+from types import SimpleNamespace
+
+import pytest
+sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts'))
+from test_task_execution_native_fixture import owner,bao_dev,compiler
+from task_execution_native_fixture import NativeFixtureAdapter,_absolute_budget
+from task_execution_journal import Refused
+
+
+def test_absolute_budget_interrupts_source_work_and_restores_timer():
+    before=signal.getitimer(signal.ITIMER_REAL)
+    started=time.monotonic()
+    with pytest.raises(Refused):
+        with _absolute_budget(.03):time.sleep(.5)
+    assert time.monotonic()-started<.3
+    assert signal.getitimer(signal.ITIMER_REAL)==before
+
+
+def test_nested_budget_cannot_refresh_outer_deadline():
+    started=time.monotonic()
+    with pytest.raises(Refused):
+        with _absolute_budget(.05):
+            time.sleep(.02)
+            with _absolute_budget(1):time.sleep(.5)
+    assert time.monotonic()-started<.3
+
+
+def test_absolute_budget_rejects_foreign_threads():
+    refused=[]
+    def run():
+        try:
+            with _absolute_budget(1):pass
+        except Refused:refused.append(True)
+    worker=threading.Thread(target=run);worker.start();worker.join(2)
+    assert refused==[True]
+
+
+def test_slow_compiler_is_interrupted_before_native_effect():
+    native=object.__new__(NativeFixtureAdapter)
+    native.deadline=datetime.now(timezone.utc)+timedelta(seconds=5.05)
+    native.monotonic_deadline=time.monotonic()+5.05
+    native.binding='synthetic';native.plan=None;native.task=object();native._next_metadata=0
+    native.journal=SimpleNamespace(_held=True,_admission='synthetic',_consumed_admission='synthetic',observe=lambda binding:{'state':'applying'})
+    native.compiler=SimpleNamespace(pins=[],before_call=lambda *args:time.sleep(.5))
+    sent=[];native._business=lambda *args,**kwargs:sent.append(True)
+    with pytest.raises(Refused):native.metadata_call(0)
+    assert sent==[]
+
+
+@pytest.mark.parametrize('phase',['headers','body','body-and-slow-close'])
+def test_absolute_transport_budget_interrupts_drip(phase,monkeypatch):
+    from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer
+    class Drip(BaseHTTPRequestHandler):
+        def log_message(self,*args):pass
+        def do_GET(self):
+            if phase=='headers':
+                try:
+                    for value in b'HTTP/1.1 200 OK\r\nX-Drip: '+b' '*1000:
+                        self.connection.sendall(bytes([value]));time.sleep(.05)
+                except (BrokenPipeError,ConnectionResetError):pass
+                return
+            self.send_response(200);self.send_header('Content-Type','application/json');self.send_header('Content-Length','1000');self.end_headers()
+            try:
+                for _ in range(1000):
+                    self.wfile.write(b' ');self.wfile.flush();time.sleep(.05)
+            except (BrokenPipeError,ConnectionResetError):pass
+    server=ThreadingHTTPServer(('127.0.0.1',0),Drip)
+    server.daemon_threads=True
+    thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start()
+    native=object.__new__(NativeFixtureAdapter);native.host='127.0.0.1';native.port=server.server_port
+    native.admission=SimpleNamespace(server_process=SimpleNamespace(poll=lambda:None))
+    if phase=='body-and-slow-close':
+        import http.client
+        original_close=http.client.HTTPConnection.close
+        def slow_close(connection):
+            original_close(connection);time.sleep(5)
+        monkeypatch.setattr(http.client.HTTPConnection,'close',slow_close)
+    start=time.monotonic()
+    try:
+        with pytest.raises(Refused):native._http('GET','drip','synthetic')
+        assert time.monotonic()-start<2.6
+    finally:server.shutdown();server.server_close();thread.join(2)
+
+
+def test_cleanup_remains_independent_but_absolute_after_business_expiry(owner,monkeypatch):
+    from test_task_execution_native_fixture import admit
+    from task_execution_cleanup import run_cleanup
+    native,journal,binding,call,*_=owner
+    with journal.locked():
+        admit(native,journal,binding);native.materialize_grant()
+        original=native._http
+        def stalled(method,path,token,body=b''):
+            if path=='auth/token/revoke-self' and token==native.task.token:time.sleep(5)
+            return original(method,path,token,body)
+        monkeypatch.setattr(native,'_http',stalled)
+        native.monotonic_deadline=time.monotonic()-1
+        native._task_monotonic_deadline=time.monotonic()-1
+        started=time.monotonic();result=run_cleanup(journal,binding,native)
+        assert time.monotonic()-started<3
+        assert not result['confirmed']
+        assert [x['outcome'] for x in result['obligations']]==['unconfirmed','confirmed','confirmed']
+        assert journal.observe(binding)['external_outcome_unknown']
+        # Explicit independent synthetic owner recovery, never adapter retry.
+        assert call('POST','auth/token/revoke-accessor',{'accessor':native.task.accessor})[0] in {200,204}
+
+
+def test_absolute_budget_invalid_input_and_conflicting_timer_refuse():
+    original_handler=signal.getsignal(signal.SIGALRM)
+    original_timer=signal.getitimer(signal.ITIMER_REAL)
+    for value in [True,None,0,-1,float('nan'),float('inf')]:
+        with pytest.raises(Refused):
+            with _absolute_budget(value):pass
+        assert signal.getsignal(signal.SIGALRM)==original_handler
+        assert signal.getitimer(signal.ITIMER_REAL)==original_timer
+    try:
+        signal.signal(signal.SIGALRM,lambda *args:None)
+        signal.setitimer(signal.ITIMER_REAL,1)
+        with pytest.raises(Refused):
+            with _absolute_budget(.1):pass
+        assert signal.getitimer(signal.ITIMER_REAL)[0]>0
+    finally:
+        signal.setitimer(signal.ITIMER_REAL,0)
+        signal.signal(signal.SIGALRM,original_handler)
+
+
+def test_actual_credential_bound_interrupts_before_later_global_deadline():
+    native=object.__new__(NativeFixtureAdapter)
+    native.deadline=datetime.now(timezone.utc)+timedelta(minutes=10)
+    native.monotonic_deadline=time.monotonic()+600
+    native._task_wall_deadline=datetime.now(timezone.utc)+timedelta(seconds=5.05)
+    native._task_monotonic_deadline=time.monotonic()+5.05
+    native.binding='synthetic';native.plan=None;native.task=object();native._next_metadata=0
+    native.journal=SimpleNamespace(_held=True,_admission='synthetic',_consumed_admission='synthetic',observe=lambda binding:{'state':'applying'})
+    native.compiler=SimpleNamespace(pins=[],before_call=lambda *args:time.sleep(.5))
+    sent=[];native._business=lambda *args,**kwargs:sent.append(True)
+    started=time.monotonic()
+    with pytest.raises(Refused):native.metadata_call(0)
+    assert time.monotonic()-started<.3 and sent==[]
+
+
+
+@pytest.mark.parametrize('case',['valid','drift','writable','symlink','fifo','oversize'])
+def test_exact_owner_source_bundle_is_bounded_and_fails_closed(tmp_path,case):
+    import hashlib
+    import os
+    from task_execution_mediator import _source_bundle
+    source=tmp_path/'owner-source.py';source.write_bytes(b'# independently installed source\n');source.chmod(0o600)
+    roles=['engine_dispatcher','platform_mediator','typed_ccr','policy_auth']
+    item={'path':str(source),'sha256':'sha256:'+hashlib.sha256(source.read_bytes()).hexdigest()}
+    descriptor={'source_bundle':{role:dict(item) for role in roles}}
+    if case=='valid':_source_bundle(descriptor);return
+    if case=='drift':source.write_bytes(b'# drift\n')
+    elif case=='writable':source.chmod(0o666)
+    elif case=='symlink':
+        target=tmp_path/'target.py';source.rename(target);source.symlink_to(target)
+    elif case=='fifo':source.unlink();os.mkfifo(source,0o600)
+    elif case=='oversize':source.write_bytes(b'X'*1048577)
+    started=time.monotonic()
+    with pytest.raises(Refused):_source_bundle(descriptor)
+    assert time.monotonic()-started<.5
diff --git a/workplans/RPF-WP-0035-credential-lane-implementation.md b/workplans/RPF-WP-0035-credential-lane-implementation.md
index f58a6b9..92a3f3f 100644
--- a/workplans/RPF-WP-0035-credential-lane-implementation.md
+++ b/workplans/RPF-WP-0035-credential-lane-implementation.md
@@ -1307,3 +1307,12 @@ capsules are owner-key HMAC authenticated, strict nested schemas/ordered cleanup
 commitments reject damage, and journal history prevents stale prepared capsules
 from hiding issuance. Fixture process-loss source proof remains separate from
 unadmitted production custody, integrity-key storage and recovery-role scope.
+
+2026-10-16 T12 independently reviewable budget/source slice: bound native HTTP
+header/body/close, compiler work and cleanup through fixed owner main-thread
+absolute budgets, with actual task expiry narrowing global authority. Actual v2
+mediator now verifies each independently installed source_bundle byte hash before
+authorization/effects. No deadline/TTL extension, remint or generic service added.
+Retained registry actual-provider proof still refuses Approval correspondence
+before CAS/native; normalized pre-approval + precise immediate CAS whole-stack
+positive remains live owner follow-up under CUST-WP-0080. T12 stays progress.