diff --git a/docs/task-execution-handoff.md b/docs/task-execution-handoff.md
new file mode 100644
index 0000000..5821175
--- /dev/null
+++ b/docs/task-execution-handoff.md
@@ -0,0 +1,78 @@
+# Strict Secrets service task-execution handoff
+
+This is a narrow source seam for existing SECRETS-WP-0013, not a native grant or
+an execute-ccr CLI. Both Access runtime-deny policy variants remain unchanged.
+The proposed service subject/transport mapping and the complete native executor
+still need independent owner admission. Positive source fixtures do not admit
+that mapping, a policy, a caller, a production key or an OpenBao entitlement.
+
+Platform's trusted owner process calls:
+
+```python
+from secrets_engine.task_execution_handoff import (
+    authorize_task_execution, consume_task_execution,
+)
+authorization = authorize_task_execution(cfg, entry, execution_contract=contract)
+record = authorization.as_owner_handoff()
+# Independently bind exact owner Plan/descriptor; reserve durable one-shot journal.
+# Persist pending strict-CAS intent before calling:
+consumption = consume_task_execution(cfg, authorization)
+# Independently GET-confirm exact original consumption with Approval Engine's
+# ExecutionConfirmationClient before journaling consume-confirmed/native work.
+```
+
+Preparation uses the existing empty fields/policy_targets/auth_targets defaults
+and existing build_action_request/claim/Check validators. No caller-selected
+transport, authorizer callback, policy result or raw handoff mapping is accepted.
+The normal authorizer retains detached serialized request, claim and authenticated
+Go-typed envelope snapshots; its existing evidence output does not expose them.
+The narrow adapter reuses the validators, rather than evaluating a policy.
+
+The adapter checks the exact descriptor in the signed context, full evaluated and
+submitted request digests, approval-free digest, exact five-field native act and
+native binding digest, service subject/resource/tenant correspondence and original
+decision/approval/deadline bounds. Initiating/represented subject is strictly
+`secrets-engine` of type `service`; the execution actor is `service:secrets-engine`.
+Signed caller provenance must report `enforce`, the proposed exact Secrets
+ServiceAccount principal, `flex-auth` audience and live expiry bounding the
+operation deadline. This consumes the signed PDP's caller attestation; it does
+not independently perform TokenReview or promote the proposed server mapping.
+Human/delegation/projected-identity flags cannot substitute for this service seam.
+No service_actor_binding request propagation or source-policy activation is added.
+
+`as_owner_handoff()` returns a detached non-secret mapping with:
+
+- execution_contract
+- approval_id, request_digest, decision_id
+- native_binding, approval_binding_digest, submitted_request_digest
+- decision_not_before, decision_expires_at, approval_expires_at
+
+The exact native binding is derived from the signed evaluated request using the
+published Approval five-field mapping, and is compared to the served claim.
+Descriptor/CCR/grant/source/catalog/target correspondence is retained end to end;
+Platform still independently compiles its owner Plan and grant from owner bytes.
+The mapping is evidence for that join, never a transferable bearer authorization.
+
+Consumption accepts only the module-created opaque prepared object in that same
+trusted owner process. It fixes source/config coordinates, reauthenticates using
+currently admitted responder keys, and rechecks claim/decision/descriptor/caller
+freshness before reading the CAS credential. Each object is reserved once before
+CAS/revalidation and cannot retry after any failure or unknown outcome. Its
+process-local guard complements Platform's durable cross-process journal; it is
+not a replacement for that journal or process isolation.
+
+The strict response path uses the existing CAS transport but rejects duplicate,
+nonfinite or excessively deep JSON, missing/mismatched approval and decision IDs,
+wrong digest/status, missing or nonboolean idempotence, historical idempotent
+success, verdict fields and absent/naive/invalid consumption timestamps. The
+consumption instant and post-CAS freshness must remain inside the original
+bounds. No field is coerced or filled from the submitted binding. A late/uncertain
+CAS may have occurred; it never licenses retry or protected native work.
+
+The returned immutable TaskExecutionConsumption contains exact approval_id,
+request_digest, decision_id, consumed_at and explicit false idempotent. It proves
+only the strict CAS response observed by Secrets. Platform must independently
+confirm original durable consumption using Approval's real provider client,
+journal it, then recheck its exact Plan, owner sources, grant and deadlines before
+any native operation. Neither a dataclass nor a caller-supplied boolean admits
+execution. Existing generic CLI CAS behavior is otherwise unchanged.
diff --git a/src/secrets_engine/approval_consume.py b/src/secrets_engine/approval_consume.py
index bb1e766..1cf4140 100644
--- a/src/secrets_engine/approval_consume.py
+++ b/src/secrets_engine/approval_consume.py
@@ -23,7 +23,7 @@ from secrets_engine.execution_contract import ExecutionContract, require_executi
 from secrets_engine.approval_claim import observe_pdp_approval_claim, validate_approval_claim
 from secrets_engine.catalog import human_control_required
 from secrets_engine.decision_check import check_decision, require_supported_pdp_address
