Skip quota source transport for status-only context hooks

loopx/control_plane/capability_hooks.py

@@ -237,12 +237,18 @@ def dispatch_turn_context_hooks(
             if any(key in payload for key in registration.readback_slots):
                 raise ValueError("context readback slot already contributed")
             result = registration.producer()
+            source = turn_context_source(payload)
+            readback_only = (
+                result.get("context_fields") == {}
+                and result.get("instructions") == []
+                and result.get("result_fields") == {}
+            )
             contribution = effect_runtime_result(
                 "capability_hook.turn_context.validate",
                 {
                     "registration": registration.contract(),
                     "result": result,
-                    "source": turn_context_source(payload),
+                    **({"readback_only": True} if readback_only else {"source": source}),
                 },
             )
             readback = contribution.pop("readback")

loopx/control_plane/turn_driver/capability_context.ts

@@ -1,6 +1,6 @@
 import type {JsonObject} from "../effect_program.ts";
 import {canonicalAuthoritySha256} from "../coordination/authority_store_codec.ts";
-import {requireJsonObject, requireNonEmptyString, requireStringArray} from "../runtime_decode.ts";
+import {requireBoolean, requireJsonObject, requireNonEmptyString, requireStringArray} from "../runtime_decode.ts";
 import {quotaActionSignatureDocument, turnEnvelopeActionSignatureDocument} from "../quota/turn_envelope.ts";
 
 // This is a local host-IO contract, not a new capability or authority vocabulary.
@@ -69,6 +69,15 @@ export function validateTurnContextContribution(input: JsonObject): JsonObject {
   }
   const normalized = {hook_id: registration.hook_id, capability_id: registration.capability_id,
     readback_slots: slots, context_fields: context, instructions, result_fields: resultFields, readback};
+  if (input.readback_only !== undefined && requireBoolean(input.readback_only, "readback_only")) {
+    // A status-only result is returned to the guard and is never a reusable
+    // context contribution. Keep its registration, slot and budget checks.
+    if (Object.hasOwn(input, "source") || Object.keys(context).length ||
+        instructions.length || Object.keys(resultFields).length) {
+      throw new Error("readback-only context validation cannot carry a source or a contribution");
+    }
+    return normalized;
+  }
   return {...normalized, basis_digest: canonicalAuthoritySha256({source: contextSource(input.source), ...normalized})};
 }
 

tests/control_plane/test_turn_capability_lifecycle.py

@@ -22,8 +22,21 @@
 )
 
 
-def test_invalid_or_duplicate_context_registration_never_reinvokes_provider() -> None:
+def test_invalid_or_duplicate_context_registration_never_reinvokes_provider(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from loopx.control_plane import capability_hooks
+
     calls = []
+    validation_requests = []
+    invoke = capability_hooks.effect_runtime_result
+
+    def observe(method, params, **kwargs):
+        if method == "capability_hook.turn_context.validate":
+            validation_requests.append(params)
+        return invoke(method, params, **kwargs)
+
+    monkeypatch.setattr(capability_hooks, "effect_runtime_result", observe)
 
     def produce():
         calls.append("read")
@@ -44,6 +57,10 @@ def produce():
     dispatch_turn_context_hooks(payload, (invalid, valid, valid))
     assert calls == ["read"]
     assert len(payload["turn_context_hook_failures"]) == 2
+    assert payload["fixture_readback"] == {}
+    assert "turn_context_contributions" not in payload
+    request, = validation_requests
+    assert request["readback_only"] is True and "source" not in request
 
     # Retrying composition on an already populated packet must not repeat IO.
     dispatch_turn_context_hooks(payload, (valid,))

tests/control_plane_ts/turn_capability_context.test.ts

@@ -16,6 +16,22 @@ test("registered context survives location-only capture binding", () => {
     source: {...source, detail_ref: {full_decision: "fixture location"}, measurements: {bytes: 500}}}), contribution);
 });
 
+test("readback-only validation never produces or accepts a reusable contribution", () => {
+  const status = {...result, context_fields: {}, instructions: [], result_fields: {}};
+  const request = {registration, result: status, readback_only: true};
+  const readback = validateTurnContextContribution(request);
+  assert.deepEqual(readback.readback, status.readback);
+  assert.equal(Object.hasOwn(readback, "basis_digest"), false);
+  assert.throws(() => projectTurnContextContribution({contribution: readback, readback: status.readback, source}));
+  assert.throws(() => validateTurnContextContribution({...request, source}));
+  assert.throws(() => validateTurnContextContribution({registration, result: status}));
+  for (const patch of [{context_fields: result.context_fields}, {instructions: result.instructions},
+    {result_fields: result.result_fields}]) {
+    assert.throws(() => validateTurnContextContribution({...request, result: {...status, ...patch}}));
+  }
+  assert.throws(() => validateTurnContextContribution({...request, result: {...status, readback: {other_slot: {}}}}));
+});
+
 const quotaSource = {goal_id: "fixture-goal", agent_id: "fixture-agent",
   ok: true, should_run: true, mode: "should-run", effective_action: "normal_run",
   interaction_contract: {schema_version: "loopx_interaction_contract_v0", mode: "bounded_delivery",