Deciding whether an authored skill applies to the state in front of you.
seed_authored_skill accepts a precondition list that mixes two kinds of
entry: a bare tool identifier -- call this first -- and a state predicate,
{"state": {"path": ..., "op": ..., "value": ...}} -- this rule is only
about US accounts. The store round-trips both faithfully.
Only the first was answerable in public. The predicate evaluator existed in
one place, symfonic.core.nodes.precondition_gate, private, inside a
LangGraph node the kernel does not run -- so an adopter composing an agent on
the kernel could store a regional rule and had nothing to ask whether it
applied. The example that teaches authored skills had to say so and stop.
This is that evaluator, in public, in the package that defines the DSL. The
legacy node now calls it rather than carrying its own copy: a second
interpreter of a stored contract drifts from the first the day either
changes, and the drift is silent because both keep answering.
No turn state in the reason. PredicateOutcome.reason says how the
state compared and never what it held, because the legacy node writes it to a
DEBUG log -- so echoing the value put an entitlement or an account id into
application logs by default.
Fail closed. A malformed predicate -- no path, an unknown operator, a
missing value, an in whose value is a bare string that would silently
substring-match -- does not match. A rule nobody can evaluate is a rule that
does not admit, because the alternative is a filter that widens whenever
someone mistypes it.
PredicateOutcome
dataclass
PredicateOutcome(matched: bool, reason: str)
Whether the predicate matched, and a line saying why.
reason names the path, the operator, the value the rule declared,
and how the turn's state compared -- never what the state held.
That last exclusion is a correction. The first version echoed the
resolved value, reasoning that a decision reason is not an audit record.
It is worse than one: filter_skills_by_state_predicates writes this
string to a DEBUG log on the legacy path, so an entitlement, an account
id or anything else a deployment puts in turn state was landing in
application logs by default. "region eq 'US' -> False (state: differs)"
is as actionable as naming the value and names nothing that belongs to
the turn.
evaluate_state_predicate
evaluate_state_predicate(state: Any, predicate: Any) -> PredicateOutcome
Answer one {"path": ..., "op": ..., "value": ...} against state.
Source code in src/symfonic/memory/layers/procedural/predicates.py
| def evaluate_state_predicate(state: Any, predicate: Any) -> PredicateOutcome:
"""Answer one ``{"path": ..., "op": ..., "value": ...}`` against ``state``."""
if not isinstance(predicate, dict):
return PredicateOutcome(False, "non-dict predicate")
path, op = predicate.get("path"), predicate.get("op")
if not isinstance(path, str) or not path:
return PredicateOutcome(False, "missing or empty `path`")
if not isinstance(op, str) or op not in ALLOWED_OPS:
return PredicateOutcome(
False, f"unknown operator `{op!r}` (allowed: {ALLOWED_OPS})"
)
resolved = resolve_path(state, path)
if op == "exists":
matched = resolved is not MISSING
return PredicateOutcome(
matched, f"`{path}` exists -> {matched} ({_describe(resolved)})"
)
if "value" not in predicate:
return PredicateOutcome(False, f"op `{op}` requires `value`")
expected = predicate["value"]
if resolved is MISSING:
# An absent key fails every value-taking operator. "absent OR equals
# X" is two predicates, written as two, rather than a special case
# here that every rule then inherits.
return PredicateOutcome(
False, f"`{path}` {op} `{expected!r}` -> False (state: absent)"
)
if op == "eq":
matched = resolved == expected
elif op == "neq":
matched = resolved != expected
else: # "in"
if not isinstance(expected, list | tuple | set):
# A string here would substring-match, so "region in 'US'" would
# quietly admit "U". Refused rather than narrowed.
return PredicateOutcome(
False,
f"`in` requires list/tuple/set value, got {type(expected).__name__}",
)
try:
matched = resolved in expected
except TypeError: # unhashable or unorderable -> closed
matched = False
return PredicateOutcome(
matched,
f"`{path}` {op} `{expected!r}` -> {matched} "
f"(state: {'matches' if matched else 'present, differs'})",
)
|
resolve_path
resolve_path(state: Any, dotted_path: str) -> Any
Walk state along a dotted path, trying UPPER then lower per segment.
Both casings because two conventions meet here: prompt-render state
arrives upper-snake and LangGraph runtime state arrives snake_case, and a
rule written against one should not silently stop matching under the
other. Attribute access is tried when a segment lands on something that
is not a dict, so dataclass and Pydantic state shapes resolve too.
Returns :data:MISSING -- never None -- when any segment fails.
Source code in src/symfonic/memory/layers/procedural/predicates.py
| def resolve_path(state: Any, dotted_path: str) -> Any:
"""Walk ``state`` along a dotted path, trying UPPER then lower per segment.
Both casings because two conventions meet here: prompt-render state
arrives upper-snake and LangGraph runtime state arrives snake_case, and a
rule written against one should not silently stop matching under the
other. Attribute access is tried when a segment lands on something that
is not a dict, so dataclass and Pydantic state shapes resolve too.
Returns :data:`MISSING` -- never ``None`` -- when any segment fails.
"""
if not isinstance(dotted_path, str) or not dotted_path:
return MISSING
cursor: Any = state
for segment in dotted_path.split("."):
if not segment: # a leading, trailing or doubled dot
return MISSING
upper, lower = segment.upper(), segment.lower()
# ``Mapping``, not ``dict``. The frozen per-turn snapshot a rule is
# handed is a ``mappingproxy``, and ``isinstance(proxy, dict)`` is
# False -- so a resolver written against ``dict`` walked past every
# real turn's state into the attribute branch and reported the path
# missing. Every predicate then failed closed, and a scoped rule
# silently stopped applying anywhere: green unit tests over plain
# dicts, and an example whose refusal turn quietly stopped refusing.
if isinstance(cursor, Mapping):
if upper in cursor:
cursor = cursor[upper]
elif lower in cursor:
cursor = cursor[lower]
else:
return MISSING
continue
if hasattr(cursor, upper):
cursor = getattr(cursor, upper)
elif hasattr(cursor, lower):
cursor = getattr(cursor, lower)
else:
return MISSING
return cursor
|
skill_applies
skill_applies(state: Any, precondition: Any) -> bool
Whether every state predicate on a skill matches state.
A skill with no state predicates applies everywhere -- that is what
writing none means, and it keeps the bare-identifier form untouched by
this module.
Source code in src/symfonic/memory/layers/procedural/predicates.py
| def skill_applies(state: Any, precondition: Any) -> bool:
"""Whether every state predicate on a skill matches ``state``.
A skill with no state predicates applies everywhere -- that is what
writing none means, and it keeps the bare-identifier form untouched by
this module.
"""
return all(
evaluate_state_predicate(state, predicate).matched
for predicate in state_predicates_in(precondition)
)
|
state_predicates_in
state_predicates_in(precondition: Any) -> tuple[dict[str, Any], ...]
The state predicates in a precondition list, unwrapped from state.
The other half of the list -- bare tool identifiers -- is a different
question with a different answer (has this tool been called yet?), so the
two are partitioned rather than evaluated together.
Source code in src/symfonic/memory/layers/procedural/predicates.py
| def state_predicates_in(precondition: Any) -> tuple[dict[str, Any], ...]:
"""The state predicates in a precondition list, unwrapped from ``state``.
The other half of the list -- bare tool identifiers -- is a different
question with a different answer (has this tool been called yet?), so the
two are partitioned rather than evaluated together.
"""
if not isinstance(precondition, list):
return ()
return tuple(
item["state"]
for item in precondition
if isinstance(item, dict) and isinstance(item.get("state"), dict)
)
|