diff --git a/loopx/control_plane/quota/context_delivery.ts b/loopx/control_plane/quota/context_delivery.ts index bed18be2e9..dc6b1489e7 100644 --- a/loopx/control_plane/quota/context_delivery.ts +++ b/loopx/control_plane/quota/context_delivery.ts @@ -1,3 +1,5 @@ +import {contextSourceIdentity, contextProjectionChanges, contextProjectionIdentity, + type ProjectionSourceValues} from "./diagnostic_projection.ts"; import {completeGoalNarrative} from "../work_items/goal_context_source.ts"; /** Presentation of immutable quota, Turn and work-context captures. No admission or state writes. * Confirmed final input and explicit native-hook continuity assumptions remain @@ -330,7 +332,8 @@ function freshPaths(context: JsonObject, surface: Surface, legacyQuotaFields: re /** Group complete CURRENT literals, not baseline references. Keep the cheaper * ordinary edit when grouping would grow metadata; both containers share this rule. */ -function groupCurrentValues(delta: Exclude, size: (value: unknown) => number): Delta { +function groupCurrentValues(delta: Exclude, size: (value: unknown) => number, + sourceValues?: ProjectionSourceValues): Delta { const children = Object.entries("items" in delta ? delta.items : delta.changed ?? {}); const values = children.filter(([, child]) => "value" in child); if (values.length < 2) return delta; @@ -340,10 +343,35 @@ function groupCurrentValues(delta: Exclude [key, (child as {value: unknown}).value])), }; if (!("items" in grouped) && !Object.keys(nested).length) delete grouped.changed; + if (sourceValues) { + const origins = new Map(); + for (const [key, child] of values) { + const path = sourceValues.get(child)?.get("value"); + if (path) origins.set(key, path); + } + if (origins.size) sourceValues.set(grouped.values!, origins); + } return size(grouped) < size(delta) ? grouped : delta; } -function changeTree(before: unknown, after: unknown, forced: Path[], preserveFileNumbers = false): Delta | null { +type SourceComparison = {identity: ReturnType; + before: Path; after: Path; values?: ProjectionSourceValues}; + +function changeTree(before: unknown, after: unknown, forced: Path[], preserveFileNumbers = false, + source?: SourceComparison): Delta | null { + const same = (a: unknown, b: unknown, aPath: Path, bPath: Path): boolean => + a === undefined || b === undefined ? a === b : source + ? source.identity(a, [...source.before, ...aPath]) === source.identity(b, [...source.after, ...bPath]) + : equal(a, b); + const copy = (value: T): T => source?.values ? value : clone(value); + const literal = (value: unknown, path: Path): Delta => { + const result = {value: copy(value)}; + if (source?.values && typeof value === "number") + source.values.set(result, new Map([["value", [...source.after, ...path]]])); + return result; + }; + const identity = (value: unknown, path: Path, side: "before" | "after"): string => source + ? digest(source.identity(value, [...source[side], ...path])) : digest(value); // Contexts have passed strict JSON digest validation; patch containers remain // immutable after sizing. Reuse child sizes within this one tree calculation. const sizes = new WeakMap(); @@ -411,8 +439,8 @@ function changeTree(before: unknown, after: unknown, forced: Path[], preserveFil index(before, []); const visit = (a: unknown, b: unknown, path: Path, relocate = true): Delta | null => { const key = pathKey(path); - if (exact.has(key)) return {value: clone(b)}; - if (!ancestors.has(key) && equal(a, b)) return null; + if (exact.has(key)) return literal(b, path); + if (!ancestors.has(key) && same(a, b, path, path)) return null; if (typeof b === "string" && strings.has(b)) return clone(strings.get(b)!); let reference: {from: Path} | undefined; let overlay: Delta | undefined; @@ -426,13 +454,13 @@ function changeTree(before: unknown, after: unknown, forced: Path[], preserveFil if (bucket && !bucket.indexed) { bucket.indexed = new Map(); for (const candidate of bucket.candidates) { - const hash = digest(candidate.value), previous = bucket.indexed.get(hash); + const hash = identity(candidate.value, candidate.reference.from, "before"), previous = bucket.indexed.get(hash); if (!previous || bytes(candidate.reference) < bytes(previous.reference)) bucket.indexed.set(hash, candidate); } } - const reusable = bucket?.indexed?.get(digest(b)); + const reusable = bucket?.indexed?.get(identity(b, path, "after")); // Size and hash lookup are indices: exact equality remains the oracle. - if (reusable && equal(reusable.value, b)) { + if (reusable && same(reusable.value, b, reusable.reference.from, path)) { if (!ancestors.has(key)) reference = reusable.reference; else if (relocate) { // Compare against this exact body at its new location. Disabling @@ -452,7 +480,7 @@ function changeTree(before: unknown, after: unknown, forced: Path[], preserveFil if (child) changed.push([field, child]); } const removed = Object.keys(old).filter(field => !Object.hasOwn(next, field)); - delta = groupCurrentValues({changed: Object.fromEntries(changed), ...(removed.length ? {removed} : {})}, size); + delta = groupCurrentValues({changed: Object.fromEntries(changed), ...(removed.length ? {removed} : {})}, size, source?.values); } else if ((Array.isArray(a) || a === undefined) && Array.isArray(b)) { const old = a as unknown[] | undefined; const items: [string, Delta][] = []; @@ -460,11 +488,11 @@ function changeTree(before: unknown, after: unknown, forced: Path[], preserveFil const child = visit(old?.[i], value, [...path, i], relocate && !overlay); if (child) items.push([String(i), child]); }); - delta = groupCurrentValues({items: Object.fromEntries(items), ...(old?.length !== b.length ? {length: b.length} : {})}, size); - } else delta = {value: clone(b)}; + delta = groupCurrentValues({items: Object.fromEntries(items), ...(old?.length !== b.length ? {length: b.length} : {})}, size, source?.values); + } else delta = literal(b, path); // Sizing does not publish a value. Copy only a replacement that is chosen, // rather than every unchanged subtree and every discarded ancestor value. - let result = inheritedNumbers.has(key) || size(delta) < size({value: b}) ? delta : {value: clone(b)}; + let result = inheritedNumbers.has(key) || size(delta) < size({value: b}) ? delta : literal(b, path); if (reference && size(reference) < size(result)) result = clone(reference); return overlay && size(overlay) < size(result) ? overlay : result; }; @@ -542,13 +570,24 @@ function surfaceOf(value: unknown): Surface { return value as Surface; } +/** The ordinary Python IO path applies this existing edit format to its + * original values. Direct JS callers retain detached JSON results. */ +function publish(request: JsonObject, result: JsonObject, values: ProjectionSourceValues): JsonObject { + if (request.source_edits !== true) return result; + const {operation: _operation, source_edits: _edits, source_number_tokens: _numbers, ...source} = request; + return contextProjectionChanges(source, result, values); +} + /** Hosts may confirm final-input retention, or explicitly select the native * hook delivery assumption. These bases are distinct; neither grants authority. */ export function prepareContextDelivery(request: JsonObject, legacyQuotaFields: readonly string[] = []): JsonObject { const context = requireJsonObject(request.context, "current projection"); + const identity = contextSourceIdentity(request, request.source_number_tokens); + const values: ProjectionSourceValues = new WeakMap(); const surface = surfaceOf(request.surface ?? "work_context"), target = targetBinding(request, context, surface); const full = (reason: string): JsonObject => ({schema_version: VERSION, mode: "full", target, - ancestry: 0, reason, context: clone(context)}); + ancestry: 0, reason, context: request.source_edits === true ? context : clone(context)}); + const fullResult = (reason: string): JsonObject => publish(request, full(reason), values); const baseline = jsonObject(request.baseline), retained = jsonObject(request.retained); const continuity = jsonObject(request.continuity); const binding = jsonObject(baseline?.target), prior = jsonObject(baseline?.context); @@ -561,12 +600,14 @@ export function prepareContextDelivery(request: JsonObject, legacyQuotaFields: r const fileBacked = !retained && continuity?.basis === "captured_file" && continuity.capture_bytes_verified === true && equal(continuity.target, binding); if (!binding || !prior || (!confirmed && !assumed && !fileBacked) || !compatible(binding, target) - || binding.context_digest !== digest(prior)) return full("baseline_unavailable"); + || binding.context_digest !== digest(prior)) return fullResult("baseline_unavailable"); const depth = baseline?.ancestry; - if (!Number.isSafeInteger(depth) || Number(depth) < 0 || Number(depth) >= MAX_ANCESTRY) return full("full_refresh_due"); + if (!Number.isSafeInteger(depth) || Number(depth) < 0 || Number(depth) >= MAX_ANCESTRY) return fullResult("full_refresh_due"); const before = freshPaths(prior, surface, legacyQuotaFields, binding), after = freshPaths(context, surface, legacyQuotaFields, target); - if (!fileBacked && (!before.valid || !after.valid)) return full("source_not_reusable"); - const changes = changeTree(prior, context, fileBacked ? [] : after.paths, fileBacked); + if (!fileBacked && (!before.valid || !after.valid)) return fullResult("source_not_reusable"); + const changes = changeTree(prior, context, fileBacked ? [] : after.paths, fileBacked, + {identity, before: ["baseline", "context"], after: ["context"], + ...(request.source_edits === true ? {values} : {})}); const delta: JsonObject = {schema_version: VERSION, mode: "delta", target, base: binding, ancestry: Number(depth) + 1, changes, ...(fileBacked ? {continuity_basis: "captured_file"} : assumed ? {continuity_basis: "native_hook_assumption"} : {}), @@ -590,13 +631,13 @@ export function prepareContextDelivery(request: JsonObject, legacyQuotaFields: r + "Values assigns complete fields (exact indices with items); changed/items recurse, removed deletes object keys, length sets array length. " + "Within the same work Turn, compaction alone requires neither another guard nor replay of already-read context. " + "Continue from the working brief and original work/settlement identity; missing facts use same-capture recovery."}; - if (bytes(delta) >= bytes(full("delta_not_smaller"))) return full("delta_not_smaller"); - if (!equal(applyTree(prior, changes), context)) return full("reconstruction_failed"); + if (bytes(delta) >= bytes(full("delta_not_smaller"))) return fullResult("delta_not_smaller"); + if (!equal(applyTree(prior, changes), context)) return fullResult("reconstruction_failed"); // Native consumers page the wire before they can use its baseline. Deliver // the existing basis and interpretation before the potentially large tree; // do not duplicate them in a second host instruction or change strict IO. - return assumed || fileBacked ? {schema_version: VERSION, mode: "delta", - continuity_basis: delta.continuity_basis, instruction: delta.instruction, ...delta} : delta; + return publish(request, assumed || fileBacked ? {schema_version: VERSION, mode: "delta", + continuity_basis: delta.continuity_basis, instruction: delta.instruction, ...delta} : delta, values); } /** Validate the exact change tree against the immutable CURRENT capture before @@ -609,7 +650,11 @@ export function restoreContextDelivery(request: JsonObject, legacyQuotaFields: r if (current.mode !== "full" || current.schema_version !== VERSION || target.projection_version !== VERSION || target.context_digest !== digest(currentContext)) { throw new Error("current full capture is invalid"); } - const fallback = (): JsonObject => ({...clone(current), reason: "automatic_full_recovery", ancestry: 0}); + const identity = contextSourceIdentity(request, request.source_number_tokens); + const values: ProjectionSourceValues = new WeakMap(); + const copy = (value: T): T => request.source_edits === true ? value : clone(value); + const fallback = (): JsonObject => publish(request, + {...copy(current), reason: "automatic_full_recovery", ancestry: 0}, values); try { const surface = surfaceOf(target.surface); const delivery = requireJsonObject(request.delivery, "delivery"), baseline = requireJsonObject(request.baseline, "baseline"); @@ -621,12 +666,15 @@ export function restoreContextDelivery(request: JsonObject, legacyQuotaFields: r || delivery.ancestry !== Number(baseline.ancestry) + 1 || Number(delivery.ancestry) > MAX_ANCESTRY) return fallback(); const before = freshPaths(context, surface, legacyQuotaFields, base), after = freshPaths(currentContext, surface, legacyQuotaFields, target); if (!fileBacked && (!before.valid || !after.valid)) return fallback(); - const expected = changeTree(context, currentContext, fileBacked ? [] : after.paths, fileBacked); + const expected = changeTree(context, currentContext, fileBacked ? [] : after.paths, fileBacked, + {identity, before: ["baseline", "context"], after: ["current", "context"], + ...(request.source_edits === true ? {values} : {})}); // Also rejects omitted unchanged controls, redundant/conflicting edits, // malformed array indices/lengths and unsupported operations before applying. - if (!equal(delivery.changes, expected)) return fallback(); + if (contextProjectionIdentity(request, {changes: delivery.changes}, request.source_number_tokens, values) + !== contextProjectionIdentity(request, {changes: expected}, request.source_number_tokens, values)) return fallback(); const result = applyTree(context, expected); if (!equal(result, currentContext)) return fallback(); - return {...clone(delivery), context: result}; + return publish(request, {...copy(delivery), context: copy(currentContext)}, values); } catch { return fallback(); } } diff --git a/loopx/control_plane/quota/diagnostic_projection.ts b/loopx/control_plane/quota/diagnostic_projection.ts index a37dab534d..0b3ddca22d 100644 --- a/loopx/control_plane/quota/diagnostic_projection.ts +++ b/loopx/control_plane/quota/diagnostic_projection.ts @@ -7,17 +7,15 @@ import type {JsonObject} from "../effect_program.ts"; import {jsonObject} from "../runtime_decode.ts"; import {TODO_CANONICAL_READ_RECORD_FIELDS} from "../coordination/coordination_state_contract.ts"; -import {canonicalAuthorityJson} from "../coordination/authority_store_codec.ts"; +import {canonicalAuthorityJson, authorityUnicodeCompare} from "../coordination/authority_store_codec.ts"; +import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts"; type ProjectionPath = (string | number)[]; type ProjectionChange = {path: ProjectionPath; value: unknown} | {path: ProjectionPath; from: ProjectionPath} | {path: ProjectionPath; remove: true}; -/** Return only display edits, never round-trip unrelated source facts through - * JavaScript numbers. References address the immutable input, including bodies - * moved into source_projection. Python applies these edits without policy. */ -export function contextProjectionChanges(source: JsonObject, projected: JsonObject): JsonObject { - const paths = new WeakMap(), changes: ProjectionChange[] = []; +function contextObjectPaths(source: JsonObject): WeakMap { + const paths = new WeakMap(); const index = (value: unknown, path: ProjectionPath): void => { if (value === null || typeof value !== "object" || paths.has(value)) return; paths.set(value, path); @@ -25,14 +23,71 @@ export function contextProjectionChanges(source: JsonObject, projected: JsonObje else Object.entries(value).forEach(([key, item]) => index(item, [...path, key])); }; index(source, []); - const visit = (before: unknown, after: unknown, path: ProjectionPath): void => { + return paths; +} + +/** Sparse IO number spellings distinguish facts already rounded by JSON.parse. + * This is source identity for pure projection, not a second decision owner or + * a new hash codec. Native JS callers need no metadata. Never copy full bodies + * just to decide whether two registered carriers can share their source. */ +export function contextSourceIdentity(source: JsonObject, supplied?: unknown): + (value: unknown, path: ProjectionPath) => string { + if (supplied === undefined) return value => JSON.stringify(canonicalAuthorityJson(value)); + if (!Array.isArray(supplied)) throw new EffectRuntimeRequestError("context source number tokens must be an array"); + const tokens: {path: ProjectionPath; token: string}[] = [], seen = new Set(); + for (const entry of supplied) { + if (!Array.isArray(entry) || entry.length !== 2 || !Array.isArray(entry[0]) || !entry[0].length || + typeof entry[1] !== "string" || !/^-?(?:0|[1-9]\d*)(?:\.\d+)?(?:[eE][+-]?\d+)?$/.test(entry[1])) { + throw new EffectRuntimeRequestError("invalid context source number token"); + } + const path: ProjectionPath = entry[0], token: string = entry[1]; + let value: unknown = source; + for (const key of path) { + if (value === null || typeof value !== "object" || + !(Array.isArray(value) ? typeof key === "number" && Number.isSafeInteger(key) && key >= 0 + : typeof key === "string") || !Object.hasOwn(value, key)) { + throw new EffectRuntimeRequestError("context source number token path is unavailable"); + } + value = (value as Record)[key]; + } + const key = JSON.stringify(path); + if (seen.has(key) || typeof value !== "number" || !Number.isFinite(value) || !Object.is(JSON.parse(token), value)) { + throw new EffectRuntimeRequestError("context source number token differs from input"); + } + seen.add(key); + tokens.push({path, token}); + } + const paths = contextObjectPaths(source); + return (value, path) => { + const original = value !== null && typeof value === "object" ? paths.get(value) : undefined; + const prefix = original ?? path; + const relative = tokens.filter(item => prefix.every((key, i) => item.path[i] === key)) + .map(item => [item.path.slice(prefix.length), item.token]); + // Sort encoded relative paths, so original object insertion order is not + // mistaken for a content difference. Keep token strings separate from data. + relative.sort((left, right) => authorityUnicodeCompare(JSON.stringify(left[0]), JSON.stringify(right[0]))); + return JSON.stringify([canonicalAuthorityJson(value), relative]); + }; +} + +/** Return only display edits, never round-trip unrelated source facts through + * JavaScript numbers. References address the immutable input, including bodies + * moved into source_projection. Python applies these edits without policy. */ +export type ProjectionSourceValues = WeakMap>; + +export function contextProjectionChanges(source: JsonObject, projected: JsonObject, + sourceValues?: ProjectionSourceValues): JsonObject { + const paths = contextObjectPaths(source), changes: ProjectionChange[] = []; + const visit = (before: unknown, after: unknown, path: ProjectionPath, parent?: object): void => { + const original = parent && sourceValues?.get(parent)?.get(String(path.at(-1))); + if (original) { changes.push({path, from: original}); return; } if (Object.is(before, after)) return; const from = after !== null && typeof after === "object" ? paths.get(after) : undefined; if (from) { changes.push({path, from}); return; } if (Array.isArray(after)) { const old = Array.isArray(before) ? before : []; if (!Array.isArray(before)) changes.push({path, value: []}); - after.forEach((value, i) => visit(old[i], value, [...path, i])); + after.forEach((value, i) => visit(old[i], value, [...path, i], after)); // Remove from the end so earlier indices retain their original meaning. for (let i = old.length - 1; i >= after.length; i--) changes.push({path: [...path, i], remove: true}); } else if (jsonObject(after)) { @@ -41,13 +96,33 @@ export function contextProjectionChanges(source: JsonObject, projected: JsonObje for (const key of Object.keys(old)) if (!Object.hasOwn(after as JsonObject, key)) changes.push({path: [...path, key], remove: true}); for (const [key, value] of Object.entries(after as JsonObject)) - visit(Object.hasOwn(old, key) ? old[key] : undefined, value, [...path, key]); + visit(Object.hasOwn(old, key) ? old[key] : undefined, value, [...path, key], after as object); } else changes.push({path, value: after}); }; visit(source, projected, []); return {changes}; } +/** Compare a pure edit result using the same original number spellings. + * References are IO lineage, not an alternate context/hash or decision owner. */ +export function contextProjectionIdentity(source: JsonObject, projected: JsonObject, + supplied?: unknown, sourceValues?: ProjectionSourceValues): string { + if (supplied === undefined) return JSON.stringify(canonicalAuthorityJson(projected)); + // Validate every supplied path/token before deriving identities from it. + contextSourceIdentity(source, supplied); + const tokens = supplied as [ProjectionPath, string][]; + const result: [ProjectionPath, string][] = []; + const edits = contextProjectionChanges(source, projected, sourceValues).changes as ProjectionChange[]; + for (const edit of edits) { + if (!("from" in edit)) continue; + for (const [path, token] of tokens) { + if (edit.from.every((key, i) => path[i] === key)) + result.push([[...edit.path, ...path.slice(edit.from.length)], token]); + } + } + return contextSourceIdentity(projected, result)(projected, []); +} + const AGENT_LIMITS: Readonly> = { first_executable_items: 3, unclaimed_priority_open_items: 3, monitor_due_items: 1, monitor_capability_blocked_due_items: 2, diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py index d6ec35bb27..1428f3d710 100644 --- a/loopx/control_plane/quota/turn_envelope.py +++ b/loopx/control_plane/quota/turn_envelope.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import math from collections.abc import Mapping from dataclasses import dataclass from typing import Any @@ -55,6 +56,10 @@ def _turn_envelope_result( """Call the coarse TS owner and fail closed on transport shape drift.""" request = {"operation": operation, **params} + if operation in {"context_delivery", "context_recovery"}: + tokens = _source_number_tokens(params) + if tokens: + request["source_number_tokens"] = tokens try: result = effect_runtime_result("quota.turn_envelope.evaluate", request) except EffectRuntimeRejected as error: @@ -69,6 +74,30 @@ def _turn_envelope_result( return dict(result) +def _source_number_tokens(value: object) -> list[list[Any]]: + """Transport sparse numeric source identity; no carrier or sharing policy.""" + tokens: list[list[Any]] = [] + + def visit(item: object, path: list[str | int]) -> None: + if isinstance(item, float) and not math.isfinite(item): + raise ValueError("Context source numbers must be finite") + # Fractional doubles retain their identity through JSON/JavaScript. + # Only integral floats (including signed zero) and unsafe integers + # can become indistinguishable from a different Python source number. + if (isinstance(item, float) and item.is_integer() + or type(item) is int and abs(item) > 2**53 - 1): + tokens.append([path, json.dumps(item, allow_nan=False)]) + elif isinstance(item, Mapping): + for key, child in item.items(): + visit(child, [*path, key]) + elif isinstance(item, (list, tuple)): + for index, child in enumerate(item): + visit(child, [*path, index]) + + visit(value, []) + return tokens + + def assert_turn_envelope_consumer_version(envelope: Mapping[str, Any]) -> None: """Use the typed wire-version gate before any Host consumes this packet.""" _turn_envelope_result("consumer_version", envelope=dict(envelope)) @@ -196,11 +225,14 @@ def prepare_context_delivery( Unsupported hosts omit retention and keep complete delivery. A candidate is not an ACK. Confirmed input and native delivery assumptions stay distinct. """ - return _turn_envelope_result("context_delivery", context=dict(context), - scope=dict(scope), capture_id=capture_id, surface=surface, + from .cli_projection import _apply_context_projection_changes + + source = dict(context=dict(context), scope=dict(scope), capture_id=capture_id, surface=surface, baseline=dict(baseline) if baseline is not None else None, retained=dict(retained) if retained is not None else None, continuity=dict(continuity) if continuity is not None else None) + result = _turn_envelope_result("context_delivery", **source, source_edits=True) + return _apply_context_projection_changes(source, result) def update_native_context_lifecycle(state: Mapping[str, Any], event: Mapping[str, Any], *, @@ -218,6 +250,9 @@ def restore_context_delivery( current: Mapping[str, Any], ) -> dict[str, Any]: """Validate/reconstruct or return the same-capture full body without effects.""" - return _turn_envelope_result("context_recovery", delivery=dict(delivery), - baseline=dict(baseline) if baseline is not None else None, + from .cli_projection import _apply_context_projection_changes + + source = dict(delivery=dict(delivery), baseline=dict(baseline) if baseline is not None else None, current=dict(current)) + result = _turn_envelope_result("context_recovery", **source, source_edits=True) + return _apply_context_projection_changes(source, result) diff --git a/tests/control_plane/test_context_delivery_cli.py b/tests/control_plane/test_context_delivery_cli.py index 4964d0cf3d..5974effe93 100644 --- a/tests/control_plane/test_context_delivery_cli.py +++ b/tests/control_plane/test_context_delivery_cli.py @@ -239,12 +239,14 @@ def test_cli_capture_exact_delta_and_same_capture_recovery(tmp_path: Path, monke scope = dict(goal_id=GOAL_ID, agent_id=AGENT_ID, session_id=profile, context_epoch="epoch-a") full = prepare_context_delivery(context, scope=scope, capture_id="capture-a") assert full["mode"] == "full" and full["context"] == context + assert json.dumps(full["context"], sort_keys=True) == json.dumps(context, sort_keys=True) warm = prepare_context_delivery(context, scope=scope, capture_id="capture-b", baseline=full, retained={"complete": True, "target": full["target"]}) assert warm["mode"] == "delta" assert len(json.dumps(warm).encode()) < len(json.dumps(full).encode()) / 2 current = prepare_context_delivery(context, scope=scope, capture_id="capture-b") - assert restore_context_delivery(warm, baseline=full, current=current)["context"] == context + restored = restore_context_delivery(warm, baseline=full, current=current) + assert json.dumps(restored["context"], sort_keys=True) == json.dumps(context, sort_keys=True) turn_full = prepare_context_delivery(turn_context, scope=scope, capture_id="turn-capture") turn_delta = prepare_context_delivery(turn_context, scope=scope, capture_id="turn-capture", baseline=full, retained={"complete": True, "target": full["target"]}) @@ -274,8 +276,10 @@ def test_cli_capture_exact_delta_and_same_capture_recovery(tmp_path: Path, monke assert body not in json.dumps(delta["changes"], ensure_ascii=False) restored = restore_context_delivery(delta, baseline=first, current=next_capture) assert restored["mode"] == "delta" and restored["context"] == projection + assert json.dumps(restored["context"], sort_keys=True) == json.dumps(projection, sort_keys=True) missing = restore_context_delivery(delta, baseline=None, current=next_capture) assert missing["mode"] == "full" and missing["context"] == projection + assert json.dumps(missing["context"], sort_keys=True) == json.dumps(projection, sort_keys=True) # Neither native source state, the signed decision nor settlement is changed # by presentation, warm reuse or recovery. A new capture is not an admission. assert state.read_bytes() == before_source @@ -799,3 +803,132 @@ def test_default_json_reader_and_generated_recovery_use_large_pages(tmp_path): assert len(page["content"].encode()) > 32000 args = parser.parse_args(shlex.split(page["next_command"])[1:]) assert collected == raw and calls <= 10 + + +@pytest.mark.parametrize("number", [1, 1.0, -0.0, 2**60 + 1, float(2**60), + 1e-7, 1e-6, 0.1, 1.0000000000000002, 5e-324, -5e-324, + sys.float_info.min, sys.float_info.max]) +def test_context_codec_keeps_original_numbers_through_full_delta_and_recovery(tmp_path, monkeypatch, number): + isolate_sqlite_runtime(tmp_path, monkeypatch) + scope = dict(goal_id="fixture", agent_id="reader", session_id="native", context_epoch="epoch") + original = {"registry": {"body": "Preserve original evidence. " * 600, "measurement": number}, + "current_control": {"measurement": number, "state": "original"}} + current = deepcopy(original) + current["current_control"]["state"] = "current" + args = dict(scope=scope, surface="quota_packet") + first = prepare_context_delivery(original, capture_id="before", **args) + assert json.dumps(first["context"], sort_keys=True) == json.dumps(original, sort_keys=True) + full = prepare_context_delivery(current, capture_id="after", **args) + warm = prepare_context_delivery(current, capture_id="after", baseline=first, + retained={"complete": True, "target": first["target"]}, **args) + assert warm["mode"] == "delta" + for baseline in [first, None]: + restored = restore_context_delivery(warm, baseline=baseline, current=full) + assert json.dumps(restored["context"], sort_keys=True) == json.dumps(current, sort_keys=True) + assert json.dumps(first["context"], sort_keys=True) == json.dumps(original, sort_keys=True) + + +@pytest.mark.parametrize("before,after", [(1, 1.0), (0, -0.0), (2**60, 2**60 + 1)]) +def test_context_codec_rejects_numeric_delta_collisions(tmp_path, monkeypatch, before, after): + isolate_sqlite_runtime(tmp_path, monkeypatch) + scope = dict(goal_id="fixture", agent_id="reader", session_id="native", context_epoch="epoch") + original = {"registry": {"body": "Keep numeric evidence distinct. " * 600, "measurement": before}} + current = deepcopy(original) + current["registry"]["measurement"] = after + args = dict(scope=scope, surface="quota_packet") + first = prepare_context_delivery(original, capture_id="before", **args) + full = prepare_context_delivery(current, capture_id="after", **args) + warm = prepare_context_delivery(current, capture_id="after", baseline=first, + retained={"complete": True, "target": first["target"]}, **args) + assert warm["mode"] == "delta" + assert warm["changes"] is not None + restored = restore_context_delivery(warm, baseline=first, current=full) + assert restored["mode"] == "delta" + assert json.dumps(restored["context"], sort_keys=True) == json.dumps(current, sort_keys=True) + # Neither omission nor a normalized-equivalent numeric literal is the + # exact current evidence, even when the legacy JS digest is unchanged. + for replacement in [None, {"changed": {"registry": {"changed": {"measurement": {"value": before}}}}}]: + damaged = deepcopy(warm) + damaged["changes"] = replacement + recovered = restore_context_delivery(damaged, baseline=first, current=full) + assert recovered["mode"] == "full" + assert json.dumps(recovered["context"], sort_keys=True) == json.dumps(current, sort_keys=True) + + +@pytest.mark.parametrize("continuity", ["native_hook_assumption", "captured_file"]) +def test_context_numeric_source_reordering_and_grouped_literals(tmp_path, monkeypatch, continuity): + isolate_sqlite_runtime(tmp_path, monkeypatch) + scope = dict(goal_id="fixture", agent_id="reader", session_id="native", context_epoch="epoch") + body = "Keep source numbers, original ordering and metadata. " * 400 + original = {"registry": {"body": body, "observations": [1.0, -0.0, 2**60 + 1]}, + "current_control": {"a": 1.0, "b": -0.0, "c": 2**60 + 1}} + current = {"registry": {"body": body, "observations": [2**60 + 1, 1.0, -0.0]}, + "current_control": {"a": -0.0, "b": 2**60 + 1, "c": 1.0}} + args = dict(scope=scope, surface="quota_packet") + first = prepare_context_delivery(original, capture_id="before", **args) + full = prepare_context_delivery(current, capture_id="after", **args) + delta = prepare_context_delivery(current, capture_id="after", baseline=first, + continuity={"basis": continuity, "capture_bytes_verified": True, "target": first["target"]}, **args) + assert delta["mode"] == "delta" and delta["continuity_basis"] == continuity + restored = restore_context_delivery(delta, baseline=first, current=full) + assert restored["mode"] == "delta" + assert json.dumps(restored["context"], sort_keys=True) == json.dumps(current, sort_keys=True) + assert json.dumps(first["context"], sort_keys=True) == json.dumps(original, sort_keys=True) + + +@pytest.mark.parametrize("number", [float("nan"), float("inf"), -float("inf")]) +def test_context_nonfinite_numbers_are_rejected_before_rpc(monkeypatch, number): + from loopx.control_plane.quota import turn_envelope + + def unexpected(*args, **kwargs): + pytest.fail("Non-finite evidence must not enter the codec") + + monkeypatch.setattr(turn_envelope, "effect_runtime_result", unexpected) + with pytest.raises(ValueError, match="finite"): + prepare_context_delivery({"nested": [{"measurement": number}]}, + scope={}, capture_id="invalid") + + +def test_fractional_observations_keep_complete_codec_cost_proportional(tmp_path, monkeypatch): + """Count both sides of the real RPC, including full and recovery paths.""" + from loopx.control_plane.quota import turn_envelope + + isolate_sqlite_runtime(tmp_path, monkeypatch) + original_rpc = turn_envelope.effect_runtime_result + costs = [] + + def size(value): + return len(json.dumps(value, ensure_ascii=False, separators=(",", ":"), + allow_nan=False).encode()) + + def measured_rpc(method, params, **kwargs): + result = original_rpc(method, params, **kwargs) + costs.append(size(params) + size(result)) + return result + + monkeypatch.setattr(turn_envelope, "effect_runtime_result", measured_rpc) + totals, source_sizes = [], [] + for fraction in (False, True): + costs.clear() + before = {"registry": {"values": [i + 0.5 if fraction else i for i in range(1000)]}} + after = deepcopy(before) + after["registry"]["values"][-1] += 1 + args = dict(scope=dict(goal_id="fixture", agent_id="reader", session_id="native", + context_epoch="epoch"), surface="quota_packet") + first = prepare_context_delivery(before, capture_id="before", **args) + current = prepare_context_delivery(after, capture_id="after", **args) + delta = prepare_context_delivery(after, capture_id="after", baseline=first, + continuity={"basis": "captured_file", "capture_bytes_verified": True, + "target": first["target"]}, **args) + assert delta["mode"] == "delta" + assert json.dumps(first["context"]) == json.dumps(before) + for baseline in (first, None): + restored = restore_context_delivery(delta, baseline=baseline, current=current) + assert json.dumps(restored["context"]) == json.dumps(after) + assert len(costs) == 5 + totals.append(sum(costs)) + source_sizes.append(size(before) + size(after)) + # Fractions need more JSON digits, not per-element identity metadata. + # Allow payload growth plus 25% structural headroom over the matched + # safe-integer workload; count complete requests and results, not deltas alone. + assert totals[1] / totals[0] <= 1.25 * source_sizes[1] / source_sizes[0] diff --git a/tests/control_plane_ts/context_source_identity.test.ts b/tests/control_plane_ts/context_source_identity.test.ts new file mode 100644 index 0000000000..9ed05f2c6e --- /dev/null +++ b/tests/control_plane_ts/context_source_identity.test.ts @@ -0,0 +1,43 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import {contextSourceIdentity, contextProjectionChanges, contextProjectionIdentity} from "../../loopx/control_plane/quota/diagnostic_projection.ts"; + +test("source identity keeps number spellings separate from strings and retains later exact owners", () => { + const source = {first: {observations: [1, 2 ** 60]}, + second: {observations: [1, 2 ** 60]}, third: {observations: [1, 2 ** 60]}, + string: {observations: ["1.0", "1152921504606846977"]}}; + const key = contextSourceIdentity(source, [ + [["first", "observations", 1], "1152921504606846976"], + [["second", "observations", 0], "1.0"], + [["second", "observations", 1], "1152921504606846977"], + [["third", "observations", 1], "1152921504606846977"], + [["third", "observations", 0], "1.0"], + ]); + assert.notEqual(key(source.first, ["first"]), key(source.second, ["second"])); + assert.equal(key(source.second, ["second"]), key(source.third, ["third"])); + assert.notEqual(key(source.second, ["second"]), key(source.string, ["string"])); +}); + +for (const tokens of [null, {}, [[[], "1"]], [[["missing"], "1"]], + [[["value"], "2"]], [[["value"], "NaN"]], [[["value"], "01"]], + [[["value"], "1"], [["value"], "1.0"]], [[["text"], "1"]], + [[["rows", "0"], "1.0"]]]) { + test(`source identity rejects invalid IO tokens: ${JSON.stringify(tokens)}`, () => { + assert.throws(() => contextSourceIdentity({value: 1, text: "1", rows: [1]}, tokens), + /context source number token/); + }); +} + +test("source edit identities follow retained objects and primitive source references", () => { + const source = {before: {measurement: 1}, current: {measurement: 1}, raw: -0}; + const tokens = [[["current", "measurement"], "1.0"], [["raw"], "-0.0"]]; + const value = {value: source.raw}, projected = {body: source.current, literal: value}; + const origins = new WeakMap([[value, new Map([["value", ["raw"]]])]]); + const edits = contextProjectionChanges(source, projected, origins).changes; + assert.ok(Array.isArray(edits)); + assert.ok(edits.some(edit => JSON.stringify(edit) === JSON.stringify({path: ["literal", "value"], from: ["raw"]}))); + assert.notEqual(contextProjectionIdentity(source, {body: source.before}, tokens), + contextProjectionIdentity(source, {body: source.current}, tokens)); + assert.notEqual(contextProjectionIdentity(source, {literal: {value: 0}}, tokens), + contextProjectionIdentity(source, {literal: value}, tokens, origins)); +});