From 2c539f0fdf92ff981d97cc905cbb28cdd9366436 Mon Sep 17 00:00:00 2001 From: Pengfei Hu Date: Tue, 8 Sep 2026 21:55:24 -0700 Subject: [PATCH 1/2] Compare closed Boolean source functions and literal Agent membership (#557) --- docs/baseline.md | 42 +++ docs/report-schema.v0.43.json | 186 ++++++++++++ src/agents_shipgate/cli/scan/sanitization.py | 6 + .../core/guard_dependencies.py | 91 ++++++ .../inputs/sdk_boolean_source.py | 262 ++++++++++++++++ .../inputs/sdk_guard_dependencies.py | 5 + src/agents_shipgate/report/markdown.py | 8 + .../schemas/guard_dependencies.py | 43 +++ tests/test_guard_dependency_verification.py | 63 ++++ tests/test_sdk_boolean_source.py | 283 ++++++++++++++++++ 10 files changed, 989 insertions(+) create mode 100644 src/agents_shipgate/inputs/sdk_boolean_source.py create mode 100644 tests/test_sdk_boolean_source.py diff --git a/docs/baseline.md b/docs/baseline.md index eff3cb54..2df85b08 100644 --- a/docs/baseline.md +++ b/docs/baseline.md @@ -193,6 +193,48 @@ input, not authenticated proof of its claimed source evidence. Redacted or ambiguous records cannot yield a predicate direction. The Markdown report shows up to eight relevant comparisons; JSON retains the full list. +An observed guard can also carry `source_behavior`, from the narrower +`sdk_boolean_function/v1` profile. This inspects the **entire** selected tool +function, including statements after a return and both arms of every branch, +and a single source-level `Agent(name="…", tools=[tool])` (or `tools=[]`). +Its module contains only the exact SDK imports, the one sibling guard import, +the decorated function and that literal Agent assignment, in definition order. +The guard module contains only the selected pure Boolean function. Every +function parameter is Boolean; expressions are Boolean literals, parameters, +`not`, `and`, `or`, and the already-resolved guard call. Statements are `if` +and `return`; an implicit return is recorded as `none`. There are no opaque +operation leaves. External calls, free/configuration variables, mutation, +additional functions or Agent configuration, handoffs and unresolved imports +leave the source model unresolved, even in apparently unreachable code. + +`source_behavior.returns[i]` is `true`, `false` or `none` for input mask `i` +over **all** sorted tool parameters, not just the parameters used by the first +guard. `configuration_reads: none` means the closed model admitted no such +read; it is absent as a claim when the model is unresolved. The literal +`binding` records the source Agent symbol/name and whether its tools list +contains the selected function. This is source membership, not a claim that +the Agent is a deployed entry point or that the installed SDK executes it. + +The comparison keeps four axes separate: all return values (`returns`), the +set of inputs returning true (`true_domain`), literal tool membership +(`binding`), and the true-return inputs reachable through that membership +(`bound_true_domain`). A narrowed first guard can still have a widened whole +function true domain; unchanged returns can accompany added membership. +Changing `False` to `None` changes returns even when the true domain is equal. +A changed Agent identity leaves the bound relation unresolved. Missing old +models, ambiguous capability identities and redacted evidence never establish +equality. The model and its inputs are carried in each side's existing guard +record and verifier-bound artifacts; it introduces no separate authority. + +**A true return is not approval.** The model does not prove a caller-controlled +Boolean came from an approver, establish an action's effect/authority or +attribute an existing policy predicate to a change. The outer capability +dependency coverage remains `incomplete` and finding exclusion remains false. +Neither a narrowed true domain nor equal source models is a safe verdict. +Joining complete source evidence to actual capability/policy dependencies and +the fixed-history evaluation remains work in #557/#515/#563. This SDK profile +does not cover #515's TypeScript MongoDB acceptance case. + Verification binds the reader's captured dependency bytes and named absent import candidates in `verification-plan.json` under `inputs.options.dependency_inputs`. A current-control read checks diff --git a/docs/report-schema.v0.43.json b/docs/report-schema.v0.43.json index 1c5327df..ebbc1fcc 100644 --- a/docs/report-schema.v0.43.json +++ b/docs/report-schema.v0.43.json @@ -1872,6 +1872,170 @@ "title": "BindingSurfaceDiff", "type": "object" }, + "BooleanSourceBehavior": { + "additionalProperties": false, + "description": "A closed source-function model, never deployed behavior or policy evidence.", + "properties": { + "binding": { + "anyOf": [ + { + "$ref": "#/$defs/BooleanSourceBinding" + }, + { + "type": "null" + } + ], + "default": null + }, + "configuration_reads": { + "anyOf": [ + { + "const": "none", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Configuration Reads" + }, + "parameters": { + "items": { + "type": "string" + }, + "title": "Parameters", + "type": "array" + }, + "reader_profile": { + "const": "sdk_boolean_function/v1", + "default": "sdk_boolean_function/v1", + "title": "Reader Profile", + "type": "string" + }, + "reason": { + "title": "Reason", + "type": "string" + }, + "returns": { + "items": { + "enum": [ + "true", + "false", + "none" + ], + "type": "string" + }, + "title": "Returns", + "type": "array" + }, + "status": { + "enum": [ + "observed", + "unresolved", + "redacted" + ], + "title": "Status", + "type": "string" + } + }, + "required": [ + "status", + "reason" + ], + "title": "BooleanSourceBehavior", + "type": "object" + }, + "BooleanSourceBinding": { + "additionalProperties": false, + "properties": { + "agent_name": { + "title": "Agent Name", + "type": "string" + }, + "agent_symbol": { + "title": "Agent Symbol", + "type": "string" + }, + "tool_bound": { + "title": "Tool Bound", + "type": "boolean" + } + }, + "required": [ + "agent_symbol", + "agent_name", + "tool_bound" + ], + "title": "BooleanSourceBinding", + "type": "object" + }, + "BooleanSourceComparison": { + "additionalProperties": false, + "properties": { + "binding": { + "default": "unresolved", + "enum": [ + "unchanged", + "added", + "removed", + "changed", + "unresolved" + ], + "title": "Binding", + "type": "string" + }, + "bound_true_domain": { + "default": "unresolved", + "enum": [ + "unchanged", + "widened", + "narrowed", + "changed", + "unresolved" + ], + "title": "Bound True Domain", + "type": "string" + }, + "finding_exclusion_eligible": { + "const": false, + "default": false, + "title": "Finding Exclusion Eligible", + "type": "boolean" + }, + "reason": { + "title": "Reason", + "type": "string" + }, + "returns": { + "default": "unresolved", + "enum": [ + "unchanged", + "changed", + "unresolved" + ], + "title": "Returns", + "type": "string" + }, + "true_domain": { + "default": "unresolved", + "enum": [ + "unchanged", + "widened", + "narrowed", + "changed", + "unresolved" + ], + "title": "True Domain", + "type": "string" + } + }, + "required": [ + "reason" + ], + "title": "BooleanSourceComparison", + "type": "object" + }, "CapabilityChangeBlock": { "additionalProperties": false, "description": "The diff-derived capability delta, grouped by direction.\n\nReviewer-facing projection over ``action_surface_diff`` /\n``tool_surface_diff`` (roadmap \u00a77.1). Four member lists \u2014\n``added`` / ``removed`` / ``broadened`` / ``narrowed`` \u2014 plus\n``enabled`` (mirrors the surface-diff enabled flag: ``False`` when no\nbase is available, so the block is a stable empty shape rather than\nabsent). Never gates on its own.", @@ -4870,6 +5034,17 @@ "title": "Reason", "type": "string" }, + "source_behavior": { + "anyOf": [ + { + "$ref": "#/$defs/BooleanSourceComparison" + }, + { + "type": "null" + } + ], + "default": null + }, "tool_id": { "anyOf": [ { @@ -5003,6 +5178,17 @@ "title": "Reason", "type": "string" }, + "source_behavior": { + "anyOf": [ + { + "$ref": "#/$defs/BooleanSourceBehavior" + }, + { + "type": "null" + } + ], + "default": null + }, "source_id": { "title": "Source Id", "type": "string" diff --git a/src/agents_shipgate/cli/scan/sanitization.py b/src/agents_shipgate/cli/scan/sanitization.py index 8eacf537..2623c341 100644 --- a/src/agents_shipgate/cli/scan/sanitization.py +++ b/src/agents_shipgate/cli/scan/sanitization.py @@ -629,6 +629,12 @@ def _public_tool_surfaces( public.status = "redacted" public.reason = "guard_evidence_redacted" public.allowed_inputs = [] + if public.source_behavior is not None: + public.source_behavior.status = "redacted" + public.source_behavior.reason = "source_behavior_evidence_redacted" + public.source_behavior.returns = [] + public.source_behavior.binding = None + public.source_behavior.configuration_reads = None if diffs.diff_reference_error: public_tool_surface_diff = disabled_tool_surface_diff( redact_data( diff --git a/src/agents_shipgate/core/guard_dependencies.py b/src/agents_shipgate/core/guard_dependencies.py index 6958aa97..1226ca1a 100644 --- a/src/agents_shipgate/core/guard_dependencies.py +++ b/src/agents_shipgate/core/guard_dependencies.py @@ -6,6 +6,7 @@ from agents_shipgate.core.domain import LoadedToolSource, Tool from agents_shipgate.schemas.guard_dependencies import ( + BooleanSourceComparison, GuardDependencyComparison, GuardDependencyEvidence, ) @@ -60,6 +61,91 @@ def _valid_predicate(row: GuardDependencyEvidence) -> bool: ) +def _valid_source(row: GuardDependencyEvidence) -> bool: + source = row.source_behavior + return bool( + _valid_predicate(row) + and source is not None + and source.status == "observed" + and source.configuration_reads == "none" + and source.binding is not None + and type(source.binding.tool_bound) is bool + and len(source.parameters) <= 8 + and source.parameters == sorted(set(source.parameters)) + and set(row.parameters) <= set(source.parameters) + and len(source.returns) == 1 << len(source.parameters) + and all(value in {"true", "false", "none"} for value in source.returns) + ) + + +def _domain_direction(before: set[int], after: set[int]) -> str: + return ( + "unchanged" + if before == after + else "widened" + if before < after + else "narrowed" + if after < before + else "changed" + ) + + +def _compare_source( + old: GuardDependencyEvidence | None, new: GuardDependencyEvidence | None +) -> BooleanSourceComparison: + result = BooleanSourceComparison(reason="base_or_head_source_model_unavailable") + if old is None or new is None or not _valid_source(old) or not _valid_source(new): + return result + before, after = old.source_behavior, new.source_behavior + if ( + old.tool_id, + old.source_id, + old.tool_path, + old.tool_symbol, + old.guard_path, + old.guard_symbol, + before.parameters, + ) != ( + new.tool_id, + new.source_id, + new.tool_path, + new.tool_symbol, + new.guard_path, + new.guard_symbol, + after.parameters, + ): + result.reason = "source_subject_or_parameter_domain_changed" + return result + result.returns = "unchanged" if before.returns == after.returns else "changed" + previous = {i for i, value in enumerate(before.returns) if value == "true"} + following = {i for i, value in enumerate(after.returns) if value == "true"} + result.true_domain = _domain_direction(previous, following) + if (before.binding.agent_symbol, before.binding.agent_name) != ( + after.binding.agent_symbol, + after.binding.agent_name, + ): + result.binding = "changed" + result.reason = "agent_source_identity_changed; bound relation is unresolved" + return result + result.binding = ( + "unchanged" + if before.binding.tool_bound == after.binding.tool_bound + else "added" + if after.binding.tool_bound + else "removed" + ) + result.bound_true_domain = _domain_direction( + previous if before.binding.tool_bound else set(), + following if after.binding.tool_bound else set(), + ) + result.reason = ( + "Compare the closed Boolean source function and its literal Agent tool membership only. " + "A true return is not approval, authority or an action effect; deployed wiring and " + "finding-predicate attribution remain unproved. No finding is excluded." + ) + return result + + def compare_guard_dependencies( current: list[GuardDependencyEvidence], base: list[GuardDependencyEvidence] ) -> list[GuardDependencyComparison]: @@ -121,6 +207,11 @@ def compare_guard_dependencies( reason=reason, before=old, after=new, + source_behavior=( + _compare_source(old, new) + if any(row.source_behavior is not None for row in old_rows + new_rows) + else None + ), ) ) return comparisons diff --git a/src/agents_shipgate/inputs/sdk_boolean_source.py b/src/agents_shipgate/inputs/sdk_boolean_source.py new file mode 100644 index 00000000..87a83a21 --- /dev/null +++ b/src/agents_shipgate/inputs/sdk_boolean_source.py @@ -0,0 +1,262 @@ +"""Finite Boolean source relations over fully inspected function and binding ASTs. + +This module interprets a private Boolean model, not Python/application code. +The caller has already captured the imported predicate and package bytes. No +opaque call, free/configuration variable or alternate binding is a model leaf. +A true return is a value, never evidence of approval, authority or an effect. +""" + +from __future__ import annotations + +import ast +from collections.abc import Callable + +from agents_shipgate.schemas.guard_dependencies import BooleanSourceBehavior, BooleanSourceBinding + +Value = Callable[[dict[str, bool]], bool] +Step = Callable[[dict[str, bool]], tuple[bool, bool | None]] + + +class _Unsupported(ValueError): + pass + + +def _body(node): + body = node.body + if ( + body + and isinstance(body[0], ast.Expr) + and isinstance(body[0].value, ast.Constant) + and isinstance(body[0].value.value, str) + ): + return body[1:] + return body + + +def _parameters(node: ast.FunctionDef) -> list[str]: + args = node.args + if ( + args.posonlyargs + or args.kwonlyargs + or args.vararg + or args.kwarg + or args.defaults + or args.kw_defaults + or getattr(node, "type_params", ()) + ): + raise _Unsupported("function_signature_not_closed") + if not ( + node.returns is None or isinstance(node.returns, ast.Name) and node.returns.id == "bool" + ): + raise _Unsupported("function_annotation_not_boolean") + if any( + not isinstance(arg.annotation, ast.Name) or arg.annotation.id != "bool" for arg in args.args + ): + raise _Unsupported("function_parameters_not_boolean") + names = [arg.arg for arg in args.args] + if len(names) > 8 or len(names) != len(set(names)) or "bool" in names: + raise _Unsupported("function_parameter_domain_not_supported") + return names + + +def _expression( + node: ast.expr, + names: set[str], + guard_name: str | None, + guard: tuple[list[str], Value] | None, + depth: int = 0, +) -> Value: + if depth > 24: + raise _Unsupported("source_model_depth_limit") + if isinstance(node, ast.Constant) and type(node.value) is bool: + return lambda values: node.value + if isinstance(node, ast.Name) and node.id in names: + return lambda values: values[node.id] + if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.Not): + operand = _expression(node.operand, names, guard_name, guard, depth + 1) + return lambda values: not operand(values) + if isinstance(node, ast.BoolOp) and isinstance(node.op, (ast.And, ast.Or)): + # Validate every operand before evaluating any input, including those + # behind a literal short circuit or an unreachable-looking return. + operands = [_expression(item, names, guard_name, guard, depth + 1) for item in node.values] + operation = all if isinstance(node.op, ast.And) else any + return lambda values: operation(operand(values) for operand in operands) + if ( + guard is not None + and isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id == guard_name + and not node.keywords + and len(node.args) == len(guard[0]) + ): + arguments = [_expression(arg, names, None, None, depth + 1) for arg in node.args] + return lambda values: guard[1]( + {name: arg(values) for name, arg in zip(guard[0], arguments, strict=True)} + ) + raise _Unsupported("function_expression_dependency_not_supported") + + +def _sequence( + body: list[ast.stmt], + names: set[str], + guard_name: str, + guard: tuple[list[str], Value], + depth: int = 0, +) -> Step: + if depth > 24: + raise _Unsupported("source_model_depth_limit") + steps: list[Step] = [] + for statement in body: + if isinstance(statement, ast.Return): + if ( + statement.value is None + or isinstance(statement.value, ast.Constant) + and statement.value.value is None + ): + steps.append(lambda values: (True, None)) + else: + value = _expression(statement.value, names, guard_name, guard) + steps.append(lambda values, value=value: (True, value(values))) + elif isinstance(statement, ast.If): + condition = _expression(statement.test, names, guard_name, guard) + yes = _sequence(statement.body, names, guard_name, guard, depth + 1) + no = _sequence(statement.orelse, names, guard_name, guard, depth + 1) + steps.append( + lambda values, condition=condition, yes=yes, no=no: ( + yes if condition(values) else no + )(values) + ) + else: + raise _Unsupported("function_statement_dependency_not_supported") + + def run(values: dict[str, bool]) -> tuple[bool, bool | None]: + for step in steps: + returned, value = step(values) + if returned: + return True, value + return False, None + + return run + + +def _binding( + tree: ast.Module, + definition: ast.FunctionDef, + guard_name: str, + guard_symbol: str, + guard_module: str, +) -> BooleanSourceBinding: + body = _body(tree) + if len(body) < 3 or body[-2] is not definition: + raise _Unsupported("tool_module_not_closed") + imported: list[str] = [] + guard_imports = 0 + for statement in body[:-2]: + if not isinstance(statement, ast.ImportFrom): + raise _Unsupported("tool_module_not_closed") + if statement.module == "agents" and statement.level == 0: + if any( + alias.asname or alias.name not in {"Agent", "function_tool"} + for alias in statement.names + ): + raise _Unsupported("sdk_binding_import_not_exact") + imported.extend(alias.name for alias in statement.names) + elif ( + statement.module == guard_module + and statement.level == 1 + and len(statement.names) == 1 + and statement.names[0].name == guard_symbol + and (statement.names[0].asname or guard_symbol) == guard_name + ): + guard_imports += 1 + else: + raise _Unsupported("tool_module_import_dependency_not_supported") + if sorted(imported) != ["Agent", "function_tool"] or guard_imports != 1: + raise _Unsupported("sdk_binding_import_not_exact") + reserved = {"Agent", "function_tool", "bool", guard_name, definition.name} + if len(reserved) != 5 or any(arg.arg in reserved for arg in definition.args.args): + raise _Unsupported("source_binding_identity_not_unique") + if ( + len(definition.decorator_list) != 1 + or not isinstance(definition.decorator_list[0], ast.Name) + or definition.decorator_list[0].id != "function_tool" + ): + raise _Unsupported("tool_decorator_not_exact") + assignment = body[-1] + if ( + not isinstance(assignment, ast.Assign) + or len(assignment.targets) != 1 + or not isinstance(assignment.targets[0], ast.Name) + or assignment.targets[0].id in reserved + ): + raise _Unsupported("agent_binding_not_unique_literal") + call = assignment.value + if ( + not isinstance(call, ast.Call) + or not isinstance(call.func, ast.Name) + or call.func.id != "Agent" + or call.args + or len(call.keywords) != 2 + or sorted(keyword.arg or "" for keyword in call.keywords) != ["name", "tools"] + ): + raise _Unsupported("agent_configuration_not_closed") + keywords = {keyword.arg: keyword.value for keyword in call.keywords} + name, tools = keywords["name"], keywords["tools"] + if not isinstance(name, ast.Constant) or not isinstance(name.value, str): + raise _Unsupported("agent_configuration_not_closed") + if ( + not isinstance(tools, ast.List) + or len(tools.elts) > 1 + or any(not isinstance(tool, ast.Name) or tool.id != definition.name for tool in tools.elts) + ): + raise _Unsupported("agent_tool_binding_not_literal") + return BooleanSourceBinding( + agent_symbol=assignment.targets[0].id, agent_name=name.value, tool_bound=bool(tools.elts) + ) + + +def read_boolean_source_behavior( + *, + tree: ast.Module, + definition: ast.FunctionDef | ast.AsyncFunctionDef, + guard_tree: ast.Module, + target: ast.FunctionDef, + guard_name: str, + guard_module: str, +) -> BooleanSourceBehavior: + """Model every statement and input, or publish one explicit refusal reason.""" + result = BooleanSourceBehavior(status="unresolved", reason="source_model_unavailable") + try: + if not isinstance(definition, ast.FunctionDef) or _body(guard_tree) != [target]: + raise _Unsupported("source_function_or_helper_module_not_closed") + parameters = _parameters(definition) + guard_parameters = _parameters(target) + if target.decorator_list: + raise _Unsupported("helper_decorator_not_supported") + helper = _body(target) + if len(helper) != 1 or not isinstance(helper[0], ast.Return) or helper[0].value is None: + raise _Unsupported("helper_function_not_closed") + predicate = _expression(helper[0].value, set(guard_parameters), None, None) + binding = _binding(tree, definition, guard_name, target.name, guard_module) + function = _sequence( + _body(definition), set(parameters), guard_name, (guard_parameters, predicate) + ) + parameters = sorted(parameters) + returns = [] + for mask in range(1 << len(parameters)): + values = {name: bool(mask & (1 << i)) for i, name in enumerate(parameters)} + _, value = function(values) + returns.append("none" if value is None else "true" if value else "false") + result = BooleanSourceBehavior( + status="observed", + reason="closed_boolean_source_function_and_literal_agent_binding", + parameters=parameters, + returns=returns, + binding=binding, + configuration_reads="none", + ) + except _Unsupported as exc: + result.reason = str(exc) + except RecursionError: + result.reason = "source_model_depth_limit" + return result diff --git a/src/agents_shipgate/inputs/sdk_guard_dependencies.py b/src/agents_shipgate/inputs/sdk_guard_dependencies.py index 423b724a..fc339715 100644 --- a/src/agents_shipgate/inputs/sdk_guard_dependencies.py +++ b/src/agents_shipgate/inputs/sdk_guard_dependencies.py @@ -19,6 +19,7 @@ ) from agents_shipgate.core.tool_identity import source_observation_id from agents_shipgate.core.trust_roots import IdentityBoundReadSession +from agents_shipgate.inputs.sdk_boolean_source import read_boolean_source_behavior from agents_shipgate.schemas.guard_dependencies import ( GuardDependencyEvidence, GuardInputEvidence, @@ -390,6 +391,10 @@ def read_guard_dependency( evidence.allowed_inputs = allowed evidence.status = "observed" evidence.reason = "bounded_source_predicate_only" + evidence.source_behavior = read_boolean_source_behavior( + tree=tree, definition=definition, guard_tree=guard_tree, target=target, + guard_name=call.func.id, guard_module=str(imported.module), + ) except _Unresolved as exc: evidence.reason = str(exc) evidence.status = "ambiguous" if "ambiguous" in evidence.reason else "unresolved" diff --git a/src/agents_shipgate/report/markdown.py b/src/agents_shipgate/report/markdown.py index 5211629f..ed1c5873 100644 --- a/src/agents_shipgate/report/markdown.py +++ b/src/agents_shipgate/report/markdown.py @@ -816,6 +816,14 @@ def _append_tool_surface_diff(lines: list[str], report: ReadinessReport) -> None def location(row): return f"{row.guard_path}:{row.guard_line}" if row and row.guard_path else "unavailable" lines.append(f"- {_safe_markdown_text(comparison.tool_name)}: {comparison.direction.replace('_', ' ')}; base `{_safe_markdown_text(location(before))}`, head `{_safe_markdown_text(location(after))}`.") + if comparison.source_behavior is not None: + source = comparison.source_behavior + lines.append( + f" Full Boolean source: returns {source.returns}; true-return inputs " + f"{source.true_domain}; literal Agent membership {source.binding}; " + f"bound true-return inputs {source.bound_true_domain}. " + "True is a return value, not approval or action authority." + ) if len(visible_guards) > 8: lines.append(f"{len(visible_guards) - 8} more guard comparisons in report.json.") lines.append("") diff --git a/src/agents_shipgate/schemas/guard_dependencies.py b/src/agents_shipgate/schemas/guard_dependencies.py index 0db42239..35391473 100644 --- a/src/agents_shipgate/schemas/guard_dependencies.py +++ b/src/agents_shipgate/schemas/guard_dependencies.py @@ -15,6 +15,43 @@ class GuardInputEvidence(BaseModel): role: Literal["tool_module", "guard_module", "package_initializer"] +class BooleanSourceBinding(BaseModel): + model_config = ConfigDict(extra="forbid") + + agent_symbol: str + agent_name: str + tool_bound: bool + + +class BooleanSourceBehavior(BaseModel): + """A closed source-function model, never deployed behavior or policy evidence.""" + + model_config = ConfigDict(extra="forbid") + + reader_profile: Literal["sdk_boolean_function/v1"] = "sdk_boolean_function/v1" + status: Literal["observed", "unresolved", "redacted"] + reason: str + parameters: list[str] = Field(default_factory=list) + # Entry i is the literal result for input mask i, over all tool parameters. + returns: list[Literal["true", "false", "none"]] = Field(default_factory=list) + binding: BooleanSourceBinding | None = None + configuration_reads: Literal["none"] | None = None + + +BooleanDomainDirection = Literal["unchanged", "widened", "narrowed", "changed", "unresolved"] + + +class BooleanSourceComparison(BaseModel): + model_config = ConfigDict(extra="forbid") + + returns: Literal["unchanged", "changed", "unresolved"] = "unresolved" + true_domain: BooleanDomainDirection = "unresolved" + binding: Literal["unchanged", "added", "removed", "changed", "unresolved"] = "unresolved" + bound_true_domain: BooleanDomainDirection = "unresolved" + reason: str + finding_exclusion_eligible: Literal[False] = False + + class GuardDependencyEvidence(BaseModel): model_config = ConfigDict(extra="forbid") @@ -38,6 +75,9 @@ class GuardDependencyEvidence(BaseModel): allowed_inputs: list[int] = Field(default_factory=list) inputs: list[GuardInputEvidence] = Field(default_factory=list) absent_paths: list[str] = Field(default_factory=list) + source_behavior: BooleanSourceBehavior | None = Field( + default=None, exclude_if=lambda value: value is None + ) dependency_coverage: Literal["incomplete"] = "incomplete" finding_exclusion_eligible: Literal[False] = False @@ -58,4 +98,7 @@ class GuardDependencyComparison(BaseModel): reason: str before: GuardDependencyEvidence | None = None after: GuardDependencyEvidence | None = None + source_behavior: BooleanSourceComparison | None = Field( + default=None, exclude_if=lambda value: value is None + ) finding_exclusion_eligible: Literal[False] = False diff --git a/tests/test_guard_dependency_verification.py b/tests/test_guard_dependency_verification.py index 90fadf88..61f80917 100644 --- a/tests/test_guard_dependency_verification.py +++ b/tests/test_guard_dependency_verification.py @@ -295,3 +295,66 @@ def test_uncaptured_helper_cannot_supply_current_authority(tmp_path, kind, relat read_current_control(out, live=_live(root)) _verify(root, archive_head=False) read_current_control(out, live=_live(root)) + + +def test_paired_source_behavior_reports_continuation_without_excluding_findings(tmp_path): + root, _ = _workspace(tmp_path) + old, _ = run_scan(config_path=root / "shipgate.yaml", output_dir=tmp_path / "base", + plugins_enabled=False) + source = root / "refund_agent/agent.py" + source.write_text(AGENT.replace("return True", "return False")) + new, _ = run_scan(config_path=root / "shipgate.yaml", output_dir=tmp_path / "head", + diff_from_path=tmp_path / "base/report.json", plugins_enabled=False) + comparison = new.tool_surface_diff.guard_comparisons[0] + assert comparison.direction == "predicate_unchanged" + assert comparison.source_behavior.true_domain == "narrowed" + assert comparison.source_behavior.finding_exclusion_eligible is False + assert {f.fingerprint for f in old.findings} == {f.fingerprint for f in new.findings} + assert old.release_decision.decision == new.release_decision.decision + assert "Full Boolean source: returns changed" in (tmp_path / "head/report.md").read_text() + + +def test_actual_committed_source_model_binds_continuation_and_membership_inputs(tmp_path): + root, _ = _workspace(tmp_path) + base = _git(root, "rev-parse", "HEAD") + source = root / "refund_agent/agent.py" + source.write_text(AGENT.replace("tools=[refund]", "tools=[]")) + _git(root, "add", "refund_agent/agent.py") + _git(root, "commit", "-qm", "remove literal tool membership") + _verify(root, base=base) + out = root / "agents-shipgate-reports" + report = json.loads((out / "report.json").read_text()) + comparison = report["tool_surface_diff"]["guard_comparisons"][0] + assert comparison["direction"] == "predicate_unchanged" + assert comparison["source_behavior"]["binding"] == "removed" + assert comparison["source_behavior"]["bound_true_domain"] == "narrowed" + assert comparison["before"]["source_behavior"]["binding"]["tool_bound"] is True + assert comparison["after"]["source_behavior"]["binding"]["tool_bound"] is False + read_current_control(out, live=_live(root)) + # Source dependencies include the full continuation and binding, even if + # Git's normal overlay cannot see a local modification to that module. + _git(root, "update-index", "--assume-unchanged", "refund_agent/agent.py") + source.write_text(source.read_text().replace("return True", "return False")) + assert _git(root, "status", "--porcelain") == "" + with pytest.raises(CurrentControlUnavailable): + read_current_control(out, live=_live(root)) + + +def test_redacted_binding_cannot_retain_an_observed_source_relation(tmp_path): + root, _ = _workspace(tmp_path) + secret = "AKIAABCDEFGHIJKLMNOP" + source = root / "refund_agent/agent.py" + source.write_text(AGENT.replace('name="Refund"', f'name="{secret}"')) + old, _ = run_scan(config_path=root / "shipgate.yaml", output_dir=tmp_path / "base", + plugins_enabled=False) + row = old.tool_surface_facts.guard_dependencies[0] + assert row.status == "redacted" + assert row.source_behavior.status == "redacted" + assert row.source_behavior.returns == [] and row.source_behavior.binding is None + new, _ = run_scan(config_path=root / "shipgate.yaml", output_dir=tmp_path / "head", + diff_from_path=tmp_path / "base/report.json", plugins_enabled=False) + assert new.tool_surface_diff.guard_comparisons[0].source_behavior.returns == "unresolved" + for directory in (tmp_path / "base", tmp_path / "head"): + for path in directory.iterdir(): + if path.is_file(): + assert secret not in path.read_text(errors="replace") diff --git a/tests/test_sdk_boolean_source.py b/tests/test_sdk_boolean_source.py new file mode 100644 index 00000000..3762a991 --- /dev/null +++ b/tests/test_sdk_boolean_source.py @@ -0,0 +1,283 @@ +from __future__ import annotations + +import pytest +from test_sdk_guard_dependencies import AGENT, read_workspace, write_workspace + +from agents_shipgate.core.guard_dependencies import compare_guard_dependencies + + +def _change(root, text): + (root / "refund_agent/agent.py").write_text(text) + + +def _compare(root, manifest, before): + (after,) = read_workspace(root, manifest) + (comparison,) = compare_guard_dependencies([after], before) + assert after.dependency_coverage == "incomplete" + assert comparison.finding_exclusion_eligible is False + assert comparison.source_behavior.finding_exclusion_eligible is False + return after, comparison + + +def test_model_covers_all_parameters_continuation_and_literal_binding(tmp_path): + manifest = write_workspace(tmp_path) + (row,) = read_workspace(tmp_path, manifest) + source = row.source_behavior + assert source.status == "observed" + assert source.parameters == ["approved", "within_limit"] + assert source.returns == ["false", "false", "false", "true"] + assert source.configuration_reads == "none" + assert source.binding.model_dump() == { + "agent_symbol": "agent", + "agent_name": "Refund", + "tool_bound": True, + } + assert type(row).model_validate_json(row.model_dump_json()) == row + + +def test_narrower_guard_can_have_wider_whole_function_true_domain(tmp_path): + manifest = write_workspace(tmp_path, "True") + _change(tmp_path, AGENT.replace("return True", "return approved and within_limit")) + before = read_workspace(tmp_path, manifest) + write_workspace(tmp_path, "approved") + _, comparison = _compare(tmp_path, manifest, before) + assert comparison.direction == "predicate_narrowed" + assert comparison.source_behavior.returns == "changed" + assert comparison.source_behavior.true_domain == "widened" + assert comparison.source_behavior.bound_true_domain == "widened" + + +@pytest.mark.parametrize( + "before_bound,after_bound,direction", + [ + (False, True, "added"), + (True, False, "removed"), + (False, False, "unchanged"), + ], +) +def test_unchanged_predicate_does_not_hide_literal_reachability_change( + tmp_path, + before_bound, + after_bound, + direction, +): + manifest = write_workspace(tmp_path) + _change(tmp_path, AGENT if before_bound else AGENT.replace("tools=[refund]", "tools=[]")) + before = read_workspace(tmp_path, manifest) + _change(tmp_path, AGENT if after_bound else AGENT.replace("tools=[refund]", "tools=[]")) + _, comparison = _compare(tmp_path, manifest, before) + assert comparison.direction == "predicate_unchanged" + assert comparison.source_behavior.binding == direction + assert comparison.source_behavior.returns == "unchanged" + assert ( + comparison.source_behavior.bound_true_domain + == { + "added": "widened", + "removed": "narrowed", + "unchanged": "unchanged", + }[direction] + ) + + +@pytest.mark.parametrize( + "tail,expected", + [ + ("return within_limit", ["false", "false", "false", "true"]), + ("return False", ["false"] * 4), + ("return None", ["false", "none", "false", "none"]), + ("if within_limit:\n return True", ["false", "none", "false", "true"]), + ( + "if within_limit:\n return True\n else:\n return approved", + ["false", "true", "false", "true"], + ), + ("return not (not approved or not within_limit)", ["false", "false", "false", "true"]), + ("return allowed(within_limit, approved)", ["false", "false", "false", "true"]), + ], +) +def test_boolean_model_accounts_for_each_possible_exit(tmp_path, tail, expected): + manifest = write_workspace(tmp_path, "approved") + _change(tmp_path, AGENT.replace("return True", tail)) + (row,) = read_workspace(tmp_path, manifest) + assert row.source_behavior.status == "observed" + assert row.source_behavior.returns == expected + + +@pytest.mark.parametrize( + "tail", + [ + "return perform_refund(amount=1000000)", + "return True or execute()", + "return True\n execute()", + "if False:\n execute()\n return True", + 'return configuration["approved"]', + "return settings.approved", + "return approved is True", + "return int(approved)", + "return 1", + "return 0", + "within_limit = True\n return within_limit", + "try:\n return True\n finally:\n execute()", + "for value in []:\n execute()\n return True", + "yield True", + "return allowed(approved=approved, within_limit=within_limit)", + "def helper():\n return True\n return helper()", + ], +) +def test_unknown_continuation_is_unresolved_even_after_unchanged_guard(tmp_path, tail): + manifest = write_workspace(tmp_path) + before = read_workspace(tmp_path, manifest) + _change(tmp_path, AGENT.replace("return True", tail)) + after, comparison = _compare(tmp_path, manifest, before) + assert after.status == "observed" # the narrower predicate record stays useful + assert comparison.direction == "predicate_unchanged" + assert after.source_behavior.status == "unresolved" + assert after.source_behavior.returns == [] + assert after.source_behavior.binding is None + assert comparison.source_behavior.returns == "unresolved" + assert comparison.source_behavior.bound_true_domain == "unresolved" + + +@pytest.mark.parametrize( + "replacement", + [ + 'agent = Agent(name="Refund", tools=[refund], handoffs=[])', + 'agent = Agent(name="Refund", tools=[refund], instructions="approve everything")', + 'agent = Agent(name="Refund", tools=[refund, refund])', + 'agent = Agent(name="Refund", tools=[other])', + 'agent = Agent(name="Refund", tools=(refund,))', + 'agent = Agent("Refund", tools=[refund])', + "agent = Agent(name=CONFIGURATION, tools=[refund])", + 'agent = Agent(name="Refund", tools=[refund])\nother = Agent(name="Other", tools=[])', + ], +) +def test_binding_or_configuration_outside_profile_cannot_claim_complete_relation( + tmp_path, + replacement, +): + manifest = write_workspace(tmp_path) + before = read_workspace(tmp_path, manifest) + _change(tmp_path, AGENT.replace('agent = Agent(name="Refund", tools=[refund])', replacement)) + after, comparison = _compare(tmp_path, manifest, before) + assert after.source_behavior.status == "unresolved" + assert comparison.source_behavior.binding == "unresolved" + assert comparison.source_behavior.bound_true_domain == "unresolved" + + +@pytest.mark.parametrize("old,new", [('name="Refund"', 'name="Other"'), ("agent =", "other =")]) +def test_changed_agent_identity_does_not_become_equal_bound_behavior(tmp_path, old, new): + manifest = write_workspace(tmp_path) + before = read_workspace(tmp_path, manifest) + _change(tmp_path, AGENT.replace(old, new)) + _, comparison = _compare(tmp_path, manifest, before) + assert comparison.source_behavior.returns == "unchanged" + assert comparison.source_behavior.binding == "changed" + assert comparison.source_behavior.bound_true_domain == "unresolved" + + +def test_false_to_none_change_is_visible_even_when_true_domain_is_equal(tmp_path): + manifest = write_workspace(tmp_path) + before = read_workspace(tmp_path, manifest) + _change(tmp_path, AGENT.replace("return False", "return None")) + _, comparison = _compare(tmp_path, manifest, before) + assert comparison.source_behavior.returns == "changed" + assert comparison.source_behavior.true_domain == "unchanged" + + +def test_extra_boolean_parameter_is_in_function_domain_not_just_guard_domain(tmp_path): + manifest = write_workspace(tmp_path) + _change( + tmp_path, + AGENT.replace("within_limit: bool)", "within_limit: bool, extra: bool)").replace( + "return True", "return extra" + ), + ) + (row,) = read_workspace(tmp_path, manifest) + assert row.parameters == ["approved", "within_limit"] + assert row.source_behavior.parameters == ["approved", "extra", "within_limit"] + assert row.source_behavior.returns == ["false"] * 7 + ["true"] + + +def test_missing_old_model_duplicate_subject_and_invalid_domain_remain_unresolved(tmp_path): + manifest = write_workspace(tmp_path) + (row,) = read_workspace(tmp_path, manifest) + old = row.model_copy(update={"source_behavior": None}) + for before in ([], [old], [row, row]): + comparison = compare_guard_dependencies([row], before)[0] + assert comparison.source_behavior.returns == "unresolved" + for change in ( + {"returns": []}, + {"parameters": ["unknown"]}, + {"binding": None}, + {"configuration_reads": None}, + {"status": "redacted"}, + ): + invalid = row.model_copy( + update={"source_behavior": row.source_behavior.model_copy(update=change)} + ) + assert ( + compare_guard_dependencies([invalid], [row])[0].source_behavior.returns == "unresolved" + ) + + +def test_closed_source_is_not_an_approval_claim(tmp_path): + manifest = write_workspace(tmp_path, "approved") + (row,) = read_workspace(tmp_path, manifest) + comparison = compare_guard_dependencies([row], [row])[0] + assert comparison.source_behavior.true_domain == "unchanged" + assert "not approval" in comparison.source_behavior.reason + assert row.dependency_coverage == "incomplete" + assert row.finding_exclusion_eligible is False + + +@pytest.mark.parametrize( + "old,new", + [ + ("def refund", "async def refund"), + ("@function_tool", "@function_tool\n@function_tool"), + ("from agents import Agent, function_tool", "from agents import function_tool"), + ( + "from .guards import permitted as allowed", + "from .guards import permitted as allowed, other", + ), + ("within_limit: bool", "within_limit: str"), + ("-> bool", "-> CustomResult"), + ], +) +def test_signature_and_import_identity_must_be_closed(tmp_path, old, new): + manifest = write_workspace(tmp_path) + before = read_workspace(tmp_path, manifest) + _change(tmp_path, AGENT.replace(old, new)) + after, comparison = _compare(tmp_path, manifest, before) + assert after.source_behavior is None or after.source_behavior.status == "unresolved" + assert comparison.source_behavior.returns == "unresolved" + + +def test_imported_helper_module_cannot_hide_another_definition(tmp_path): + manifest = write_workspace(tmp_path) + before = read_workspace(tmp_path, manifest) + guard = tmp_path / "refund_agent/guards.py" + guard.write_text(guard.read_text() + "\ndef other(flag: bool):\n return external(flag)\n") + after, comparison = _compare(tmp_path, manifest, before) + assert after.status == "observed" + assert after.source_behavior.status == "unresolved" + assert comparison.source_behavior.returns == "unresolved" + + +def test_changed_parameter_domain_and_ambiguous_canonical_subject_refuse_comparison(tmp_path): + manifest = write_workspace(tmp_path) + before = read_workspace(tmp_path, manifest) + _change(tmp_path, AGENT.replace("within_limit: bool)", "within_limit: bool, extra: bool)")) + _, comparison = _compare(tmp_path, manifest, before) + assert comparison.source_behavior.returns == "unresolved" + ambiguous = before[0].model_copy(update={"tool_id": None, "status": "ambiguous"}) + assert ( + compare_guard_dependencies([ambiguous], before)[0].source_behavior.returns == "unresolved" + ) + + +def test_deep_boolean_model_refuses_without_crashing(tmp_path): + manifest = write_workspace(tmp_path) + _change(tmp_path, AGENT.replace("return True", "return " + "not " * 30 + "approved")) + (row,) = read_workspace(tmp_path, manifest) + assert row.source_behavior.status == "unresolved" + assert row.source_behavior.reason == "source_model_depth_limit" From 00fd44190dc40ce4fbd092f306a16743820025a7 Mon Sep 17 00:00:00 2001 From: Pengfei Hu Date: Tue, 8 Sep 2026 22:12:43 -0700 Subject: [PATCH 2/2] Bound helper evaluation with a validated Boolean table --- .../inputs/sdk_boolean_source.py | 20 +++++--- tests/test_sdk_boolean_source.py | 48 +++++++++++++++++++ 2 files changed, 62 insertions(+), 6 deletions(-) diff --git a/src/agents_shipgate/inputs/sdk_boolean_source.py b/src/agents_shipgate/inputs/sdk_boolean_source.py index 87a83a21..0336feee 100644 --- a/src/agents_shipgate/inputs/sdk_boolean_source.py +++ b/src/agents_shipgate/inputs/sdk_boolean_source.py @@ -15,6 +15,7 @@ Value = Callable[[dict[str, bool]], bool] Step = Callable[[dict[str, bool]], tuple[bool, bool | None]] +GuardTable = tuple[list[str], tuple[bool, ...]] class _Unsupported(ValueError): @@ -63,7 +64,7 @@ def _expression( node: ast.expr, names: set[str], guard_name: str | None, - guard: tuple[list[str], Value] | None, + guard: GuardTable | None, depth: int = 0, ) -> Value: if depth > 24: @@ -90,9 +91,9 @@ def _expression( and len(node.args) == len(guard[0]) ): arguments = [_expression(arg, names, None, None, depth + 1) for arg in node.args] - return lambda values: guard[1]( - {name: arg(values) for name, arg in zip(guard[0], arguments, strict=True)} - ) + return lambda values: guard[1][ + sum((1 << i) if argument(values) else 0 for i, argument in enumerate(arguments)) + ] raise _Unsupported("function_expression_dependency_not_supported") @@ -100,7 +101,7 @@ def _sequence( body: list[ast.stmt], names: set[str], guard_name: str, - guard: tuple[list[str], Value], + guard: GuardTable, depth: int = 0, ) -> Step: if depth > 24: @@ -238,8 +239,15 @@ def read_boolean_source_behavior( raise _Unsupported("helper_function_not_closed") predicate = _expression(helper[0].value, set(guard_parameters), None, None) binding = _binding(tree, definition, guard_name, target.name, guard_module) + # The entire helper expression was validated before any evaluation. + # Each helper input is evaluated once, independently of the number of + # tool parameters or repeated calls; AST limits must not multiply. + table = tuple( + predicate({name: bool(mask & (1 << i)) for i, name in enumerate(guard_parameters)}) + for mask in range(1 << len(guard_parameters)) + ) function = _sequence( - _body(definition), set(parameters), guard_name, (guard_parameters, predicate) + _body(definition), set(parameters), guard_name, (guard_parameters, table) ) parameters = sorted(parameters) returns = [] diff --git a/tests/test_sdk_boolean_source.py b/tests/test_sdk_boolean_source.py index 3762a991..b4c57083 100644 --- a/tests/test_sdk_boolean_source.py +++ b/tests/test_sdk_boolean_source.py @@ -281,3 +281,51 @@ def test_deep_boolean_model_refuses_without_crashing(tmp_path): (row,) = read_workspace(tmp_path, manifest) assert row.source_behavior.status == "unresolved" assert row.source_behavior.reason == "source_model_depth_limit" + + +def test_high_fanout_helper_is_evaluated_once_per_distinct_boolean_input(tmp_path, monkeypatch): + import ast + + from agents_shipgate.inputs import sdk_boolean_source, sdk_guard_dependencies + + manifest = write_workspace(tmp_path) + parameters = ", ".join(f"p{i}: bool" for i in range(8)) + source = ( + "from agents import Agent, function_tool\n" + "from .guards import permitted as allowed\n\n" + f"@function_tool\ndef refund({parameters}) -> bool:\n" + " if not allowed(p0):\n return False\n" + " return " + " and ".join(["allowed(p0)"] * 790) + "\n\n" + 'agent = Agent(name="Refund", tools=[refund])\n' + ) + helper = "def permitted(p0: bool) -> bool:\n return " + " and ".join(["True"] * 4000) + "\n" + for text in (source, helper): + assert sdk_guard_dependencies.guard_module_metadata(ast.parse(text), text)[1] + _change(tmp_path, source) + (tmp_path / "refund_agent/guards.py").write_text(helper) + original = sdk_boolean_source._expression + evaluations = 0 + + def count_helper(node, *args, **kwargs): + expression = original(node, *args, **kwargs) + if isinstance(node, ast.BoolOp) and len(node.values) == 4000: + + def evaluate(values): + nonlocal evaluations + evaluations += 1 + # Structural validation above still visits every real operand. + # This all-True helper is constant: count calls without doing + # the old implementation's 809,984,000 leaf evaluations. + return True + + return evaluate + return expression + + monkeypatch.setattr(sdk_boolean_source, "_expression", count_helper) + (row,) = read_workspace(tmp_path, manifest) + assert row.source_behavior.status == "observed" + assert row.source_behavior.returns == ["true"] * 256 + assert evaluations == 2 + monkeypatch.setattr(sdk_boolean_source, "_expression", original) + (actual,) = read_workspace(tmp_path, manifest) + assert actual == row