-from secrets_engine.decision_signature import load_trust
+from secrets_engine.decision_signature import load_trust, strict_json
 from secrets_engine.authorization import (
     build_action_request,
     validate_decision_envelope,
@@ -70,6 +70,8 @@ class AuthorizedAction:
     binding: ConsumeBinding
     decision_id: str
     expires_at: str
+    # Detached original served artifacts for the narrow task-execution handoff.
+    artifacts: str = field(default="", repr=False, compare=False)
 
     @property
     def id(self) -> str:
@@ -380,6 +382,8 @@ def authorize_action(
         ),
         decision_id=validated.decision_id,
         expires_at=validated.expires_at,
+        artifacts=json.dumps({"request":expected_request,"claim":observation.claim,"envelope":envelope},
+                             sort_keys=True,separators=(",",":"),allow_nan=False),
     )
 
 
@@ -391,6 +395,7 @@ def consume_approval(
     binding: ConsumeBinding,
     timeout_seconds: float = 3,
     opener: Callable[..., Any] = credential_urlopen,
+    strict_response: bool = False,
 ) -> ConsumedApproval:
     """POST /v1/approvals/{id}/consume. Fail closed on anything but confirmed use."""
     if not base_url or not base_url.startswith(("http://", "https://")):
@@ -438,8 +443,8 @@ def consume_approval(
     if status != 200 or len(raw) > _MAX_BODY:
         raise DecisionError(_status_message(status if status != 200 else 502))
     try:
-        payload = json.loads(raw.decode("utf-8"))
-    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        payload = (strict_json if strict_response else json.loads)(raw.decode("utf-8"))
+    except (UnicodeDecodeError, ValueError, RecursionError) as exc:
         raise DecisionError("approval consume returned invalid JSON") from exc
     if not isinstance(payload, dict):
         raise DecisionError("approval consume returned invalid payload")
@@ -447,6 +452,19 @@ def consume_approval(
         raise DecisionError("approval consumption was not confirmed")
     if payload.get("request_digest") != binding.request_digest:
         raise DecisionError("approval consume digest does not match the request")
+    if strict_response:
+        from datetime import datetime
+        if (payload.get("approval_id") != approval_id
+                or payload.get("decision_id") != binding.decision_id
+                or payload.get("idempotent") is not False
+                or any(key in payload for key in ("effect","allow","deny","decision"))):
+            raise DecisionError("strict task consumption correspondence refused")
+        try:
+            stamp = datetime.fromisoformat(payload["consumed_at"].replace("Z","+00:00"))
+            if stamp.tzinfo is None or stamp.utcoffset() is None:
+                raise ValueError("timezone required")
+        except (KeyError,TypeError,AttributeError,ValueError):
+            raise DecisionError("strict task consumption timestamp refused") from None
     decision_id = payload.get("decision_id")
     if decision_id is not None and (
         not isinstance(decision_id, str) or not decision_id
diff --git a/src/secrets_engine/task_execution_handoff.py b/src/secrets_engine/task_execution_handoff.py
new file mode 100644
index 0000000..9772861
--- /dev/null
+++ b/src/secrets_engine/task_execution_handoff.py
@@ -0,0 +1,194 @@
+"""Narrow service-task PEP handoff; records never grant native execution.
+
+Platform calls preparation itself, binds its independently compiled owner Plan,
+reserves/journals once, then calls strict consumption and independent Approval
+GET confirmation. No inbound mapping/boolean or constructor creates authority.
+Actual policy/server input mapping and runtime entitlement remain unadmitted.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+import json
+from threading import Lock
+
+from secrets_engine.approval_auth import approval_token, credential_urlopen
+from secrets_engine.approval_claim import binding_from_check_request, claim_binding_digest, validate_approval_claim
+from secrets_engine.approval_consume import authorize_action, consume_approval
+from secrets_engine.application_time import read_window, validity_bounds
+from secrets_engine.authorization import validate_decision_envelope, request_digest, approval_binding_digest
+from secrets_engine.decision_signature import authenticate_response, load_trust, strict_json
+from secrets_engine.errors import DecisionError
+from secrets_engine.execution_contract import ExecutionContract, require_execution_deadline
+
+_CALLER = 'system:serviceaccount:secrets-engine:secrets-engine'
+_NATIVE_FIELDS = ('action','actor','principal','purpose','target')
+_COORDINATES = ('approval_url','approval_token_file','approval_client_secret_file','keycape_token_url','keycape_issuer',
+                'pdp_url','pdp_token_file','pdp_trust_file','pdp_verifier','clock_trust_file',
+                'authorization_subject_id','authorization_subject_type','authorization_policy_package','authorization_policy_version')
+_ISSUED = object()
+
+
+def _refuse():
+    raise DecisionError('strict task execution handoff correspondence refused')
+
+
+def _coordinates(cfg):
+    return tuple(str(getattr(cfg,name,'') or '') for name in _COORDINATES)
+
+
+def _stamp(value):
+    try:
+        parsed=datetime.fromisoformat(value.replace('Z','+00:00'))
+        if parsed.tzinfo is None or parsed.utcoffset() is None:
+            _refuse()
+        return parsed
+    except (AttributeError,TypeError,ValueError):
+        _refuse()
+
+
+class TaskExecutionAuthorization:
+    """Opaque process-local prepared state; not a bearer credential or allow."""
+    __slots__=('_action','_contract','_coordinates','_handoff','_used','_lock')
+
+    def __init__(self, issuer=None, *, action=None, contract=None, coordinates=None, handoff=None):
+        if issuer is not _ISSUED:
+            _refuse()
+        object.__setattr__(self,'_action',action)
+        object.__setattr__(self,'_contract',contract)
+        object.__setattr__(self,'_coordinates',coordinates)
+        object.__setattr__(self,'_handoff',json.dumps(handoff,sort_keys=True,separators=(',',':'),allow_nan=False))
+        object.__setattr__(self,'_used',False)
+        object.__setattr__(self,'_lock',Lock())
+
+    def __setattr__(self,name,value):
+        raise AttributeError('prepared task handoff is immutable')
+
+    def as_owner_handoff(self):
+        return strict_json(self._handoff)
+
+
+@dataclass(frozen=True)
+class TaskExecutionConsumption:
+    approval_id: str
+    request_digest: str
+    decision_id: str
+    consumed_at: str
+    idempotent: bool
+
+
+def _validate(cfg, action, contract):
+    if type(contract) is not ExecutionContract or not action.artifacts:
+        _refuse()
+    descriptor=contract.as_dict()
+    artifacts=strict_json(action.artifacts)
+    request,claim,envelope=(artifacts[name] for name in ('request','claim','envelope'))
+    # Rotation/removal and installed-helper changes cannot hide behind a cached
+    # prepared label. Reauthenticate before each strict CAS, without networking.
+    authenticated=authenticate_response(json.dumps(envelope,allow_nan=False).encode(),
+                                        trust=load_trust(getattr(cfg,'pdp_trust_file',None)),
+                                        verifier=getattr(cfg,'pdp_verifier',None))
+    if authenticated != envelope:
+        _refuse()
+    window=read_window(cfg)
+    lower,upper=validity_bounds(window=window)
+    validated=validate_decision_envelope(envelope,request,
+        accepted_policy_packages={str(cfg.authorization_policy_package)},
+        accepted_policy_versions={str(cfg.authorization_policy_version)},
+        expected_approval_binding_digest=claim['binding']['pdp_digest'],time_window=window)
+    validate_approval_claim(claim,approval_id=action.binding.approval_id,
+        expected_pdp_digest=envelope['binding']['approval_binding_digest'],
+        require_human_control=True,time_window=window)
+    require_execution_deadline(request,time_window=window,
+        bounds=(claim['validity']['expires_at'],envelope['lifetime']['expires_at']))
+    binding=envelope['binding']
+    if (envelope['lifetime'].get('kind')!='ttl'
+            or _stamp(envelope['lifetime'].get('not_before'))>lower):
+        _refuse()
+    if (request['action']!='execute-ccr' or binding.get('action')!='execute-ccr'
+            or request['tenant']!='tenant:platform' or binding.get('tenant')!='tenant:platform'
+            or binding.get('context')!=request['context']
+            or request['context'].get('execution_contract')!=descriptor
+            or binding.get('request_digest')!=request_digest(binding)
+            or binding.get('approval_binding_digest')!=approval_binding_digest(binding)
+            or action.binding.request_digest!=validated.request_digest
+            or action.binding.decision_id!=validated.decision_id):
+        _refuse()
+    subject=binding['subject']; resource=binding['resource']
+    if (request['subject']!={'id':'secrets-engine','type':'service'}
+            or subject.get('id')!='secrets-engine' or subject.get('type')!='service'
+            or subject.get('tenant') not in (None,'tenant:platform')
+            or 'principal' in subject.get('attributes',{})
+            or envelope['subject']!=subject or envelope['resource']!=resource
+            or descriptor['execution_actor']!='service:secrets-engine'
+            or any(resource.get(key)!=request['resource'].get(key) for key in ('id','type','system'))
+            or resource.get('tenant') not in (None,'tenant:platform')
+            or any(resource.get('attributes',{}).get(key)!=value for key,value in request['resource']['attributes'].items())
+            or any(key in binding['context'] for key in ('task_execution_projection','initiator','represented_principal','delegation','verified','owner_admission'))):
+        _refuse()
+    caller=envelope['provenance'].get('caller')
+    if (not isinstance(caller,dict) or caller.get('mode')!='enforce'
+            or caller.get('principal')!=_CALLER or caller.get('audience')!='flex-auth'
+            or _stamp(caller.get('not_after')) < _stamp(descriptor['operation_deadline'])
+            or upper >= _stamp(caller.get('not_after'))):
+        _refuse()
+    native=binding_from_check_request(binding)
+    supplied=claim['binding']
+    # Recompute the served native act too: Python equality alone conflates
+    # boolean/integer enrichment and cannot establish canonical correspondence.
+    if (any(supplied.get(key)!=native[key] for key in _NATIVE_FIELDS)
+            or supplied.get('digest')!=claim_binding_digest(**native)
+            or supplied.get('digest')!=claim_binding_digest(**{key:supplied.get(key) for key in _NATIVE_FIELDS})):
+        _refuse()
+    return {'execution_contract':descriptor,'approval_id':action.binding.approval_id,
+        'request_digest':validated.request_digest,'decision_id':validated.decision_id,
+        'native_binding':native,'approval_binding_digest':binding['approval_binding_digest'],
+        'submitted_request_digest':validated.submitted_request_digest,
+        'decision_not_before':envelope['lifetime'].get('not_before') or '',
+        'decision_expires_at':envelope['lifetime']['expires_at'],
+        'approval_expires_at':claim['validity']['expires_at']}
+
+
+def authorize_task_execution(cfg, entry, *, execution_contract):
+    if type(execution_contract) is not ExecutionContract:
+        _refuse()
+    # No injected transports or policy callbacks: use the actual existing PIP/PDP
+    # clients and mandatory signature validators in the owner process.
+    action=authorize_action(cfg,entry,'execute-ccr',execution_contract=execution_contract)
+    if action is None:
+        _refuse()
+    try:
+        handoff=_validate(cfg,action,execution_contract)
+        return TaskExecutionAuthorization(_ISSUED,action=action,contract=ExecutionContract.parse(execution_contract.as_dict()),
+                                          coordinates=_coordinates(cfg),handoff=handoff)
+    except (ValueError,KeyError,TypeError,RecursionError) as exc:
+        raise DecisionError('strict task execution handoff malformed') from exc
+
+
+def consume_task_execution(cfg, authorization):
+    if type(authorization) is not TaskExecutionAuthorization:
+        _refuse()
+    with authorization._lock:
+        if authorization._used or _coordinates(cfg)!=authorization._coordinates:
+            _refuse()
+        # Reserve once before all work; failures/unknown outcomes never retry this
+        # prepared object. Platform's durable journal is the cross-process guard.
+        object.__setattr__(authorization,'_used',True)
+    try:
+        handoff=_validate(cfg,authorization._action,authorization._contract)
+        if handoff!=authorization.as_owner_handoff():
+            _refuse()
+        result=consume_approval(base_url=str(cfg.approval_url),
+            token_provider=lambda:approval_token(cfg,scope='approval:consume'),
+            binding=authorization._action.binding,opener=credential_urlopen,strict_response=True)
+        stamp=_stamp(result.consumed_at)
+        window=read_window(cfg);_,upper=validity_bounds(window=window)
+        if (stamp>upper or (handoff['decision_not_before'] and stamp<_stamp(handoff['decision_not_before']))
+                or any(stamp>=_stamp(value) for value in (handoff['decision_expires_at'],handoff['approval_expires_at'],
+                                                         handoff['execution_contract']['operation_deadline']))):
+            _refuse()
+        # Late CAS completion is historical evidence only; refuse native dispatch.
+        _validate(cfg,authorization._action,authorization._contract)
+        return TaskExecutionConsumption(result.approval_id,result.request_digest,result.decision_id,result.consumed_at,result.idempotent)
+    except (ValueError,KeyError,TypeError,RecursionError) as exc:
+        raise DecisionError('strict task execution consumption malformed') from exc
diff --git a/tests/test_task_execution_handoff.py b/tests/test_task_execution_handoff.py
new file mode 100644
index 0000000..2204da2
--- /dev/null
+++ b/tests/test_task_execution_handoff.py
@@ -0,0 +1,231 @@
+"""Real local HTTP/producer signatures, synthetic approval/caller only.
+
+These source fixtures never enable Access's runtime-deny policy or admit a real
+caller, key, owner grant or native backend. Whole-provider integration is separate.
+"""
+import copy
+from datetime import datetime, timedelta, timezone
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+from pathlib import Path
+from threading import Thread
+
+import pytest
+
+from secrets_engine.approval_claim import binding_from_check_request, claim_binding_digest
+from secrets_engine.approval_consume import _expected_request
+from secrets_engine.authorization import request_digest, approval_binding_digest
+from secrets_engine.errors import DecisionError
+from secrets_engine.execution_contract import ExecutionContract
+from secrets_engine.task_execution_handoff import (authorize_task_execution,consume_task_execution,
+    TaskExecutionAuthorization)
+from tests.authorization_stub import AuthorizationStub
+from tests.signature_support import signature_tools, admit_tools, sign_envelope
+from tests.test_consume_binding_join import _Cfg,_token,APPROVAL_ID
+from tests.test_execution_context import entry,descriptor
+
+
+def stamp(moment):
+    return moment.isoformat()
+
+
+@pytest.fixture
+def lane(tmp_path,signature_tools):
+    e=entry();contract=ExecutionContract.parse(descriptor(e))
+    cfg=_Cfg(_token(tmp_path),authorization_subject_id='secrets-engine',authorization_subject_type='service')
+    admit_tools(cfg,tmp_path,signature_tools)
+    original=_expected_request(cfg,e,'execute-ccr',execution_contract=contract)
+    stub=AuthorizationStub(approval_id=APPROVAL_ID,package=cfg.authorization_policy_package,version=cfg.authorization_policy_version)
+    stub.bind_request(original)
+    state={'calls':[],'claim_change':None,'decision_change':None,'response_change':None,'raw_response':None}
+    claim=stub.claim();native=binding_from_check_request(original)
+    claim['binding'].update(native,digest=claim_binding_digest(**native),human_control=True)
+    state['claim']=claim
+
+    class Handler(BaseHTTPRequestHandler):
+        def log_message(self,*_args):pass
+        def reply(self,payload):
+            data=json.dumps(payload).encode() if isinstance(payload,dict) else payload
+            self.send_response(200);self.send_header('Content-Type','application/json');self.send_header('Content-Length',str(len(data)));self.end_headers();self.wfile.write(data)
+        def do_GET(self):
+            state['calls'].append('claim')
+            value=copy.deepcopy(state['claim'])
+            if state['claim_change']:state['claim_change'](value)
+            self.reply(value)
+        def do_POST(self):
+            submitted=json.loads(self.rfile.read(int(self.headers['Content-Length'])))
+            if self.path=='/v1/check':
+                state['calls'].append('check')
+                envelope=stub.decision(submitted)
+                envelope['provenance']['caller']={'mode':'enforce','principal':'system:serviceaccount:secrets-engine:secrets-engine',
+                    'audience':'flex-auth','not_after':stamp(datetime.now(timezone.utc)+timedelta(minutes=20))}
+                if state['decision_change']:state['decision_change'](envelope)
+                self.reply(sign_envelope(envelope,signature_tools))
+            else:
+                state['calls'].append('consume')
+                payload={'approval_id':APPROVAL_ID,'status':'consumed','request_digest':submitted['request_digest'],
+                         'decision_id':submitted['decision_id'],'consumed_at':stamp(datetime.now(timezone.utc)),'idempotent':False}
+                if state['response_change']:state['response_change'](payload)
+                self.reply(state['raw_response'] if state['raw_response'] is not None else payload)
+    httpd=ThreadingHTTPServer(('127.0.0.1',0),Handler)
+    thread=Thread(target=httpd.serve_forever,kwargs={'poll_interval':0.01},daemon=True);thread.start()
+    cfg.approval_url=cfg.pdp_url=f'http://127.0.0.1:{httpd.server_port}'
+    try:yield cfg,e,contract,state
+    finally:httpd.shutdown();httpd.server_close();thread.join(2)
+
+
+def prepared(lane):
+    cfg,e,contract,state=lane
+    return authorize_task_execution(cfg,e,execution_contract=contract)
+
+
+def test_exact_preparation_then_one_strict_cas(lane):
+    cfg,e,contract,state=lane
+    result=prepared(lane)
+    assert state['calls']==['claim','check']
+    handoff=result.as_owner_handoff()
+    assert set(handoff)=={'execution_contract','approval_id','request_digest','decision_id','native_binding','approval_binding_digest',
+                         'submitted_request_digest','decision_not_before','decision_expires_at','approval_expires_at'}
+    assert handoff['execution_contract']==contract.as_dict()
+    assert handoff['native_binding']==binding_from_check_request(_expected_request(cfg,e,'execute-ccr',execution_contract=contract))
+    changed=result.as_owner_handoff();changed['execution_contract']['ccr_id']='changed'
+    assert result.as_owner_handoff()==handoff
+    with pytest.raises(AttributeError):result._handoff='changed'
+    receipt=consume_task_execution(cfg,result)
+    assert receipt.approval_id==handoff['approval_id'] and receipt.request_digest==handoff['request_digest']
+    assert receipt.decision_id==handoff['decision_id'] and receipt.idempotent is False
+    assert state['calls']==['claim','check','consume']
+    with pytest.raises(DecisionError):consume_task_execution(cfg,result)
+    assert state['calls'].count('consume')==1
+
+
+@pytest.mark.parametrize('field',['action','actor','principal','purpose','target','digest'])
+def test_changed_native_claim_refuses_before_cas(lane,field):
+    cfg,e,contract,state=lane
+    def change(claim):claim['binding'][field]={'id':'other'} if field=='target' else 'other'
+    state['claim_change']=change
+    with pytest.raises(DecisionError):prepared(lane)
+    assert 'consume' not in state['calls']
+
+
+@pytest.mark.parametrize('field',['mode','principal','audience','not_after'])
+def test_signed_wrong_caller_refuses_before_cas(lane,field):
+    cfg,e,contract,state=lane
+    state['decision_change']=lambda envelope:envelope['provenance']['caller'].__setitem__(field,'wrong')
+    with pytest.raises(DecisionError):prepared(lane)
+    assert 'consume' not in state['calls']
+
+
+@pytest.mark.parametrize('field',['execution_actor','execution_grant_ccr_id','grant_immutable_sha256','ccr_id','backend_origin'])
+def test_signed_changed_descriptor_refuses_before_cas(lane,field):
+    cfg,e,contract,state=lane
+    def change(envelope):envelope['binding']['context']['execution_contract'][field]='wrong'
+    state['decision_change']=change
+    with pytest.raises(DecisionError):prepared(lane)
+    assert 'consume' not in state['calls']
+
+
+@pytest.mark.parametrize('field',['subject','resource','request_digest','submitted_request_digest','approval_binding_digest'])
+def test_signed_changed_request_correspondence_refuses_before_cas(lane,field):
+    cfg,e,contract,state=lane
+    def change(envelope):
+        envelope['binding'][field]={'id':'other'} if field in ('subject','resource') else 'sha256:'+'0'*64
+    state['decision_change']=change
+    with pytest.raises(DecisionError):prepared(lane)
+    assert 'consume' not in state['calls']
+
+
+@pytest.mark.parametrize('field',['approval_id','decision_id','request_digest','status','consumed_at','idempotent'])
+def test_cas_correspondence_never_coerces_or_falls_back(lane,field):
+    cfg,e,contract,state=lane;result=prepared(lane)
+    state['response_change']=lambda payload:payload.__setitem__(field,'wrong')
+    with pytest.raises(DecisionError):consume_task_execution(cfg,result)
+    with pytest.raises(DecisionError):consume_task_execution(cfg,result)
+    assert state['calls'].count('consume')==1
+
+
+@pytest.mark.parametrize('value',[True,1,0,'false',None])
+def test_nonfresh_or_nonboolean_idempotence_refuses(lane,value):
+    cfg,e,contract,state=lane;result=prepared(lane)
+    state['response_change']=lambda payload:payload.__setitem__('idempotent',value)
+    with pytest.raises(DecisionError):consume_task_execution(cfg,result)
+
+
+@pytest.mark.parametrize('body',[b'{"status":"consumed","status":"consumed"}',b'{"status":NaN}',b'['*2000+b'0'+b']'*2000])
+def test_duplicate_nonfinite_or_deep_consume_body_refuses(lane,body):
+    cfg,e,contract,state=lane;result=prepared(lane)
+    state['raw_response']=body
+    with pytest.raises(DecisionError):consume_task_execution(cfg,result)
+
+
+@pytest.mark.parametrize('field',['pdp_url','approval_url','authorization_subject_id','authorization_policy_version','pdp_verifier'])
+def test_changed_owner_coordinates_refuse_before_cas(lane,field):
+    cfg,e,contract,state=lane;result=prepared(lane);setattr(cfg,field,'changed')
+    with pytest.raises(DecisionError):consume_task_execution(cfg,result)
+    assert 'consume' not in state['calls']
+
+
+def test_removed_signer_refuses_before_cas(lane):
+    cfg,e,contract,state=lane;result=prepared(lane)
+    trust=json.loads(cfg.pdp_trust_file.read_text());trust['keys']={'other':next(iter(trust['keys'].values()))}
+    cfg.pdp_trust_file.write_text(json.dumps(trust))
+    with pytest.raises(DecisionError):consume_task_execution(cfg,result)
+    assert 'consume' not in state['calls']
+
+
+def test_mapping_boolean_and_public_constructor_are_not_authority(lane):
+    cfg,e,contract,state=lane
+    for value in [True,{},object()]:
+        with pytest.raises(DecisionError):consume_task_execution(cfg,value)
+    with pytest.raises(DecisionError):TaskExecutionAuthorization()
+    assert not state['calls']
+
+
+@pytest.mark.parametrize('field',['approval_id','decision_id','consumed_at','idempotent'])
+def test_absent_cas_fields_cannot_fall_back(lane,field):
+    cfg,e,contract,state=lane;result=prepared(lane)
+    state['response_change']=lambda payload:payload.pop(field)
+    with pytest.raises(DecisionError):consume_task_execution(cfg,result)
+    assert state['calls'].count('consume')==1
+
+
+def test_expired_preparation_refuses_before_cas(lane,monkeypatch):
+    from types import SimpleNamespace
+    import secrets_engine.task_execution_handoff as module
+    cfg,e,contract,state=lane;result=prepared(lane)
+    later=int((datetime.now(timezone.utc)+timedelta(minutes=2)).timestamp()*1_000_000_000)
+    monkeypatch.setattr(module,'read_window',lambda cfg:SimpleNamespace(lower_ns=later,upper_ns=later))
+    with pytest.raises(DecisionError):consume_task_execution(cfg,result)
+    assert 'consume' not in state['calls']
+
+
+@pytest.mark.parametrize('change',['deny','expired','missing-caller','missing-start'])
+def test_signed_decision_missing_authority_refuses_before_cas(lane,change):
+    cfg,e,contract,state=lane
+    def alter(envelope):
+        if change=='deny':envelope['effect']='deny'
+        elif change=='expired':envelope['lifetime']['expires_at']=stamp(datetime.now(timezone.utc)-timedelta(seconds=1))
+        elif change=='missing-start':envelope['lifetime'].pop('not_before')
+        else:envelope['provenance'].pop('caller')
+    state['decision_change']=alter
+    with pytest.raises(DecisionError):prepared(lane)
+    assert 'consume' not in state['calls']
+
+
+def test_native_binding_equality_does_not_coerce_boolean_enrichment(lane):
+    cfg,e,contract,state=lane
+    evaluated=_expected_request(cfg,e,'execute-ccr',execution_contract=contract)
+    evaluated['resource']['attributes']['synthetic_rank']=1
+    native=binding_from_check_request(evaluated)
+    state['claim']['binding'].update(copy.deepcopy(native),digest=claim_binding_digest(**native),
+                                     pdp_digest=approval_binding_digest(evaluated))
+    # Python dict equality treats True == 1. This malformed provider fact must
+    # refuse even when its reported native digest names the integer counterpart.
+    state['claim']['binding']['target']['attributes']['synthetic_rank']=True
+    def enrich(envelope):
+        envelope['resource']['attributes']['synthetic_rank']=1
+        envelope['binding']['request_digest']=request_digest(envelope['binding'])
+        envelope['binding']['approval_binding_digest']=approval_binding_digest(envelope['binding'])
+    state['decision_change']=enrich
+    with pytest.raises(DecisionError):prepared(lane)
+    assert 'consume' not in state['calls']
diff --git a/workplans/SECRETS-WP-0013-task-scoped-openbao-execution.md b/workplans/SECRETS-WP-0013-task-scoped-openbao-execution.md
index 01be511..a8a3845 100644
--- a/workplans/SECRETS-WP-0013-task-scoped-openbao-execution.md
+++ b/workplans/SECRETS-WP-0013-task-scoped-openbao-execution.md
@@ -9,7 +9,7 @@ flavor: planning
 owner: codex
 topic_slug: netkingdom
 created: "2026-10-11"
-updated: "2026-10-11"
+updated: "2026-10-16"
 depends_on: []
 state_hub_workstream_id: "5e38aecc-85ca-5f55-94c6-9fbb08604819"
 ---
@@ -110,3 +110,17 @@ This is source preparation awaiting independent technical review/publication,
 not a formal owner assent, policy deployment or native execution milestone. T01
 remains progress; T03 and SECRETS-WP-0008-T06 remain open. Detailed limitations
 and retained failed test observations are in the source preparation receipt.
+
+
+2026-10-16 bounded source handoff preparation: `task_execution_handoff.py` uses
+the actual serving approval/mandatory signed Check path and retained detached
+artifacts to check exact descriptor/request/native approval/caller/lifetime joins,
+then performs one strict exact-response CAS only after Platform journal admission.
+It accepts no inbound boolean/mapping/authorizer callback and grants no native
+execution. Platform independently confirms durable consumption with the actual
+Approval provider client before native dispatch. Focused local HTTP/native Go
+producer-signature source checks cover altered joins/caller/grant, missing fields,
+coercion/idempotence, expiry, signer removal and replay refusal. No service binding
+propagation, policy gate flip, identity mapping admission or production trust/key
+change was made. Independent source integration/review and actual whole-provider
+proof remain required. T01/T03 and WP8-T06 retain their open native obligations.