// Tool-surface narrowing: decide which tools stay on the model's surface
// after a turn, and project that decision onto a registry and a policy.
//
// The post-turn seam calls in with the turn's attempted tools; `std/agent/stance`
// and `std/agent/lanes` reuse the registry/policy projections directly.
import "std/agent/loop_result_status"
import { agent_session_record_skill_event } from "std/agent/state"
fn __tool_surface_name(entry) {
return entry?.name ?? entry?.function?.name ?? ""
}
fn __tool_surface_current_entries(opts) {
const registry = opts?.tools
const entries = if type_of(registry) == "dict" {
registry?.tools ?? []
} else if type_of(registry) == "list" {
registry
} else {
[]
}
let selected = []
let names = []
for entry in entries {
const name = __tool_surface_name(entry)
if name != "" && !contains(names, name) {
names = names.appending(name)
selected = selected.appending(entry)
}
}
return selected
}
/**
* Internal, shared with `std/agent/postturn`: the tool names currently on
* the surface.
*
* @effects: []
* @errors: []
* @api_stability: experimental
*/
pub fn __tool_surface_current_names(opts) -> list {
let names = []
for entry in __tool_surface_current_entries(opts) {
names = names.appending(__tool_surface_name(entry))
}
return names
}
fn __tool_surface_call_name(call) {
return call?.name ?? call?.function?.name ?? call?.tool ?? call?.tool_name ?? ""
}
/**
* Internal, shared with `std/agent/postturn`: the tool names this turn
* attempted, from the model's calls and the dispatch results.
*
* @effects: []
* @errors: []
* @api_stability: experimental
*/
pub fn __tool_surface_attempted_names(llm_result, dispatch_results) -> list {
let names = []
for call in llm_result?.tool_calls ?? [] {
const name = __tool_surface_call_name(call)
if name != "" && !contains(names, name) {
names = names.appending(name)
}
}
for result in dispatch_results {
const name = __tool_result_name(result)
if name != "" && !contains(names, name) {
names = names.appending(name)
}
}
return names
}
fn __tool_surface_append_history(history, attempted, window_turns) {
let next = history.appending(attempted)
while len(next) > window_turns {
next = next[1:]
}
return next
}
fn __tool_surface_window_used(history) {
let used = []
for turn in history {
for name in turn {
if name != "" && !contains(used, name) {
used = used.appending(name)
}
}
}
return used
}
fn __tool_surface_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 __tool_surface_text(value) {
if value == nil {
return ""
}
return lowercase(trim(to_string(value)))
}
fn __tool_surface_enabled(value) {
return type_of(value) == "bool" && value
}
fn __tool_surface_entry_class(entry) {
const annotations = __tool_surface_annotations(entry)
const side_effect = __tool_surface_text(
annotations?.side_effect_level ?? annotations?.sideEffectLevel ?? annotations?.side_effect
?? annotations
?.sideEffect,
)
const kind = __tool_surface_text(
annotations?.kind ?? annotations?.tool_kind ?? annotations?.toolKind,
)
const name = __tool_surface_text(__tool_surface_name(entry))
if __tool_surface_enabled(annotations?.agent_lifecycle)
|| __tool_surface_enabled(
annotations?.structural,
)
|| contains(["session_control", "lifecycle", "suspend", "resume"], kind)
|| contains(
["load_skill", "agent_await_resumption", "subagent_pause", "subagent_resume"],
name,
) {
return "session_control"
}
if __tool_surface_enabled(annotations?.approval)
|| __tool_surface_enabled(
annotations?.requires_approval,
)
|| contains(["approval", "human_gate"], kind) {
return "approval"
}
if __tool_surface_enabled(annotations?.progress) || kind == "progress" {
return "progress"
}
if __tool_surface_enabled(annotations?.result_reader)
|| __tool_surface_enabled(
annotations?.result_polling,
)
|| kind == "result_polling"
|| starts_with(name, "read_command_output") {
return "result_polling"
}
if side_effect != "" && side_effect != "none" && side_effect != "read_only"
&& side_effect != "read" {
return "mutating"
}
if contains(
[
"edit",
"write",
"execute",
"run",
"command",
"scaffold",
"delete",
"mutation",
"mutate",
"network",
],
kind,
) {
return "mutating"
}
if side_effect == "none" || side_effect == "read_only" || side_effect == "read" {
return "read_only"
}
if contains(["read", "search", "fetch", "think", "lookup", "inspect"], kind) {
return "read_only"
}
if __tool_surface_enabled(annotations?.readOnlyHint)
&& !__tool_surface_enabled(
annotations?.destructiveHint,
) {
return "read_only"
}
return "unknown"
}
fn __tool_surface_entry_detail(entry, class, reason) {
const annotations = __tool_surface_annotations(entry)
return {
name: __tool_surface_name(entry),
class: class,
kind: __tool_surface_text(annotations?.kind ?? annotations?.tool_kind ?? annotations?.toolKind),
side_effect_level: __tool_surface_text(
annotations?.side_effect_level ?? annotations?.sideEffectLevel ?? annotations?.side_effect
?? annotations
?.sideEffect,
),
reason: reason,
}
}
fn __tool_surface_entry_prunable(entry, config) {
const class = __tool_surface_entry_class(entry)
const keep_classes = config?.keep_classes ?? []
if contains(keep_classes, class) {
return false
}
if class == "unknown" {
const unknown_tool_policy = config?.unknown_tool_policy ?? "keep"
return unknown_tool_policy == "prune"
}
const mode = config?.mode ?? "safe"
if mode == "aggressive" {
return true
}
return contains(config?.prune_classes ?? ["read_only"], class)
}
fn __tool_surface_removed(candidates, used, config) {
const hard_keep = config?.hard_keep ?? []
let removed = []
for entry in candidates {
const name = __tool_surface_name(entry)
if name == "" || contains(used, name) || contains(hard_keep, name) {
continue
}
const class = __tool_surface_entry_class(entry)
if __tool_surface_entry_prunable(entry, config) {
removed = removed.appending(
__tool_surface_entry_detail(entry, class, "unused_prunable_class"),
)
}
}
return removed
}
fn __tool_surface_removed_names(details) {
let names = []
for detail in details {
const name = detail?.name ?? ""
if name != "" && !contains(names, name) {
names = names.appending(name)
}
}
return names
}
fn __tool_surface_keep_details(candidates, removed, used, hard_keep) {
let details = []
for entry in candidates {
const name = __tool_surface_name(entry)
if name == "" || contains(removed, name) {
continue
}
const class = __tool_surface_entry_class(entry)
const reason = if contains(used, name) {
"used_in_window"
} else if contains(hard_keep, name) {
"hard_keep"
} else {
"class_kept"
}
details = details.appending(__tool_surface_entry_detail(entry, class, reason))
}
return details
}
fn __tool_surface_remaining(candidates, removed) {
let remaining = []
for name in candidates {
if !contains(removed, name) {
remaining = remaining.appending(name)
}
}
return remaining
}
/**
* Internal, shared with `std/agent/postturn`: fold this turn into the
* narrowing window and return the resulting history and surface.
*
* @effects: []
* @errors: []
* @api_stability: experimental
*/
pub fn __tool_surface_narrowing_post_turn(
agent: HarnessAgent,
session,
llm_result,
dispatch_results,
opts,
) -> dict {
const config = opts?.tool_surface_narrowing ?? {}
if !(config?.enabled ?? true) {
return {history: [], narrowed_tools: nil, changed: false}
}
const window_turns = config?.window_turns ?? 5
const attempted = __tool_surface_attempted_names(llm_result, dispatch_results)
const history = __tool_surface_append_history(
opts?._tool_surface_narrowing_history ?? [],
attempted,
window_turns,
)
const candidate_entries = __tool_surface_current_entries(opts)
const candidates = __tool_surface_current_names(opts)
if len(candidates) == 0 || len(history) < window_turns {
return {history: history, narrowed_tools: opts?._tool_surface_narrowed_tools, changed: false}
}
const used = __tool_surface_window_used(history)
const hard_keep = config?.hard_keep ?? []
const removed_details = __tool_surface_removed(candidate_entries, used, config)
const removed = __tool_surface_removed_names(removed_details)
if len(removed) == 0 {
return {history: history, narrowed_tools: opts?._tool_surface_narrowed_tools, changed: false}
}
const remaining = __tool_surface_remaining(candidates, removed)
const kept_details = __tool_surface_keep_details(candidate_entries, removed, used, hard_keep)
const reason = "unused across last " + to_string(window_turns) + " turns"
agent_session_record_skill_event(
agent,
session.session_id,
"skill_narrow",
{
reason: reason,
removed_tools: removed,
remaining_tools: remaining,
window_turns: window_turns,
policy: {
mode: config?.mode ?? "safe",
prune_classes: config?.prune_classes ?? ["read_only"],
keep_classes: config?.keep_classes ?? [],
unknown_tool_policy: config?.unknown_tool_policy ?? "keep",
hard_keep: hard_keep,
},
removed_tool_details: removed_details,
kept_tool_details: kept_details,
},
)
return {
history: history,
narrowed_tools: remaining,
changed: true,
removed_tools: removed,
remaining_tools: remaining,
removed_tool_details: removed_details,
kept_tool_details: kept_details,
reason: reason,
}
}
fn __tool_surface_keep_entry(entry, names) {
return contains(names, __tool_surface_name(entry))
}
/**
* Internal, shared with `std/agent/stance`: filter a tool registry down
* to the named entries.
*
* @effects: []
* @errors: []
* @api_stability: experimental
*/
pub fn __tool_surface_filter_registry(registry, names) {
if len(names) == 0 {
return registry
}
if type_of(registry) == "dict" {
let kept = []
for entry in registry?.tools ?? [] {
if __tool_surface_keep_entry(entry, names) {
kept = kept.appending(entry)
}
}
return registry + {tools: kept}
}
if type_of(registry) == "list" {
let kept = []
for entry in registry {
if __tool_surface_keep_entry(entry, names) {
kept = kept.appending(entry)
}
}
return kept
}
return registry
}
fn __tool_surface_policy_allowlist(names) {
if len(names) == 0 {
return ["__harn_tool_surface_no_tools__"]
}
return names
}
/**
* Internal, shared with `std/agent/stance`: intersect a policy's tool
* allowlist with the named entries.
*
* @effects: []
* @errors: []
* @api_stability: experimental
*/
pub fn __tool_surface_policy_tools(policy, names) {
if policy != nil && type_of(policy) != "dict" {
return policy
}
const base = policy ?? {}
const existing = base?.tools ?? []
if type_of(existing) == "list" && len(existing) > 0 {
let intersected = []
for name in existing {
if contains(names, name) {
intersected = intersected.appending(name)
}
}
return base + {tools: __tool_surface_policy_allowlist(intersected)}
}
return base + {tools: __tool_surface_policy_allowlist(names)}
}
/**
* agent_apply_tool_surface_narrowing.
*
* @effects: []
* @errors: []
* @api_stability: experimental
*/
pub fn agent_apply_tool_surface_narrowing(opts) {
const names = opts?._tool_surface_narrowed_tools
if len(names ?? []) == 0 {
return opts
}
return opts
+ {
tools: __tool_surface_filter_registry(opts?.tools, names),
policy: __tool_surface_policy_tools(opts?.policy, names),
}
}
/**
* agent_reset_tool_surface_narrowing.
*
* @effects: []
* @errors: []
* @api_stability: experimental
*/
pub fn agent_reset_tool_surface_narrowing(opts) {
return opts
+ {_tool_surface_narrowed_tools: nil, _tool_surface_narrowing_history: []}
}