import { ast_search } from "std/ast"
import { AgentGateBindingScope, agent_gate_binding_scopes_reads } from "std/dev/agent_gate_bindings"
import { git_run } from "std/git"
/** Behavior configuration owned by the runner, stop decision, or stall handler. */
pub type AgentGateEntry = {
name: string,
kind: string,
layer: "runner" | "stop decision" | "stall handler",
default: string,
readers: list<{file: string, line: int, reader: string}>,
reachability: "yes" | "no" | "unknown",
evidence: string,
verdict: "DELETE" | "PROMOTE-TO-DEFAULT" | "KEEP-with-expiry",
reason: string,
expiry?: string?,
}
type AgentGateEntries = list<AgentGateEntry>
/** Structural selectors capture a key independently of the registered names. */
pub type AgentGateScope = {
id: string,
kind: "flag" | "option" | "environment" | "boundary",
glob: string,
language: string,
query: string,
prefix: string,
key_form: "literal" | "field",
}
/** A dynamic forwarding boundary must explain who validates the forwarded key. */
pub type AgentGateForwarder = {
scope: string,
file: string,
expression: string,
owner: string,
evidence: string,
/** Local consumer origins whose removal would make this classification unsafe. */
consumers?: list<{scope: string, function: string, parameter: string, path?: list<string>}>,
}
/** Source boundary for one configuration owner. */
pub type AgentGateBindingSource = {
glob: string,
extra_globs?: list<string>,
config: AgentGateBindingScope,
}
/** Per-repository policy data consumed by the shared structural audit. */
pub type AgentGateRegistry = {
schema_version: int,
repo: string,
entry_files: list<string>,
scopes: list<AgentGateScope>,
binding_scopes: list<AgentGateBindingSource>,
binding_scope_files?: list<string>,
forwarder_files: list<string>,
non_behavior_files: list<string>,
name_prefixes: list<{from: string, to: string}>,
name_aliases: dict<string, string>,
projections: list<{layer: string, file: string}>,
}
/** One inspected source read, including the structural selector that found it. */
pub type AgentGateRead = {
scope: string,
file: string,
line: int,
expression: string,
name: string,
reader: string,
}
type AgentGateClassifications = list<AgentGateForwarder>
fn read_classifications(fs: HarnessFs, files: list<string>) -> AgentGateClassifications {
let rows: AgentGateClassifications = []
for file in files {
const batch = schema_expect(json_parse(fs.read_text(file)), schema_of(AgentGateClassifications))
if len(batch) == 0 {
throw "agent gate registry: empty classification shard ${file}"
}
rows = rows + batch
}
return rows
}
fn validate_classified_consumers(
registry: AgentGateRegistry,
classifications: AgentGateClassifications,
) {
for classification in classifications {
for consumer in classification.consumers ?? [] {
const covered = registry.binding_scopes.any(
fn(source) { return source.config.id == consumer.scope
&& source.config.parameters.any(
fn(parameter) { return parameter.function == consumer.function
&& parameter.name == consumer.parameter
&& (parameter.path ?? []) == (consumer.path ?? []) },
) },
)
if !covered {
throw "agent gate registry: missing classified consumer origin ${consumer.scope} ${consumer.function}(${consumer.parameter}) for ${classification.file} ${classification.expression}"
}
}
}
}
fn read_entries(fs: HarnessFs, registry: AgentGateRegistry) -> list<AgentGateEntry> {
let entries: list<AgentGateEntry> = []
let names: list<string> = []
for file in registry.entry_files {
const batch = schema_expect(json_parse(fs.read_text(file)), schema_of(AgentGateEntries))
if len(batch) == 0 {
throw "agent gate registry: empty entry shard ${file}"
}
for entry in batch {
if entry.name.trim() == "" || contains(names, entry.name) {
throw "agent gate registry: empty or duplicate name ${entry.name}"
}
if entry.default.trim() == "" || entry.evidence.trim() == "" || entry.reason.trim() == "" {
throw "agent gate registry: missing decision evidence for ${entry.name}"
}
if entry.verdict == "KEEP-with-expiry" && (entry.expiry ?? "").trim() == "" {
throw "agent gate registry: retention has no expiry for ${entry.name}"
}
if entry.expiry != nil
&& len(regex_captures("^[0-9]{4}-[0-9]{2}-[0-9]{2}$", entry.expiry)) == 0 {
throw "agent gate registry: expiry must be an ISO date for ${entry.name}"
}
names = names + [entry.name]
entries = entries + [entry]
}
}
if len(entries) == 0 {
throw "agent gate registry: no entries; measured nothing"
}
return entries
}
fn key_name(expression: string, form: "literal" | "field") -> string? {
const text = expression.trim()
if text.starts_with("\"") && text.ends_with("\"") {
const decoded = json_parse(text)
if type_of(decoded) == "string" {
return to_string(decoded)
}
return nil
}
// Only grammar field captures give an identifier a literal key identity.
if form == "field" && len(regex_captures("^[A-Za-z_][A-Za-z0-9_]*$", text)) > 0 {
return text
}
return nil
}
fn source_files(
harness: {fs: HarnessFs, process: HarnessProcess},
pattern: string,
) -> list<string> {
const files = if harness.fs.exists(pattern) {
[pattern]
} else {
const listing = git_run(
harness.process,
["ls-files", "--cached", "--others", "--exclude-standard", "-z", "--", ":(glob)" + pattern],
{repo: harness.fs.cwd(), max_inline_bytes: 4000000},
)
if !listing.success || listing.timed_out == true {
throw "agent gate registry: source enumeration failed for ${pattern}"
}
const output = if listing.stdout_path != nil {
harness.fs.read_text(listing.stdout_path)
} else {
listing.stdout
}
if len(output) >= 4000000 || (output != "" && !output.ends_with("\0")) {
throw "agent gate registry: incomplete source enumeration for ${pattern}"
}
output.split("\0").filter({ path -> path != "" }).unique().sorted()
}
if len(files) == 0 {
throw "agent gate registry: scope ${pattern} matched no files"
}
return files
}
fn resolve_binding_sources(fs: HarnessFs, registry: AgentGateRegistry) -> AgentGateRegistry {
let sources = registry.binding_scopes
for file in registry.binding_scope_files ?? [] {
const source = schema_expect(json_parse(fs.read_text(file)), schema_of(AgentGateBindingSource))
sources = sources + [source]
}
return registry + {binding_scopes: sources, binding_scope_files: []}
}
/**
* Scan every declared source scope; empty, partial, or invalid parsing fails closed.
* @effects: [host]
* @errors: [validation, backend]
* @api_stability: experimental
*/
pub fn agent_gate_reads(
harness: {ast: HarnessAst, fs: HarnessFs, process: HarnessProcess},
registry_input: AgentGateRegistry,
) -> list<AgentGateRead> {
const registry = resolve_binding_sources(harness.fs, registry_input)
let reads: list<AgentGateRead> = []
for scope in registry.scopes {
const files = source_files({fs: harness.fs, process: harness.process}, scope.glob)
let scope_count = 0
for file in files {
const result = ast_search(
harness.ast,
{path: file, language: scope.language, query: scope.query, max_matches: 0},
)
if result?.result != "ok" || result?.had_errors != false || result?.truncated != false {
throw "agent gate registry: incomplete structural scan ${scope.id} ${file}: ${json_stringify(result)}"
}
for found in result.matches {
const capture = found?.captures?.key
if capture == nil {
throw "agent gate registry: selector ${scope.id} did not capture key"
}
const expression = to_string(capture.text)
const key = key_name(expression, scope.key_form)
reads = reads
+ [
{
scope: scope.id,
reader: to_string(
found.captures?.callee?.text ?? found.captures?.accessor?.text
?? found.captures?.owner?.text
?? found.captures?.constant?.text
?? found.captures?.method?.text
?? scope.id,
),
file: file,
line: to_int(capture.range.start_row) + 1,
expression: expression,
name: if key == nil {
""
} else {
scope.prefix + key
},
},
]
scope_count += 1
}
}
if scope_count == 0 {
throw "agent gate registry: scope ${scope.id} measured no reads"
}
}
let owners_by_file: dict<string, list<AgentGateBindingScope>> = {}
let observed_owners: list<string> = []
let observed_parameters: list<{scope: string, function: string, name: string}> = []
for scope in registry.binding_scopes {
let files: list<string> = []
for pattern in [scope.glob] + (scope.extra_globs ?? []) {
files = files + source_files({fs: harness.fs, process: harness.process}, pattern)
}
for file in files.unique() {
owners_by_file[file] = (owners_by_file[file] ?? []) + [scope.config]
}
}
for file in owners_by_file.keys().sorted() {
const owners = owners_by_file[file]
if owners == nil {
throw "agent gate registry: missing configuration owners for ${file}"
}
const found = agent_gate_binding_scopes_reads({ast: harness.ast, fs: harness.fs}, file, owners)
observed_owners = observed_owners + found.owners
observed_parameters = observed_parameters + found.parameters
reads = reads + found.reads
for unresolved in found.unresolved {
reads = reads
+ [
{
scope: unresolved.scope,
reader: unresolved.reader,
file: unresolved.file,
line: unresolved.line,
expression: unresolved.expression,
name: "",
},
]
}
}
for scope in registry.binding_scopes {
for parameter in scope.config.parameters {
if !contains(
observed_parameters,
{scope: scope.config.id, function: parameter.function, name: parameter.name},
) {
throw "agent gate registry: stale parameter origin ${scope.config.id} ${parameter.function}(${parameter.name})"
}
}
if !contains(observed_owners, scope.config.id) {
throw "agent gate registry: binding scope ${scope.config.id} has no parsed configuration origin"
}
}
if len(reads) == 0 {
throw "agent gate registry: no structural reads; measured nothing"
}
return reads.map(
fn(read) {
const exact = registry.name_aliases[read.name]
if exact != nil {
return read + {name: exact}
}
let selected: {from: string, to: string}? = nil
for alias in registry.name_prefixes {
if read.name.starts_with(alias.from)
&& (selected == nil || len(alias.from) > len(selected.from)) {
selected = alias
}
}
return if selected == nil {
read
} else {
read + {name: selected.to + read.name.slice(len(selected.from))}
}
},
)
.to_list()
}
fn cell(text: string) -> string {
return replace(replace(replace(text, "|", "\\|"), "_", "\\_"), "\n", " ")
}
/**
* Generate the reviewable projection from the same typed entries the checker reads.
* @effects: []
* @errors: []
* @api_stability: experimental
*/
pub fn agent_gate_markdown(repo: string, layer: string, entries: list<AgentGateEntry>) -> string {
let lines = [
"# " + layer + " configuration",
"",
"Generated from the typed agent gate registry. Edit the registry, then regenerate.",
"",
"Reachability records its evidence; source wiring does not prove runtime efficacy.",
"",
"| Name | Repo | Default | Readers | Headless reachability and evidence | Verdict | Reason / expiry |",
"| --- | --- | --- | --- | --- | --- | --- |",
]
for entry in entries {
if entry.layer != layer {
continue
}
let readers: list<string> = []
for reader in entry.readers {
readers = readers + [reader.file + ":" + to_string(reader.line) + " (" + reader.reader + ")"]
}
lines = lines
+ [
"| " + cell(entry.name) + " | " + cell(repo) + " | " + cell(entry.default)
+ " | "
+ cell(readers.join("; "))
+ " | "
+ cell(entry.reachability + ": " + entry.evidence)
+ " | "
+ entry.verdict
+ " | "
+ cell(entry.reason + " / " + (entry.expiry ?? "none"))
+ " |",
]
}
return lines.join("\n") + "\n"
}
fn classification_index(read: AgentGateRead, rows: list<AgentGateForwarder>) -> int? {
let selected: int? = nil
for index in range(0, len(rows)) {
const row = rows[index]
if row.scope != read.scope || row.file != read.file || row.expression != read.expression {
continue
}
if selected != nil {
throw "agent gate registry: duplicate classification ${read.file} ${read.expression}"
}
if row.owner.trim() == "" || row.evidence.trim() == "" {
throw "agent gate registry: classification lacks owner/evidence"
}
selected = index
}
return selected
}
fn observed_entries(
entries: list<AgentGateEntry>,
reads: list<AgentGateRead>,
) -> list<AgentGateEntry> {
return entries.map(
fn(entry) {
let readers: list<{file: string, line: int, reader: string}> = []
let seen: list<string> = []
for read in reads {
if read.name != entry.name {
continue
}
const reader = {file: read.file, line: read.line, reader: read.reader}
const identity = json_stringify(reader)
if !contains(seen, identity) {
readers = readers + [reader]
seen = seen + [identity]
}
}
return entry + {readers: readers}
},
)
.to_list()
}
fn write_observed_entries(fs: HarnessFs, files: list<string>, entries: list<AgentGateEntry>) {
for file in files {
const previous = schema_expect(json_parse(fs.read_text(file)), schema_of(AgentGateEntries))
const names = previous.map(fn(entry) { return entry.name })
fs.write_text(
file,
json_stringify_pretty(entries.filter(fn(entry) { return contains(names, entry.name) }))
+ "\n",
)
}
}
/**
* Audit discovered reads without treating an empty discovery as a passing census.
* @effects: []
* @errors: [validation]
* @api_stability: experimental
*/
pub fn agent_gate_findings(
entries: list<AgentGateEntry>,
reads: list<AgentGateRead>,
forwarders: list<AgentGateForwarder>,
non_behavior_reads: list<AgentGateForwarder> = [],
flag_scopes: list<string> = [],
) -> {failures: list<string>, forwarded: int, non_behavior: int} {
if len(entries) == 0 || len(reads) == 0 {
throw "agent gate registry: empty census; measured nothing"
}
const names = entries.map(fn(entry) { return entry.name })
let failures: list<string> = []
let forwarded = 0
let non_behavior = 0
let used_forwarders: list<int> = []
let used_exclusions: list<int> = []
for read in reads {
const excluded = classification_index(read, non_behavior_reads)
const forwarding = classification_index(read, forwarders)
if excluded != nil && forwarding != nil {
throw "agent gate registry: conflicting classifications ${read.file} ${read.expression}"
}
if excluded != nil {
if contains(flag_scopes, read.scope) {
throw "agent gate registry: flags cannot be excluded as domain facts"
}
used_exclusions = used_exclusions + [excluded]
non_behavior += 1
continue
}
if forwarding != nil {
if read.expression.trim().starts_with("\"") {
throw "agent gate registry: a literal key cannot be exempted as forwarding"
}
used_forwarders = used_forwarders + [forwarding]
forwarded += 1
} else if read.name == "" || !contains(names, read.name) {
failures = failures
+ [read.file + ":" + to_string(read.line) + " " + read.scope + " " + read.expression]
}
}
for index in range(0, len(forwarders)) {
const boundary = forwarders[index]
if !contains(used_forwarders, index) {
failures = failures
+ ["stale forwarding boundary " + boundary.file + " " + boundary.expression]
}
}
for index in range(0, len(non_behavior_reads)) {
const fact = non_behavior_reads[index]
if !contains(used_exclusions, index) {
failures = failures
+ ["stale domain fact classification " + fact.file + " " + fact.expression]
}
}
return {failures: failures, forwarded: forwarded, non_behavior: non_behavior}
}
/** Counts distinguish registration failures from unresolved reachability and policy review. */
pub type AgentGateAudit = {
entries: int,
reads: int,
forwarded: int,
non_behavior: int,
reachability_unknown: int,
retained_for_review: int,
unread_entries: list<string>,
unread_binding_scopes: list<string>,
pending: int,
failures: list<string>,
}
/**
* Audit all configured sources and check or regenerate their review tables.
* Invalid source or registry input throws; a complete audit reports every
* unregistered read and stale projection in `failures`.
*
* @effects: [host]
* @errors: [validation, backend]
* @api_stability: experimental
*/
pub fn agent_gate_audit(
harness: {fs: HarnessFs, ast: HarnessAst, process: HarnessProcess},
registry_path: string,
write: bool = false,
) -> AgentGateAudit {
const registry = resolve_binding_sources(
harness.fs,
schema_expect(json_parse(harness.fs.read_text(registry_path)), schema_of(AgentGateRegistry)),
)
if registry.schema_version != 1 || len(registry.scopes) + len(registry.binding_scopes) == 0 {
throw "agent gate registry: unsupported schema or empty scopes"
}
let prefixes: list<string> = []
for alias in registry.name_prefixes {
if alias.from == "" || alias.to == "" || !alias.from.ends_with(".")
|| !alias.to.ends_with(".")
|| contains(prefixes, alias.from) {
throw "agent gate registry: invalid or duplicate configuration owner prefix ${alias.from}"
}
prefixes = prefixes + [alias.from]
}
let scope_ids: list<string> = []
for scope in registry.scopes {
if scope.id.trim() == "" || contains(scope_ids, scope.id) {
throw "agent gate registry: empty or duplicate scope id ${scope.id}"
}
scope_ids = scope_ids + [scope.id]
}
for scope in registry.binding_scopes {
if scope.config.id.trim() == "" || contains(scope_ids, scope.config.id) {
throw "agent gate registry: empty or duplicate scope id ${scope.config.id}"
}
scope_ids = scope_ids + [scope.config.id]
}
const projected_layers = registry.projections.map(fn(projection) { return projection.layer })
.sorted()
if projected_layers != ["runner", "stall handler", "stop decision"] {
throw "agent gate registry: exactly one projection per layer is required"
}
const entries = read_entries(harness.fs, registry)
const forwarders = read_classifications(harness.fs, registry.forwarder_files)
const non_behavior = read_classifications(harness.fs, registry.non_behavior_files)
validate_classified_consumers(registry, forwarders)
validate_classified_consumers(registry, non_behavior)
const reads = agent_gate_reads(
{ast: harness.ast, fs: harness.fs, process: harness.process},
registry,
)
const flag_scopes = registry.scopes.filter(fn(scope) { return scope.kind == "flag" }).map(
fn(scope) { return scope.id },
)
const findings = agent_gate_findings(entries, reads, forwarders, non_behavior, flag_scopes)
let failures = findings.failures
const observed = observed_entries(
entries,
reads.filter(
fn(read) { return classification_index(read, forwarders) == nil
&& classification_index(read, non_behavior) == nil },
),
)
if write {
write_observed_entries(harness.fs, registry.entry_files, observed)
} else {
for index in range(0, len(entries)) {
const expected_entry = entries[index]
const observed_entry = observed[index]
if expected_entry == nil || observed_entry == nil {
throw "missing observed entry"
}
if expected_entry.readers != observed_entry.readers {
failures = failures + ["stale readers " + expected_entry.name]
}
}
}
for projection in registry.projections {
const expected = agent_gate_markdown(registry.repo, projection.layer, observed)
if write {
harness.fs.write_text(projection.file, expected)
} else if harness.fs.read_text(projection.file) != expected {
failures = failures + ["stale projection " + projection.file]
}
}
const unread_entries = observed.filter(fn(entry) { return len(entry.readers) == 0 }).map(
fn(entry) { return entry.name },
)
const unread_owners = registry.binding_scopes.filter(
fn(scope) { return len(reads.filter(fn(read) { return read.scope == scope.config.id })) == 0 },
)
.map(fn(scope) { return scope.config.id })
return {
entries: len(entries),
reads: len(reads),
forwarded: findings.forwarded,
non_behavior: findings.non_behavior,
reachability_unknown: len(entries.filter(fn(entry) { return entry.reachability == "unknown" })),
retained_for_review: len(
entries.filter(fn(entry) { return entry.verdict == "KEEP-with-expiry" }),
),
unread_entries: unread_entries,
unread_binding_scopes: unread_owners,
pending: len(failures),
failures: failures,
}
}