Reuse validated canonical Todo source bytes and scalar comparisons

loopx/control_plane/todos/read_transaction.ts

@@ -78,13 +78,13 @@ function validateCanonicalSource(canonical: JsonObject, expected: JsonObject): v
 
 /** Full-tree transport, with provider facts immutable. Independent additions and
  * the acceptance owner's observation do not acquire provider provenance. */
-function resolveCanonicalSource(fields: JsonObject, reference: JsonObject, expectedField?: string): JsonObject {
+function resolveCanonicalSource(fields: JsonObject, reference: JsonObject, expectedField?: string, fieldDigests?: JsonObject): JsonObject {
   if ((reference.field !== "agent_todos" && reference.field !== "user_todos") ||
       expectedField !== undefined && reference.field !== expectedField) {
     throw new EffectRuntimeRequestError("quota source field must name the consumed Todo role");
   }
   const field = requireJsonObject(fields[reference.field], "quota source field");
-  if (canonicalAuthoritySha256(field) !== reference.field_sha256) {
+  if ((fieldDigests?.[reference.field] ?? canonicalAuthoritySha256(field)) !== reference.field_sha256) {
     throw new EffectRuntimeRequestError("quota source projection no longer matches the original complete read");
   }
   const record = JSON.parse(JSON.stringify(field)) as JsonObject;
@@ -114,8 +114,13 @@ function resolveCanonicalSource(fields: JsonObject, reference: JsonObject, expec
       delete parent[last];
     } else Object.defineProperty(parent, last, {value: change.value, enumerable: true, writable: true, configurable: true});
   }
+  // JSON cloning owns containers while retaining primitive values. Exact
+  // equality already proves an unchanged primitive; containers and altered
+  // scalar values retain their full canonical digest comparison.
   for (const [key, value] of Object.entries(field)) {
-    if (key !== "goal_acceptance_contract" && (!Object.hasOwn(record, key) || canonicalAuthoritySha256(value) !== canonicalAuthoritySha256(record[key]))) {
+    const unchangedScalar = value === record[key] && (value === null || typeof value !== "object");
+    if (key !== "goal_acceptance_contract" && (!Object.hasOwn(record, key) ||
+        !unchangedScalar && canonicalAuthoritySha256(value) !== canonicalAuthoritySha256(record[key]))) {
       throw new EffectRuntimeRequestError("canonical Todo facts cannot be replaced by caller source differences");
     }
   }
@@ -157,7 +162,11 @@ export async function readCanonicalQuotaProjection(input: JsonObject, dependenci
         validateCanonicalSource(canonical, expected);
         reads.set(key, projected);
       }
-      const source = {record: resolveCanonicalSource(requireJsonObject(projected.fields, "quota source fields"), reference),
+      // These digests were produced with these fields by the fresh read above;
+      // no caller-supplied digest or field survives between requests.
+      const fieldDigests = requireJsonObject(requireJsonObject(projected.source_projection,
+        "quota source projection").field_digests, "quota source field digests");
+      const source = {record: resolveCanonicalSource(requireJsonObject(projected.fields, "quota source fields"), reference, undefined, fieldDigests),
         python_values: reference.python_values ?? []};
       request = {...request, summary_request: {...summary, source}};
     }

loopx/control_plane/todos/summary_item.py

@@ -53,15 +53,16 @@ def bind_canonical_todo_sources(result: dict[str, Any], read: dict[str, Any]) ->
         digest = metadata.get("field_digests", {}).get(field)
         if not isinstance(value, dict) or not isinstance(digest, str):
             continue
-        basis = json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8", "backslashreplace")
         observed = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8", "backslashreplace")
         if sha256(observed).hexdigest() != digest:
             # A modified response or a numeric JSON spelling mismatch cannot
             # mint a native reference. Keep the complete supplied source codec.
             continue
+        # The basis is decoded as JSON; mapping insertion order is not a
+        # source difference. Retain the complete bytes already validated here.
         fields[field] = _CanonicalTodoSummary(value, read={key: value for key, value in metadata.get("replay_read", read).items() if key != "return_projection"},
             expected=metadata["source"], field=field, digest=digest,
-            basis=basis, request_identity=identity,
+            basis=observed, request_identity=identity,
             evaluations=(result.get("field_evaluations") or {}).get(field),
             source_row_fields=tuple(metadata.get("source_row_fields", [])))
 

tests/capabilities/test_periodic_report_pending_intent.py

@@ -65,6 +65,8 @@ def _enable_calendar(registry):
 
 
 def test_calendar_actual_cli_preserves_capability_action_without_host(tmp_path, capsys, monkeypatch):
+    from copy import deepcopy
+
     from loopx.cli import main
     from loopx.control_plane.turn_driver.loop_controller import decide_loop_disposition
 
@@ -114,13 +116,22 @@ def no_host(*args, **kwargs):
         assert turn["capability_action"]["intent"]["command"] == command
         assert turn["capability_action"]["command_argv"][:5] == ["loopx", "--registry", str(registry), "--runtime-root", str(runtime)]
         assert turn["capability_action"]["executed"] is False
+        # Current plans expose one complete source; the controller consumes it
+        # directly instead of requiring the historical executor adapter.
+        assert "turn_envelope" not in turn
         work = turn["work_contract"]
         assert work["effective_action"] == "governed_capability_intent"
         assert work["interaction_contract"]["cli_channel"]["next_cli_actions"] == [command]
         assert not any(turn["effects"].values())
+        # The source may retain competing replan advice; capability precedence
+        # must still keep the host idle and the owned consumer next.
         disposition = decide_loop_disposition(turn_receipt=None, quota_decision=work)
         assert disposition["disposition"] == "capability_action_required"
         assert not disposition["launches_host"]
+        changed = deepcopy(work)
+        changed["pending_capability_intent"]["command"] = "loopx turn run-once --execute"
+        with pytest.raises(ValueError):
+            decide_loop_disposition(turn_receipt=None, quota_decision=changed)
     assert not list((runtime / "goals" / GOAL_ID / "periodic_reports").glob("*/editorial_request.json"))
 
 

tests/control_plane/test_canonical_quota_projection.py

@@ -25,6 +25,15 @@ def test_provider_projection_uses_complete_source_differences_and_scope_exit(tmp
         reference = canonical_todo_source_reference(source)
         assert reference is not None, "the real native read must supply a validated reference"
         assert reference["changes"] == []
+        # Mapping insertion order is not a caller change. Reordering a complete
+        # native summary must retain its source without hiding scalar changes.
+        reordered = deepcopy(source)
+        entries = list(reordered.items())
+        reordered.clear()
+        reordered.update(reversed(entries))
+        assert canonical_todo_source_reference(reordered) == reference
+        reordered["items"][0]["done"] = 0
+        assert canonical_todo_source_reference(reordered) is None
         copied = deepcopy(source)
         proof = {"agent_id": fixture.AGENT_ID, "state": "waiting", "evidence": {"revision": 9007199254740993, "ratio": -0.0}}
         copied["vision_wait_states"] = [proof]
@@ -150,6 +159,15 @@ def observed(method, params, **kwargs):
         assert len(requests) == 1 and len(requests[0]["requests"]) == 2
         assert all("source_reference" in row["summary_request"] for row in requests[0]["requests"])
         assert all(not row["summary_request"]["consume_display_views"] for row in requests[0]["requests"])
+        # Each role must match its own freshly read full field, even when both
+        # requests share one provider read. A rejected pair cannot poison reuse.
+        invalid = deepcopy(requests[0])
+        references = [row["summary_request"]["source_reference"] for row in invalid["requests"]]
+        assert references[0]["field_sha256"] != references[1]["field_sha256"]
+        references[1]["field_sha256"] = references[0]["field_sha256"]
+        with pytest.raises(ValueError, match="original complete read"):
+            project_todo_read(invalid)
+        assert select_quota_todo_summaries(user, agent, agent_identity=identity) == expected
         stale = deepcopy(agent)
         stale.expected = {**stale.expected, "cursor": "foreign"}
         with pytest.raises(ValueError, match="canonical source changed"):

tests/control_plane_ts/postgresql_authority_store.integration.test.ts

@@ -175,6 +175,29 @@ if (database && installed) {
     const forged = structuredClone(requests);
     ((forged[1]!.summary_request as JsonObject).source_reference as JsonObject).changes = [{path: ["items", 0, "note"], value: "Forged source fact"}];
     await assert.rejects(readCanonicalQuotaProjection({requests: forged}, dependencies), /canonical Todo facts/);
+    const agentField = fields.agent_todos as JsonObject;
+    const sourceItem = (agentField.items as JsonObject[])[0]!;
+    for (const changes of [
+      [{path: ["items", 0, "required_write_scopes", 0], value: "different.json"}],
+      [{path: ["open_count"], value: Number(agentField.open_count) + 1}],
+      [{path: ["items"], delete: true}],
+    ]) {
+      const changed = structuredClone(requests);
+      ((changed[1]!.summary_request as JsonObject).source_reference as JsonObject).changes = changes;
+      await assert.rejects(readCanonicalQuotaProjection({requests: changed}, dependencies), /canonical Todo facts/);
+    }
+    // Source protection compares the final complete value. Temporary changes
+    // that are restored, including a parent replacement and deletion, retain
+    // the same read result and never mutate provider data.
+    const restored = structuredClone(requests);
+    const originalChanges = ((requests[1]!.summary_request as JsonObject).source_reference as JsonObject).changes as JsonObject[];
+    ((restored[1]!.summary_request as JsonObject).source_reference as JsonObject).changes = [
+      {path: ["items"], value: structuredClone(agentField.items)},
+      {path: ["items", 0, "note"], delete: true},
+      {path: ["items", 0, "note"], value: sourceItem.note},
+      ...structuredClone(originalChanges),
+    ];
+    assert.deepEqual(await readCanonicalQuotaProjection({requests: restored}, dependencies), result);
     const untouched = await store.loadAuthority();
     assert.equal(untouched.status, "loaded");
     if (untouched.status !== "loaded") throw new Error("synthetic provider disappeared");