Skip to content

Validate quota audit references before a staged replan #8113

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
huangruiteng merged 1 commit into from
Nov 6, 2026
Merged

Validate quota audit references before a staged replan #8113

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
137 changes: 56 additions & 81 deletions loopx/control_plane/quota/cli_projection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -291,57 +291,6 @@ def _reference_inline_replan_authoring(
return compact


def _vision_continuation_ref(audit: dict[str, Any]) -> dict[str, Any]:
compact = {
key: audit[key]
for key in ("schema_version", "required", "decision")
if key in audit
}
judge = audit.get("vision_gap_judge")
if isinstance(judge, dict):
compact["vision_gap_judge"] = {
key: judge[key] for key in ("done", "decision") if key in judge
}
compact["vision_gap_judge"]["projection_ref"] = (
"$.vision_continuation_audit.vision_gap_judge"
)
compact["projection_ref"] = "$.vision_continuation_audit"
return compact


def _replace_nested_vision_continuation_audits(
payload: dict[str, Any],
*,
source_audit: dict[str, Any],
audit_ref: dict[str, Any],
) -> None:
source_key = _json_content_key(source_audit)
frontier = payload.get("goal_frontier_projection")
if (
isinstance(frontier, dict)
and _json_content_key(frontier.get("vision_continuation_audit")) == source_key
):
compact_frontier = dict(frontier)
compact_frontier["vision_continuation_audit"] = dict(audit_ref)
payload["goal_frontier_projection"] = compact_frontier

interaction = payload.get("interaction_contract")
if not isinstance(interaction, dict):
return
compact_interaction = dict(interaction)
for channel_name in ("agent_channel", "cli_channel"):
channel = compact_interaction.get(channel_name)
if (
not isinstance(channel, dict)
or _json_content_key(channel.get("vision_continuation_audit")) != source_key
):
continue
compact_channel = dict(channel)
compact_channel["vision_continuation_audit"] = dict(audit_ref)
compact_interaction[channel_name] = compact_channel
payload["interaction_contract"] = compact_interaction


def _deduplicate_next_action_warning(
warning: dict[str, Any],
*,
Expand Down Expand Up @@ -491,7 +440,8 @@ def _promote_interaction_contract(payload: dict[str, Any]) -> dict[str, Any]:

def _reference_context(
payload: dict[str, Any], *, task_title: bool = True, explore_guidance: bool = True,
goal_acceptance: bool = False, diagnostics: dict[str, Any] | None = None,
goal_acceptance: bool = False, vision_audit: bool = False,
diagnostics: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""One coarse transport call for the existing typed context owner."""
contract = payload.get("interaction_contract")
Expand All @@ -502,25 +452,25 @@ def _reference_context(
if not needs_context and (diagnostics is None or (
diagnostics.get("agent_detail") is True and diagnostics.get("user_detail") is True
)):
if not goal_acceptance:
return payload
# A reference can only be created or relocated when two literal owners
# are present. Check structural presence here, leave equality, schema
# and the choice of owner to TypeScript.
def contains_contract(value: Any, *path: str) -> bool:
for key in path:
value = value.get(key) if isinstance(value, dict) else None
return isinstance(value, dict) and bool(value)

carriers = (
any(contains_contract(row, "content", "goal_acceptance_contract") for row in sources)
if isinstance(sources, list) else False,
contains_contract(payload, "planning_horizon", "core_goal", "acceptance_contract"),
contains_contract(payload, "autonomous_replan_obligation", "replan_context", "core_goal", "acceptance_contract"),
contains_contract(payload, "agent_todo_summary", "goal_acceptance_contract"),
)
if sum(carriers) < 2:
if not goal_acceptance and not vision_audit:
return payload
if goal_acceptance:
# A reference can only be created or relocated when two literal
# owners are present. Equality, schema and owner choice stay in TS.
def contains_contract(value: Any, *path: str) -> bool:
for key in path:
value = value.get(key) if isinstance(value, dict) else None
return isinstance(value, dict) and bool(value)

carriers = (
any(contains_contract(row, "content", "goal_acceptance_contract") for row in sources)
if isinstance(sources, list) else False,
contains_contract(payload, "planning_horizon", "core_goal", "acceptance_contract"),
contains_contract(payload, "autonomous_replan_obligation", "replan_context", "core_goal", "acceptance_contract"),
contains_contract(payload, "agent_todo_summary", "goal_acceptance_contract"),
)
if sum(carriers) < 2 and not vision_audit:
return payload
from .turn_envelope import _turn_envelope_result

# The typed projection reads and edits these roots. Untouched capture
Expand All @@ -530,9 +480,39 @@ def contains_contract(value: Any, *path: str) -> bool:
"todo_summary_projection", "planning_horizon", "autonomous_replan_obligation",
"interaction_contract", "agent_lane_next_action", "selected_todo",
) if key in payload}
if (
isinstance(contract, dict)
and isinstance(agent, dict)
and isinstance(context, dict)
and isinstance(context.get("sources"), list)
):
# Edits address rows by their original index. Keep unrelated rows in
# place without sending their bodies through the display RPC.
rows = [
row if isinstance(row, dict) and (
(task_title and row.get("source") == "selected_todo")
or (explore_guidance and row.get("kind") == "explore_turn_context")
or (goal_acceptance and row.get("source") == "goal_acceptance")
) else None
for row in context["sources"]
]
source["interaction_contract"] = {
**contract,
"agent_channel": {
**agent,
"work_context": {**context, "sources": rows},
},
}
if vision_audit:
source["vision_continuation_audit"] = payload["vision_continuation_audit"]
frontier = payload.get("goal_frontier_projection")
if isinstance(frontier, dict) and "vision_continuation_audit" in frontier:
source["goal_frontier_projection"] = {"vision_continuation_audit": frontier["vision_continuation_audit"]}
result = _turn_envelope_result("reference_context", payload=source,
task_title=task_title, explore_guidance=explore_guidance,
goal_acceptance=goal_acceptance, **({"diagnostics": diagnostics} if diagnostics is not None else {}))
goal_acceptance=goal_acceptance,
**({"vision_audit": True} if vision_audit else {}),
**({"diagnostics": diagnostics} if diagnostics is not None else {}))
return _apply_context_projection_changes(payload, result)


Expand Down Expand Up @@ -609,15 +589,7 @@ def compact_quota_should_run_cli_payload(
compact = dict(compact)
compact["goal_boundary"] = compact_goal_boundary
vision_audit = payload.get("vision_continuation_audit")
if not include_vision_detail and isinstance(vision_audit, dict):
compact = dict(compact)
compact_vision_audit = _compact_vision_continuation_audit(vision_audit)
compact["vision_continuation_audit"] = compact_vision_audit
_replace_nested_vision_continuation_audits(
compact,
source_audit=vision_audit,
audit_ref=_vision_continuation_ref(compact_vision_audit),
)
share_audit = not include_vision_detail and isinstance(vision_audit, dict)
replan_action = payload.get("replan_action_packet")
if not include_vision_detail and isinstance(replan_action, dict):
compact_replan_action = _reference_inline_replan_authoring(
Expand Down Expand Up @@ -657,7 +629,10 @@ def compact_quota_should_run_cli_payload(
diagnostics = {"agent_detail": include_todo_summary_detail,
"user_detail": include_user_todo_summary_detail}
compact = _reference_context(compact, task_title=not include_todo_summary_detail,
explore_guidance=not include_vision_detail, goal_acceptance=True, diagnostics=diagnostics)
goal_acceptance=True, vision_audit=share_audit,
explore_guidance=not include_vision_detail, diagnostics=diagnostics)
if share_audit and isinstance(vision_audit, dict):
compact = {**compact, "vision_continuation_audit": _compact_vision_continuation_audit(vision_audit)}
return _promote_interaction_contract(
_promote_runtime_capability_reentry(compact)
)
Expand Down
4 changes: 3 additions & 1 deletion loopx/control_plane/quota/turn_envelope.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,9 @@ def _turn_envelope_result(
"""Call the coarse TS owner and fail closed on transport shape drift."""

request = {"operation": operation, **params}
if operation == "reference_context" and params.get("goal_acceptance"):
if operation == "reference_context" and (
params.get("goal_acceptance") or params.get("vision_audit")
):
tokens = source_number_tokens(params["payload"])
if tokens:
request["source_number_tokens"] = tokens
Expand Down
67 changes: 67 additions & 0 deletions loopx/control_plane/quota/turn_envelope.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -386,6 +386,70 @@ function referenceExploreGuidance(payload: JsonObject): JsonObject {
work_context: {...context, sources}}}};
}

/** Keep the complete vision audit once. These existing projection references
* share identical facts only; a channel-specific obligation is never a summary
* of the root. The old display adapter delegates here before adding metadata. */
function visionAuditReference(audit: JsonObject): JsonObject {
const reference: JsonObject = {projection_ref: "$.vision_continuation_audit"};
for (const key of ["schema_version", "required", "decision"]) {
if (Object.hasOwn(audit, key)) reference[key] = audit[key];
}
if (typeof audit.vision_gap_judge === "object" && audit.vision_gap_judge !== null &&
!Array.isArray(audit.vision_gap_judge)) {
const judge = object(audit.vision_gap_judge);
reference.vision_gap_judge = {projection_ref: "$.vision_continuation_audit.vision_gap_judge"};
for (const key of ["done", "decision"]) {
if (Object.hasOwn(judge, key)) (reference.vision_gap_judge as JsonObject)[key] = judge[key];
}
}
return reference;
}

function mapNestedVisionAudits(payload: JsonObject,
visit: (audit: JsonObject, path: (string | number)[]) => JsonObject): JsonObject {
const map = (value: unknown, path: (string | number)[]) => {
const parent = object(value), audit = parent.vision_continuation_audit;
if (typeof audit !== "object" || audit === null || Array.isArray(audit)) return value;
return {...parent, vision_continuation_audit: visit(audit as JsonObject, [...path, "vision_continuation_audit"])};
};
const rawInteraction = payload.interaction_contract, interaction = object(rawInteraction);
const hasInteraction = typeof rawInteraction === "object" && rawInteraction !== null && !Array.isArray(rawInteraction);
return {...payload,
...(Object.hasOwn(payload, "goal_frontier_projection") ? {
goal_frontier_projection: map(payload.goal_frontier_projection, ["goal_frontier_projection"])} : {}),
...(hasInteraction ? {interaction_contract: {...interaction,
...Object.fromEntries(["agent_channel", "cli_channel"].filter(key => Object.hasOwn(interaction, key))
.map(key => [key, map(interaction[key], ["interaction_contract", key])]))}} : {})};
}

function requireVisionAuditReferences(payload: JsonObject): void {
const owner = object(payload.vision_continuation_audit);
const chained = Object.hasOwn(owner, "projection_ref") || Object.hasOwn(object(owner.vision_gap_judge), "projection_ref");
if (chained) throw new EffectRuntimeRequestError("Vision audit reference requires a complete inline root owner");
mapNestedVisionAudits(payload, audit => {
if (Object.hasOwn(audit, "projection_ref") || Object.hasOwn(object(audit.vision_gap_judge), "projection_ref")) {
if (!Object.keys(owner).length ||
JSON.stringify(canonicalValue(audit)) !== JSON.stringify(canonicalValue(visionAuditReference(owner)))) {
throw new EffectRuntimeRequestError("Vision audit reference must match its complete inline root owner without conflicting facts");
}
}
return audit;
});
}

function referenceVisionContinuation(payload: JsonObject,
identity = contextSourceIdentity(payload)): JsonObject {
requireVisionAuditReferences(payload);
const owner = object(payload.vision_continuation_audit);
if (!Object.keys(owner).length) return payload;
const encoded = JSON.stringify(canonicalValue(owner)), reference = visionAuditReference(owner);
// A short audit is cheaper inline. Do not create a reference merely to count sharing.
if (Buffer.byteLength(encoded, "utf8") <= Buffer.byteLength(JSON.stringify(reference), "utf8")) return payload;
const source = identity(owner, ["vision_continuation_audit"]);
return mapNestedVisionAudits(payload, (audit, path) =>
identity(audit, path) === source ? structuredClone(reference) : audit);
}

function currentTaskTitleOwner(payload: JsonObject): {index: number; todo: JsonObject} | null {
const lane = object(payload.agent_lane_next_action), selected = object(payload.selected_todo);
const context = object(object(object(payload.interaction_contract).agent_channel).work_context);
Expand Down Expand Up @@ -582,6 +646,7 @@ function replanView(request: JsonObject): JsonObject {
if (request.source_hash !== canonicalHash(payload) || payload.ok !== true) {
throw new EffectRuntimeRequestError("Replan read requires the original successful decision and matching source hash");
}
requireVisionAuditReferences(payload);
const packet = object(payload.replan_action_packet);
const obligation = object(payload.autonomous_replan_obligation);
const context = object(obligation.replan_context);
Expand Down Expand Up @@ -1590,6 +1655,8 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject {
if (operation === "reference_context") {
const source = requireJsonObject(request.payload, "quota decision");
let payload = source;
if (request.vision_audit === true) payload = referenceVisionContinuation(source,
contextSourceIdentity(source, request.source_number_tokens));
if (jsonObject(request.diagnostics)) payload = projectQuotaDiagnostics(payload, request.diagnostics as JsonObject);
if (request.goal_acceptance === true) payload = referenceGoalAcceptance(payload,
contextSourceIdentity(source, request.source_number_tokens));
Expand Down
Loading
Loading