Skip to content
Merged
Changes from all commits
Commits
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
24 changes: 14 additions & 10 deletions loopx/control_plane/todos/decision_scope.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
"""Legacy input codec for the single typed decision-dependency rule owner."""
from __future__ import annotations

from typing import Any
from typing import Any, cast

from ..effect_runtime import effect_runtime_result
from .contract import (
Expand Down Expand Up @@ -90,8 +90,10 @@ def standing_decision_authority_for_agent(authority: dict[str, Any] | None, *,
agent_id: str | None) -> dict[str, Any] | None:
if _authority(authority) is None:
return None
return _evaluate("standing", authority=_authority(authority),
agent_id=normalize_todo_claimed_by(agent_id))
return cast(dict[str, Any] | None, _evaluate(
"standing", authority=_authority(authority),
agent_id=normalize_todo_claimed_by(agent_id),
))


def build_required_decision_scope_consistency(
Expand All @@ -101,14 +103,14 @@ def build_required_decision_scope_consistency(
user_source_items: list[dict[str, Any]] | None = None,
standing_decision_authority: dict[str, Any] | None = None,
) -> dict[str, Any]:
return _evaluate("consistency",
return cast(dict[str, Any], _evaluate("consistency",
agent_items=_source(agent_todo_summary, agent_source_items, _AGENT_SUMMARY_ITEM_KEYS),
user_items=_source(user_todo_summary, user_source_items, _USER_SUMMARY_ITEM_KEYS),
agent_id=normalize_todo_claimed_by(agent_id),
registered_agents=sorted({value for raw in registered_agent_ids or []
if (value := normalize_todo_claimed_by(raw))}),
standing_authority=_authority(standing_decision_authority),
)
))


def build_required_decision_scope_repair_hint(
Expand Down Expand Up @@ -193,26 +195,28 @@ def decision_scope_covers(gate_scope: Any, required_scope: Any) -> bool:
required = normalize_todo_decision_scope(required_scope)
if not gate or not required:
return False
return _evaluate("covers", gate_scope=gate, required_scope=required)
return cast(bool, _evaluate("covers", gate_scope=gate, required_scope=required))


def decision_scope_gate_relation(gate: dict[str, Any], agent_item: dict[str, Any]) -> dict[str, Any] | None:
return _evaluate("scope_relation", gate=_facts(gate), item=_facts(agent_item))
return cast(dict[str, Any] | None, _evaluate("scope_relation", gate=_facts(gate), item=_facts(agent_item)))


def exact_todo_gate_relation(gate: dict[str, Any], agent_item: dict[str, Any]) -> dict[str, Any] | None:
return _evaluate("exact_relation", gate=_facts(gate), item=_facts(agent_item))
return cast(dict[str, Any] | None, _evaluate("exact_relation", gate=_facts(gate), item=_facts(agent_item)))


def todo_gate_relation(gate: dict[str, Any], agent_item: dict[str, Any]) -> dict[str, Any] | None:
return _evaluate("relation", gate=_facts(gate), item=_facts(agent_item))
return cast(dict[str, Any] | None, _evaluate("relation", gate=_facts(gate), item=_facts(agent_item)))


def todo_gate_relations(gates: list[dict[str, Any]], items: list[dict[str, Any]]) -> list[list[dict[str, Any] | None]]:
"""Evaluate a consumer's candidate set in one RPC, retaining positional identity."""
if not gates or not items:
return [[] for _ in gates]
return _evaluate("relations", gates=[_facts(gate) for gate in gates], items=[_facts(item) for item in items])
return cast(list[list[dict[str, Any] | None]], _evaluate(
"relations", gates=[_facts(gate) for gate in gates], items=[_facts(item) for item in items],
))


def todo_gate_relation_blocks_agent(relation: dict[str, Any] | None) -> bool:
Expand Down