// @harn-entrypoint-category llm.stdlib
//
// std/llm/structural_validator - deterministic pre-dispatch turn checks.
//
// Opt in via `agent_loop(harness, ..., {structural_validator: with_structural_validator({...})})`.
// The validator intercepts the agent loop's internal
// `__structural_validator_turn__` pre-dispatch probe, emits a
// `structural_validator_decision` event, and either regenerates with
// feedback or raises.
import { agent_emit_event } from "std/agent/state"
import { protocol_violation_messages } from "std/llm/tool_parse_result"
import "std/schema"
fn __sv_tool_name() -> string {
return "__structural_validator_turn__"
}
fn __sv_dict(value) -> dict {
if type_of(value) == "dict" {
return value
}
return {}
}
fn __sv_list(value) -> list {
if type_of(value) == "list" {
return value
}
return []
}
fn __sv_string(value) -> string {
if value == nil {
return ""
}
return to_string(value)
}
fn __sv_string_list(value) -> list {
let out = []
for item in __sv_list(value) {
const text = trim(__sv_string(item))
if text != "" {
out = out.appending(text)
}
}
return out
}
fn __sv_default_no_phantom_completion_catalog() {
return {
en: [
"i fixed",
"fixed the",
"updated the",
"implemented the",
"i implemented",
"i completed",
"completed the",
"successfully",
"all set",
"done",
],
}
}
fn __sv_merge_catalog_locale(base, override_items) {
let merged = []
for item in __sv_string_list(base) + __sv_string_list(override_items) {
const normalized = lowercase(item)
if normalized != "" && !contains(merged, normalized) {
merged = merged.appending(normalized)
}
}
return merged
}
fn __sv_normalize_no_phantom_completion_catalog(value) {
const defaults = __sv_default_no_phantom_completion_catalog()
if value == nil {
return defaults
}
let raw = __sv_dict(value)
if len(raw.keys()) == 0 && type_of(value) != "dict" {
throw "with_structural_validator: `no_phantom_completion_catalog` must be a dict"
}
let out = defaults
for locale in raw.keys() {
out = out + {[locale]: __sv_merge_catalog_locale(out[locale] ?? [], raw[locale])}
}
return out
}
fn __sv_locale(value) -> string {
const locale = trim(__sv_string(value))
if locale == "" {
return "en"
}
return locale
}
fn __sv_rule_names() {
return [
"non_empty_when_writes_expected",
"no_phantom_completion",
"tool_calls_well_formed",
"output_token_cap_with_zero_calls",
]
}
fn __sv_normalize_rule_entry(value) {
if type_of(value) == "string" {
const name = trim(__sv_string(value))
if name == "" {
throw "with_structural_validator: rule names must be non-empty strings"
}
return {name: name, warn_only: false}
}
const entry = __sv_dict(value)
const name = trim(__sv_string(entry?.name ?? entry?.rule))
if name == "" {
throw "with_structural_validator: rule entries need `name`"
}
return {name: name, warn_only: entry?.warn_only ?? false}
}
fn __sv_normalize_rules(value) {
if value == nil {
return [__sv_normalize_rule_entry("non_empty_when_writes_expected")]
}
let raw = __sv_list(value)
if len(raw) == 0 {
return [__sv_normalize_rule_entry("non_empty_when_writes_expected")]
}
let rules = []
for item in raw {
const entry = __sv_normalize_rule_entry(item)
const name = entry.name
if !contains(rules.map({ rule -> rule.name }), name) {
rules = rules.appending(entry)
}
}
if len(rules) == 0 {
return [__sv_normalize_rule_entry("non_empty_when_writes_expected")]
}
return rules
}
fn __sv_rule_entry_schema() {
return schema_union(
[
schema_string() + {min_length: 1},
schema_object(
{
name: schema_field(schema_string() + {min_length: 1}, false),
rule: schema_field(schema_string() + {min_length: 1}, false),
warn_only: schema_default(schema_bool(), false),
},
{additional_properties: true},
),
],
)
}
fn __sv_options_schema() {
return schema_object(
{
on_failure: schema_default(
schema_enum(["regenerate_with_feedback", "raise"]),
"regenerate_with_feedback",
),
max_attempts: schema_default(schema_int() + {min: 1}, 3),
locale: schema_default(schema_string(), "en"),
no_phantom_completion_catalog: schema_field(schema_dict(schema_list(schema_string())), false),
rules: schema_field(schema_list(__sv_rule_entry_schema()), false),
},
{additional_properties: true},
)
}
fn __sv_typed_opts(opts) {
const cfg = if opts == nil {
{}
} else {
opts
}
if type_of(cfg) != "dict" {
throw "with_structural_validator: opts must be a dict or nil; got " + type_of(cfg)
}
const report = get_typed_report(cfg, __sv_options_schema(), true)
if !report.ok {
throw "with_structural_validator: " + report.message
}
return report.value
}
fn __sv_validate_opts(opts) {
const cfg = __sv_typed_opts(opts)
let rules = __sv_normalize_rules(cfg?.rules)
for rule in rules {
if !contains(__sv_rule_names(), rule.name) {
throw "with_structural_validator: unknown rule `" + rule.name + "`"
}
}
return {
on_failure: cfg.on_failure,
max_attempts: cfg.max_attempts,
rules: rules,
locale: __sv_locale(cfg?.locale),
no_phantom_completion_catalog: __sv_normalize_no_phantom_completion_catalog(
cfg?.no_phantom_completion_catalog,
),
}
}
fn __sv_tool_annotations(entry) {
const direct = entry?.annotations
if type_of(direct) == "dict" {
return direct
}
const func = entry?.function
if type_of(func) == "dict" && type_of(func?.annotations) == "dict" {
return func.annotations
}
return {}
}
fn __sv_annotation_enabled(value) -> bool {
return type_of(value) == "bool" && value
}
fn __sv_tool_entry_is_structural(entry) -> bool {
const annotations = __sv_tool_annotations(entry)
return __sv_annotation_enabled(annotations?.structural)
|| __sv_annotation_enabled(
annotations?.agent_lifecycle,
)
}
fn __sv_tool_entry_has_write_capability(entry) -> bool {
if __sv_tool_entry_is_structural(entry) {
return false
}
const annotations = __sv_tool_annotations(entry)
const side_effect_level = lowercase(
__sv_string(annotations?.side_effect_level ?? annotations?.sideEffectLevel),
)
if side_effect_level == "none" || side_effect_level == "read_only" {
return false
}
const kind = lowercase(__sv_string(annotations?.kind))
if contains(["read", "search", "think", "fetch"], kind) {
return false
}
return true
}
fn __sv_workspace_has_write_capability(payload) -> bool {
const policy = __sv_dict(payload?.policy)
const ceiling = lowercase(__sv_string(policy?.side_effect_level))
if ceiling == "none" || ceiling == "read_only" {
return false
}
const tools = __sv_dict(payload?.tools)
let entries = __sv_list(tools?.tools)
for entry in entries {
if type_of(entry) == "dict" && __sv_tool_entry_has_write_capability(entry) {
return true
}
}
return false
}
fn __sv_has_non_structural_tools(payload) -> bool {
const tools = __sv_dict(payload?.tools)
for entry in __sv_list(tools?.tools) {
if type_of(entry) == "dict" && !__sv_tool_entry_is_structural(entry) {
return true
}
}
return false
}
fn __sv_done_marker_present(payload) -> bool {
return trim(__sv_string(payload?.parsed_done_marker)) != ""
}
fn __sv_registry_tool_entry(registry, name) {
const tools = __sv_dict(registry)
let entries = __sv_list(tools?.tools)
for entry in entries {
const direct_name = __sv_string(entry?.name)
const function = entry?.function
const function_name = if type_of(function) == "dict" {
__sv_string(function?.name)
} else {
""
}
if direct_name == name || function_name == name {
return entry
}
}
return nil
}
fn __sv_any_prior_write_tools(payload) -> bool {
const tools = payload?.tools
const prior = __sv_string_list(payload?.prior_successful_tools)
+ __sv_string_list(
payload?.prior_rejected_tools,
)
for name in prior {
const entry = __sv_registry_tool_entry(tools, name)
if entry != nil && __sv_tool_entry_has_write_capability(entry) {
return true
}
}
return false
}
fn __sv_no_phantom_completion_phrases(cfg, rule_cfg) {
const locale = __sv_locale(rule_cfg?.locale ?? cfg?.locale)
const catalog = __sv_dict(cfg?.no_phantom_completion_catalog)
const phrases = __sv_string_list(catalog[locale] ?? catalog?.en ?? [])
return __sv_merge_catalog_locale([], phrases)
}
fn __sv_claims_completion(payload, cfg, rule_cfg) -> bool {
const text = lowercase(
trim(__sv_string(payload?.assistant_text ?? payload?.raw_text ?? payload?.visible_text ?? "")),
)
if text == "" {
return false
}
for phrase in __sv_no_phantom_completion_phrases(cfg, rule_cfg) {
if phrase != "" && text.contains(phrase) {
return true
}
}
return false
}
fn __sv_non_empty_when_writes_expected(payload) {
if !__sv_workspace_has_write_capability(payload) {
return nil
}
if len(__sv_list(payload?.tool_calls)) > 0 {
return nil
}
if __sv_any_prior_write_tools(payload) {
return nil
}
if __sv_done_marker_present(payload) {
return nil
}
return {
rule: "non_empty_when_writes_expected",
diagnostic: "Assistant emitted no tool calls while writable tools were available.",
recommended_action:
"Emit the concrete write or edit tool call needed for the task, or only mark the task done after that work is complete.",
}
}
fn __sv_no_phantom_completion(payload, cfg, rule_cfg) {
if !__sv_workspace_has_write_capability(payload) {
return nil
}
if len(__sv_list(payload?.tool_calls)) > 0 {
return nil
}
if __sv_any_prior_write_tools(payload) {
return nil
}
if !__sv_claims_completion(payload, cfg, rule_cfg) {
return nil
}
return {
rule: "no_phantom_completion",
diagnostic:
"Assistant claimed completion before any write-capable tool call occurred in this session.",
recommended_action:
"Call the concrete write or edit tool that performs the work before claiming the task is complete.",
}
}
fn __sv_join_messages(values) -> string {
return join(__sv_string_list(values), "; ")
}
fn __sv_tool_call_name(call) -> string {
return trim(__sv_string(call?.name ?? call?.tool_name))
}
fn __sv_tool_call_arguments(call) {
let raw = call?.arguments ?? call?.tool_args ?? {}
if type_of(raw) == "dict" {
return raw
}
return nil
}
fn __sv_tool_entry_parameters(entry) {
const function = entry?.function
if type_of(function) == "dict" && type_of(function?.parameters) == "dict" {
return function.parameters
}
if type_of(entry?.parameters) == "dict" {
return entry.parameters
}
if type_of(entry?.input_schema) == "dict" {
return entry.input_schema
}
if type_of(entry?.inputSchema) == "dict" {
return entry.inputSchema
}
return {}
}
fn __sv_parameter_entries(parameters) {
if type_of(parameters?.properties) == "dict" {
return parameters.properties
}
let entries = {}
for name in __sv_dict(parameters).keys() {
if !contains(
["type", "properties", "required", "additionalProperties", "description", "title", "$schema"],
name,
) {
entries = entries + {[name]: parameters[name]}
}
}
return entries
}
fn __sv_param_is_required(parameters, name, schema) -> bool {
if type_of(parameters?.properties) == "dict" {
return contains(__sv_string_list(parameters?.required), name)
}
if type_of(schema) == "dict" {
if schema?.default != nil {
return false
}
if type_of(schema?.required) == "bool" {
return schema.required
}
if type_of(schema?.optional) == "bool" {
return !schema.optional
}
}
return true
}
/**
* The names of a tool's REQUIRED parameters, for the corrective that asks a
* model to emit a call it described but never sent.
*
* Names only, never values. An earlier form of this returned `[{key, value}]`
* with a type placeholder for each value (`"<string>"`), on the theory that a
* model copying the exemplar verbatim would produce an obviously unfilled call.
* It does not: live smoke showed a model copy `"<string>"` into the repair call
* and then into every subsequent call for the rest of the session, filling
* optional parameters it had never used before with the same token. A
* placeholder inside a JSON string is indistinguishable from a value, so the
* only safe exemplar is one that carries no values at all — the corrective
* renders the bare call envelope and names the required keys in prose.
*
* Empty for an unknown tool, a tool that declares no parameters, and a tool
* whose parameters are all optional. Underscore-prefixed parameters are the
* harness's to fill rather than the model's, but a REQUIRED one still has to
* be sent, so it is still named.
*
* This lives beside the parameter readers because this module already owns the
* shape tolerance for a tool entry's parameters: JSON-schema
* `properties`/`required`, and the flat `{name: schema}` form with per-field
* `required`/`optional`/`default`. A corrective that re-derived those rules
* would be a second owner of the same policy.
*
* @effects: []
* @errors: []
* @api_stability: experimental
*/
pub fn tool_required_argument_names(registry, tool_name) -> list<string> {
const entry = __sv_registry_tool_entry(registry, __sv_string(tool_name))
if entry == nil {
return []
}
const parameters = __sv_tool_entry_parameters(entry)
const entries = __sv_parameter_entries(parameters)
let out: list<string> = []
for name in __sv_dict(entries).keys() {
if __sv_param_is_required(parameters, name, entries[name]) {
out = out.appending(name)
}
}
return out
}
/**
* Compatibility projection for embedders that consumed the former exemplar
* helper. New model-facing code must use `tool_required_argument_names` and
* keep placeholder values out of the call envelope; this function preserves
* the published experimental API without making it a second owner of which
* parameters are required.
*
* @effects: []
* @errors: []
* @api_stability: experimental
*/
pub fn tool_required_argument_exemplar(registry, tool_name) -> list {
const entry = __sv_registry_tool_entry(registry, __sv_string(tool_name))
if entry == nil {
return []
}
const entries = __sv_parameter_entries(__sv_tool_entry_parameters(entry))
let out = []
for name in tool_required_argument_names(registry, tool_name) {
const types = __sv_schema_type_names(entries[name])
const placeholder = if len(types) > 0 {
"<" + types[0] + ">"
} else {
"<value>"
}
out = out.appending({key: name, value: placeholder})
}
return out
}
/**
* The output schema that makes a turn's completion a single parseable tool
* call and nothing else.
*
* Projects the registry into a JSON Schema for the flat call envelope the
* text parser already accepts, keyed with the CANONICAL argument alias
* (`{"name": .., "args": {..}}`) so the forced completion matches the shape
* every standing prompt and corrective teaches, one `anyOf` branch per tool:
* `name` is pinned to that tool and `args` must carry the tool's REQUIRED
* parameter keys (live smoke evidence: with an open arguments object the
* model lazily emits `{}` and burns a turn re-issuing the call). Argument
* values stay shallow — presence is enforced by the grammar, value shape by
* the dispatch-time validator that owns it. Internal underscore-prefixed
* parameters are the harness's to fill, not the model's to write, so they
* are left out of the grammar unless the registry marks one required (live
* smoke evidence: an offered `_nl_intent` key is what the model reaches for
* under constraint).
*
* When the corrective claimed the turn for one named tool, `claimed_tool`
* narrows the schema to that tool's branch: the claim already told the model
* which call to make, and the full `anyOf` lets it wander to a different
* tool instead (observed live: a `look` claim answered with a junk `search`).
* An empty or unrecognized `claimed_tool` keeps every branch.
*
* A tool with no model-facing parameters gets a bare `{"type": "object"}` args
* branch rather than an empty `properties` object, which llama.cpp's grammar
* conversion mishandles in a way that fails the whole request.
*
* Returns nil for an empty or unreadable registry, in which case no
* contract should be applied.
*
* This lives beside the parameter readers for the same reason as
* `tool_required_argument_names`: this module owns the shape tolerance
* for tool registry entries, and a second reader would be a second owner of
* the same policy.
*
* @effects: []
* @errors: []
* @api_stability: experimental
*/
pub fn tool_call_output_schema(registry, claimed_tool = "") -> any? {
const tools = __sv_list(__sv_dict(registry)?.tools)
const claimed = trim(__sv_string(claimed_tool))
let branches = []
let claimed_branch = nil
let seen = []
for entry in tools {
const direct_name = __sv_string(entry?.name)
const function = entry?.function
const function_name = if type_of(function) == "dict" {
__sv_string(function?.name)
} else {
""
}
const name = if direct_name != "" {
direct_name
} else {
function_name
}
if name == "" || contains(seen, name) {
continue
}
seen = seen.appending(name)
const parameters = __sv_tool_entry_parameters(entry)
const entries = __sv_parameter_entries(parameters)
let props = {}
let req = []
for pname in __sv_dict(entries).keys() {
const schema = entries[pname]
const required = __sv_param_is_required(parameters, pname, schema)
if starts_with(pname, "_") && !required {
continue
}
props = props + {[pname]: __sv_argument_property_schema(schema)}
if required {
req = req.appending(pname)
}
}
// A tool with nothing for the model to write gets the bare object type, and
// in particular no empty `properties`. llama.cpp's grammar conversion
// mishandles `{"type": "object", "properties": {}}` (ggml-org/llama.cpp
// #25923) and the malformed production fails the WHOLE combined request,
// so one zero-parameter tool in the registry would take every other branch
// down with it. Nothing is lost by leaving the keys open here: there are no
// enumerated keys to close against, and dispatch still rejects an argument
// the tool does not declare.
//
// Otherwise additionalProperties stays false: llama.cpp's grammar
// conversion treats an open object as an arbitrary key-value production,
// which is both slow and an invitation to hallucinate argument names; every
// accepted key is already enumerated from the registry, which dispatch
// validates against.
let args_schema = if len(__sv_dict(props).keys()) == 0 {
{type: "object"}
} else {
{type: "object", properties: props, additionalProperties: false}
}
if len(req) > 0 {
args_schema = args_schema + {required: req}
}
const branch = {
type: "object",
properties: {name: {type: "string", enum: [name]}, args: args_schema},
required: ["name", "args"],
additionalProperties: false,
}
if claimed != "" && name == claimed {
claimed_branch = branch
}
branches = branches.appending(branch)
}
if claimed_branch != nil {
return claimed_branch
}
if len(branches) == 0 {
return nil
}
if len(branches) == 1 {
return branches[0]
}
return {anyOf: branches}
}
/**
* Project one named registry tool into the canonical call-envelope schema.
*
* Unlike `tool_call_output_schema`, this fails closed for an absent name. A
* one-turn execution claim must never degrade into the full tool union: doing
* so would turn "call verify next" into "call anything next" while still
* looking structurally constrained to its caller.
*
* @effects: []
* @errors: []
* @api_stability: experimental
*/
pub fn exact_tool_call_output_schema(registry, claimed_tool: string) -> any? {
const claimed = trim(claimed_tool)
if claimed == "" {
return nil
}
const schema = tool_call_output_schema(registry, claimed)
if schema == nil || schema?.properties?.name?.enum != [claimed] {
return nil
}
return schema
}
/**
* A deliberately shallow JSON-Schema fragment for one tool argument.
*
* Only a single unambiguous scalar/container type is projected; anything
* richer (unions, nested objects, enums) becomes a permissive `{}` so the
* decode grammar never forbids a value the dispatch-time validator would
* accept. Constrained emission against a deep union schema is a known
* field-dropping hazard on llama.cpp; presence is enforced here, shape is
* enforced where it is owned — at dispatch.
*
* @effects: []
* @errors: []
* @api_stability: experimental
*/
fn __sv_argument_property_schema(schema) -> dict {
const types = __sv_schema_type_names(schema)
if len(types) != 1 {
return {}
}
const mapped = __sv_json_schema_type_name(types[0])
if mapped == "" {
return {}
}
return {type: mapped}
}
fn __sv_json_schema_type_name(name) -> string {
if name == "string" {
return "string"
}
if name == "integer" || name == "int" {
return "integer"
}
if name == "number" || name == "float" {
return "number"
}
if name == "boolean" || name == "bool" {
return "boolean"
}
if name == "array" || name == "list" {
return "array"
}
if name == "object" || name == "dict" {
return "object"
}
return ""
}
fn __sv_schema_type_names(schema) {
let raw = nil
if type_of(schema) == "string" {
raw = schema
} else if type_of(schema) == "dict" {
raw = schema?.type
}
let names = []
if type_of(raw) == "list" {
for item in raw {
const name = lowercase(trim(__sv_string(item)))
if name != "" {
names = names.appending(name)
}
}
} else {
const name = lowercase(trim(__sv_string(raw)))
if name != "" {
names = names.appending(name)
}
}
return names
}
fn __sv_value_matches_schema_type(value, type_name) -> bool {
const actual = type_of(value)
if type_name == "any" || type_name == "unknown" {
return true
}
if type_name == "string" {
return actual == "string"
}
if type_name == "integer" || type_name == "int" {
return actual == "int"
}
if type_name == "number" || type_name == "float" {
return actual == "int" || actual == "float"
}
if type_name == "boolean" || type_name == "bool" {
return actual == "bool"
}
if type_name == "array" || type_name == "list" {
return actual == "list"
}
if type_name == "object" || type_name == "dict" {
return actual == "dict"
}
return true
}
fn __sv_type_violation(tool_name, arg_name, value, schema) {
if value == nil {
return nil
}
const expected = __sv_schema_type_names(schema)
if len(expected) == 0 {
return nil
}
for type_name in expected {
if __sv_value_matches_schema_type(value, type_name) {
return nil
}
}
return "Tool '"
+ tool_name
+ "' parameter '"
+ arg_name
+ "' expected "
+ join(expected, "|")
+ " but got "
+ type_of(value)
+ "."
}
fn __sv_tool_schema_violations(payload) {
let violations = []
const registry = payload?.tools
for call in __sv_list(payload?.tool_calls) {
const tool_name = __sv_tool_call_name(call)
if tool_name == "" {
violations = violations.appending("Tool call is missing a tool name.")
continue
}
const entry = __sv_registry_tool_entry(registry, tool_name)
if entry == nil {
violations = violations.appending("Unknown tool '" + tool_name + "'.")
continue
}
const args = __sv_tool_call_arguments(call)
if args == nil {
violations = violations.appending("Tool '" + tool_name + "' arguments must be a dict.")
continue
}
const parameters = __sv_tool_entry_parameters(entry)
let entries = __sv_parameter_entries(parameters)
let missing = []
for name in entries.keys() {
const schema = entries[name]
if __sv_param_is_required(parameters, name, schema) && args[name] == nil {
missing = missing.appending(name)
} else {
const type_error = __sv_type_violation(tool_name, name, args[name], schema)
if type_error != nil {
violations = violations.appending(type_error)
}
}
}
if len(missing) > 0 {
violations = violations.appending(
"Tool '"
+ tool_name
+ "' is missing required parameter(s): "
+ join(missing, ", ")
+ ". Provide all required parameters and try again.",
)
}
}
return violations
}
fn __sv_tool_calls_well_formed(payload) {
// Both `text` (tagged/heredoc) and `json` (fenced-JSON) ride the TEXT channel
// and surface `tool_parse_errors`/`protocol_violations` from the text parser,
// so the well-formedness veto must enforce for either. Gating on `text` alone
// would silently stop vetoing malformed turns once `json` became the global
// default. (Native rides the provider channel and is never enforced here.)
const format = lowercase(__sv_string(payload?.tool_format))
const is_text_tool_format = format == "text" || format == "json"
const should_enforce_text_protocol = is_text_tool_format && __sv_has_non_structural_tools(payload)
const parse_errors = if should_enforce_text_protocol {
__sv_string_list(payload?.tool_parse_errors)
} else {
[]
}
// `protocol_violations` are the recoverable wrapper/stray-prose class
// (leading narration outside `<assistant_prose>`, a call emitted bare or
// angle-wrapped instead of in `<tool_call>` tags, etc.). The text parser
// already RECOVERS those calls and surfaces them in `tool_calls`, so when
// dispatchable calls exist the turn did real work — vetoing it here would
// silently drop a fully-parsed `edit({...})` and strand the model in a
// re-emit loop until the stall detector fires. Only let a
// protocol-violation complaint veto when no calls were recovered (a turn
// that was pure stray prose). Genuine `parse_errors` (a call that failed to
// parse) and `schema_violations` (a parsed call with bad args) still veto
// regardless, because those represent broken calls, not wrapper noise.
const has_dispatchable_calls = len(__sv_list(payload?.tool_calls)) > 0
const protocol_violations = if should_enforce_text_protocol && !has_dispatchable_calls {
protocol_violation_messages(payload?.protocol_violations ?? [])
} else {
[]
}
const schema_violations = __sv_tool_schema_violations(payload)
if len(parse_errors) == 0 && len(protocol_violations) == 0 && len(schema_violations) == 0 {
return nil
}
const details = __sv_join_messages(parse_errors + schema_violations + protocol_violations)
return {
rule: "tool_calls_well_formed",
diagnostic: "Assistant emitted malformed tool calls: " + details,
recommended_action:
"Emit only well-formed tool calls that match the bound tool schemas and Harn tool-call protocol.",
}
}
fn __sv_output_token_cap_with_zero_calls(payload) {
if !__sv_workspace_has_write_capability(payload) {
return nil
}
if len(__sv_list(payload?.tool_calls)) > 0 {
return nil
}
const max_output_tokens = to_int(payload?.max_output_tokens) ?? 0
if max_output_tokens <= 0 {
return nil
}
const output_tokens = to_int(payload?.output_tokens) ?? 0
if output_tokens * 100 < max_output_tokens * 95 {
return nil
}
return {
rule: "output_token_cap_with_zero_calls",
diagnostic:
"Assistant used nearly the full output-token budget without emitting any tool calls.",
recommended_action:
"The model appears stuck in a prose loop. Emit the next tool call directly or shorten the narration and try again.",
}
}
fn __sv_feedback_payload(verdict) {
return json_stringify(
{
rule: verdict.rule,
diagnostic: verdict.diagnostic,
recommended_action: verdict.recommended_action,
},
)
}
fn __sv_emit_decision(
agent: HarnessAgent,
session_id,
iteration,
cfg,
verdict,
attempts,
vetoed = true,
skipped = false,
reason = nil,
) {
agent_emit_event(
agent,
session_id,
"structural_validator_decision",
{
iteration: iteration,
rule: verdict?.rule ?? "",
diagnostic: verdict?.diagnostic ?? "",
recommended_action: verdict?.recommended_action ?? "",
vetoed: vetoed,
skipped: skipped,
reason: reason,
on_failure: cfg.on_failure,
attempts: attempts,
max_attempts: cfg.max_attempts,
},
)
}
fn __sv_pass_result(call, configured, skipped = false, reason = nil, extra = nil) {
const result = {configured: configured, vetoed: false, skipped: skipped, reason: reason}
let merged = if type_of(extra) == "dict" {
result + extra
} else {
result
}
return {
ok: true,
status: "ok",
tool_name: call.tool_name,
tool_call_id: call.call_id,
arguments: call.tool_args,
result: merged,
rendered_result: "",
observation: "",
error: nil,
error_category: nil,
executor: "harn",
}
}
fn __sv_veto_result(call, cfg, verdict) {
return {
ok: true,
status: "ok",
tool_name: call.tool_name,
tool_call_id: call.call_id,
arguments: call.tool_args,
result: {
configured: true,
vetoed: true,
skipped: false,
rule: verdict.rule,
diagnostic: verdict.diagnostic,
recommended_action: verdict.recommended_action,
feedback: __sv_feedback_payload(verdict),
on_failure: cfg.on_failure,
},
rendered_result: verdict.diagnostic,
observation: "",
error: nil,
error_category: nil,
executor: "harn",
}
}
fn __sv_rule_verdict(payload, cfg, rule_cfg) {
const rule = __sv_string(rule_cfg?.name)
if rule == "non_empty_when_writes_expected" {
return __sv_non_empty_when_writes_expected(payload)
}
if rule == "no_phantom_completion" {
return __sv_no_phantom_completion(payload, cfg, rule_cfg)
}
if rule == "tool_calls_well_formed" {
return __sv_tool_calls_well_formed(payload)
}
if rule == "output_token_cap_with_zero_calls" {
return __sv_output_token_cap_with_zero_calls(payload)
}
return nil
}
fn __sv_handle_turn(agent: HarnessAgent, call, cfg) {
const payload = __sv_dict(call?.tool_args)
const session_id = __sv_string(payload?.session_id ?? call?.turn?.session_id)
const iteration = to_int(payload?.iteration ?? call?.turn?.iteration) ?? 0
const attempts = to_int(payload?.attempts) ?? 0
if attempts >= cfg.max_attempts {
__sv_emit_decision(
agent,
session_id,
iteration,
cfg,
nil,
attempts,
false,
true,
"max_attempts_reached",
)
return __sv_pass_result(call, true, true, "max_attempts_reached")
}
for rule_cfg in cfg.rules {
const verdict = __sv_rule_verdict(payload, cfg, rule_cfg)
if verdict != nil {
if rule_cfg?.warn_only ?? false {
__sv_emit_decision(
agent,
session_id,
iteration,
cfg,
verdict,
attempts,
false,
false,
"warn_only",
)
return __sv_pass_result(
call,
true,
false,
"warn_only",
{
warned: true,
rule: verdict.rule,
diagnostic: verdict.diagnostic,
recommended_action: verdict.recommended_action,
},
)
}
__sv_emit_decision(agent, session_id, iteration, cfg, verdict, attempts)
return __sv_veto_result(call, cfg, verdict)
}
}
return __sv_pass_result(call, true)
}
/**
* with_structural_validator(opts) -> caller
*
* Deterministic pre-dispatch turn validator. The current landable rule
* set covers deterministic tool-use and completion-shape checks.
*
* Options:
* on_failure: "regenerate_with_feedback" | "raise"
* max_attempts: int > 0 (default 3)
* locale: string (default "en")
* no_phantom_completion_catalog: {locale: [phrase, ...]}
* rules: ["non_empty_when_writes_expected", ...]
* or [{name: "tool_calls_well_formed", warn_only: true}]
*
* @effects: [host]
* @errors: [runtime]
* @api_stability: experimental
* Pass the returned closure via `agent_loop(harness, {structural_validator: ...})`.
*
* @example: with_structural_validator({on_failure: "regenerate_with_feedback"})
*/
pub fn with_structural_validator(agent: HarnessAgent, opts = nil) {
const cfg = __sv_validate_opts(opts)
return { call, next ->
if call?.tool_name != __sv_tool_name() {
return next(call)
}
return __sv_handle_turn(agent, call, cfg)
}
}