/**
* std/verification — deterministic verification facts and helper shapes.
*
* Import: import "std/verification"
*
* This module is the Harn-owned home for reusable verification intelligence
* primitives. It intentionally starts with fact capture, not policy: callers
* can bind diagnostics and background check results to file snapshots without
* duplicating hostlib plumbing or inventing product-specific hash logic.
*/
import { command_cancel, command_run, command_wait } from "std/command"
import { shell_quote } from "std/runtime"
import { regex_first_capture, truncate_text } from "std/text"
import "std/verification_core"
import "std/verification_targets"
import "std/verification_types"
pub fn __verification_profile_row_id(row, index: int) -> string {
const id = trim(to_string(row?.id ?? row?.name ?? ""))
if id != "" {
return id
}
return "profile/" + to_string(index)
}
pub fn __verification_check(row) -> dict {
if type_of(row?.check) == "dict" {
return row.check
}
return {}
}
pub fn __verification_rung_text(value) -> string {
return uppercase(trim(to_string(value ?? "")))
}
pub fn __verification_rung_rank(value) {
const rung = __verification_rung_text(value)
const direct = to_int(rung)
if direct != nil && direct >= 0 && direct <= 5 {
return direct
}
if len(rung) == 2 && starts_with(rung, "R") {
const parsed = to_int(substring(rung, 1))
if parsed != nil && parsed >= 0 && parsed <= 5 {
return parsed
}
}
return nil
}
pub fn __verification_rung_bound(value, fallback: int) -> int {
const rank = __verification_rung_rank(value)
return rank ?? fallback
}
pub fn __verification_resource_class(row, check) -> string {
const raw = check?.resourceClass ?? check?.resource_class ?? row?.resourceClass
?? row?.resource_class
?? "moderate"
const value = lowercase(trim(to_string(raw)))
return value == "" ? "moderate" : value
}
pub fn __verification_resource_rank(value) -> int {
const resource = lowercase(trim(to_string(value ?? "")))
if resource == "cheap" {
return 0
}
if resource == "moderate" {
return 1
}
if resource == "heavy" {
return 2
}
if resource == "remote_ok" || resource == "remote-ok" || resource == "remote" {
return 3
}
return 4
}
pub fn __verification_lower_string_list(value) -> list<string> {
let out: list<string> = []
for item in __verification_string_list(value) {
const text = lowercase(item)
if !contains(out, text) {
out = out + [text]
}
}
return out
}
pub fn __verification_resource_allowed(resource: string, opts) -> bool {
const allowed = __verification_lower_string_list(
opts?.resource_classes ?? opts?.allowed_resource_classes,
)
if len(allowed) == 0 {
return true
}
return contains(allowed, lowercase(resource))
}
pub fn __verification_warm_state_fact_for(row_id: string, opts) {
const facts = opts?.warm_state_facts ?? opts?.warm_states ?? []
if type_of(facts) == "dict" {
return facts[row_id]
}
for fact in facts {
const id = trim(to_string(fact?.row_id ?? fact?.id ?? ""))
if id == row_id {
return fact
}
}
return nil
}
pub fn __verification_row_timing_kind(row_id: string, row, opts, warm_state) -> string {
const explicit = lowercase(trim(to_string(opts?.timing_kind ?? "")))
if explicit != "" && explicit != "auto" {
return explicit
}
if warm_state != nil {
return warm_state?.warm ?? false ? "warm" : "cold"
}
const mode = __verification_warm_state_mode(row)
if mode?.warm != nil {
return mode.warm ? "warm" : "cold"
}
if mode?.ready != nil {
return mode.ready ? "warm" : "cold"
}
return "warm"
}
pub fn __verification_row_p95_ms(row, opts, timing_kind: string) -> int {
const kind = lowercase(trim(to_string(timing_kind)))
const timings = row?.timings ?? {}
const raw = if kind == "cold" {
timings?.coldMs?.p95
} else if kind == "any" {
timings?.warmMs?.p95 ?? timings?.coldMs?.p95
} else {
timings?.warmMs?.p95 ?? timings?.coldMs?.p95
}
return to_int(raw) ?? to_int(opts?.unknown_p95_ms) ?? 999999
}
pub fn __verification_row_command(row, check) {
return check?.command ?? check?.spec ?? row?.command ?? row?.spec
}
pub fn __verification_command_present(command) -> bool {
if command == nil {
return false
}
if type_of(command) == "string" {
return trim(command) != ""
}
return true
}
pub fn __verification_skip(row_id: string, reason: string, profile_match) -> dict {
return {
row_id: row_id,
reason: reason,
specificity: profile_match?.specificity ?? 0,
index: profile_match?.index ?? 0,
}
}
pub fn __verification_ladder_candidate(profile_match, opts) {
const row = profile_match?.row ?? {}
const row_id = __verification_profile_row_id(row, to_int(profile_match?.index) ?? 0)
const check = __verification_check(row)
const rung_rank = __verification_rung_rank(check?.rung ?? row?.rung)
if rung_rank == nil {
return {entry: nil, skip: __verification_skip(row_id, "missing_or_invalid_rung", profile_match)}
}
const rung = "R" + to_string(rung_rank)
const min_rank = __verification_rung_bound(opts?.min_rung, 0)
const max_rank = __verification_rung_bound(opts?.max_rung, 5)
if rung_rank < min_rank {
return {entry: nil, skip: __verification_skip(row_id, "below_min_rung", profile_match)}
}
if rung_rank > max_rank {
return {entry: nil, skip: __verification_skip(row_id, "above_max_rung", profile_match)}
}
const resource = __verification_resource_class(row, check)
if !__verification_resource_allowed(resource, opts) {
return {entry: nil, skip: __verification_skip(row_id, "resource_class_filtered", profile_match)}
}
if opts?.allow_stale_prone == false
&& (check?.staleProne ?? check?.stale_prone ?? row?.staleProne
?? row
?.stale_prone
?? false) {
return {entry: nil, skip: __verification_skip(row_id, "stale_prone_filtered", profile_match)}
}
const command = __verification_row_command(row, check)
if !__verification_command_present(command) {
return {entry: nil, skip: __verification_skip(row_id, "missing_command", profile_match)}
}
const warm_state = __verification_warm_state_fact_for(row_id, opts)
const timing_kind = __verification_row_timing_kind(row_id, row, opts, warm_state)
const p95_ms = __verification_row_p95_ms(row, opts, timing_kind)
const specificity = to_int(profile_match?.specificity) ?? 0
const index = to_int(profile_match?.index) ?? 0
const sort_key = [
rung_rank,
__verification_resource_rank(resource),
p95_ms,
0 - specificity,
index,
]
return {
entry: {
row_id: row_id,
row: row,
rung: rung,
rung_rank: rung_rank,
command: command,
runnable: true,
granularity: check?.granularity ?? row?.granularity,
trust: check?.trust ?? check?.trust_level ?? row?.trust ?? row?.trust_level,
resource_class: resource,
p95_ms: p95_ms,
timing_kind: timing_kind,
warm_state: warm_state,
specificity: specificity,
index: index,
sort_key: sort_key,
background: check?.background ?? row?.background ?? false,
},
skip: nil,
}
}
pub fn __verification_duration_label(ms_value) -> string {
let ms = to_int(ms_value) ?? 0
if ms < 0 {
ms = 0
}
const seconds = ms / 1000
if seconds < 60 {
return to_string(seconds) + "s"
}
const minutes = seconds / 60
const remainder = seconds % 60
return to_string(minutes) + "m" + to_string(remainder) + "s"
}
pub fn __verification_command_label(command) -> string {
if command == nil {
return ""
}
if type_of(command) == "string" {
return trim(to_string(command))
}
if type_of(command) == "dict" {
const shell = trim(to_string(command?.command ?? ""))
if shell != "" {
return shell
}
const argv = command?.argv ?? command?.args
if type_of(argv) == "list" {
let parts: list<string> = []
for part in argv {
parts = parts + [to_string(part)]
}
return trim(parts.join(" "))
}
}
return trim(json_stringify(command))
}
pub fn __verification_hud_query(state, opts) -> dict {
let query: dict = {}
const path = trim(to_string(state?.path ?? state?.file ?? ""))
if path != "" {
query = query + {path: path}
}
const language = trim(to_string(state?.language ?? ""))
if language != "" {
query = query + {language: language}
}
const task = trim(to_string(state?.task ?? state?.task_kind ?? ""))
if task != "" {
query = query + {task: task}
}
if type_of(state?.query) == "dict" {
query = query + state.query
}
if type_of(opts?.query) == "dict" {
query = query + opts.query
}
return query
}
pub fn __verification_hud_plan_options(dir: string, opts) -> dict {
let plan_options = opts?.plan_options ?? {}
if plan_options?.dir == nil {
plan_options = plan_options + {dir: dir}
}
if opts?.resource_classes != nil && plan_options?.resource_classes == nil {
plan_options = plan_options + {resource_classes: opts.resource_classes}
}
if opts?.allow_stale_prone != nil && plan_options?.allow_stale_prone == nil {
plan_options = plan_options + {allow_stale_prone: opts.allow_stale_prone}
}
if opts?.timing_kind != nil && plan_options?.timing_kind == nil {
plan_options = plan_options + {timing_kind: opts.timing_kind}
}
if opts?.warm_state_facts != nil && plan_options?.warm_state_facts == nil {
plan_options = plan_options + {warm_state_facts: opts.warm_state_facts}
}
return plan_options
}
pub fn __verification_hud_entry(entry) {
if type_of(entry) != "dict" {
return nil
}
const command = __verification_command_label(entry?.command)
return {
row_id: entry?.row_id,
rung: entry?.rung,
command: command,
label: trim(to_string(entry?.row?.label ?? entry?.row?.name ?? entry?.row_id ?? "")),
p95_ms: entry?.p95_ms,
p95: __verification_duration_label(entry?.p95_ms),
timing_kind: entry?.timing_kind ?? "warm",
resource_class: entry?.resource_class,
granularity: entry?.granularity,
trust: entry?.trust,
row: entry?.row,
}
}
pub fn __verification_hud_last_check(state, current_hashes, now_ms: int) {
const last = state?.last_check ?? state?.lastCheck
if type_of(last) != "dict" {
return nil
}
const at_ms = to_int(last?.at_ms ?? last?.atMs ?? last?.ended_ms ?? last?.started_ms ?? now_ms)
?? now_ms
const age_ms = now_ms - at_ms
const snapshot = last?.bound_snapshot ?? last?.boundSnapshot ?? last?.snapshot
const freshness = if type_of(snapshot) == "dict" && type_of(current_hashes) == "dict" {
verification_snapshot_staleness(snapshot, current_hashes)
} else if type_of(snapshot) == "dict" {
{
schema: "harn.verification.snapshot_staleness.v1",
status: "unknown_current_hashes",
stale: false,
fresh: false,
staleFiles: [],
stale_files: [],
changed_files: [],
advisory: true,
reason: "current file hashes unavailable; freshness unknown",
}
} else {
{
schema: "harn.verification.snapshot_staleness.v1",
status: "unbound",
stale: false,
fresh: false,
staleFiles: [],
stale_files: [],
changed_files: [],
advisory: true,
reason: "last check has no snapshot binding",
}
}
return {
rung: to_string(last?.rung ?? "R?"),
verdict: to_string(last?.verdict ?? last?.status ?? "?"),
at_ms: at_ms,
age_ms: age_ms,
age: __verification_duration_label(age_ms),
freshness: freshness,
}
}
pub fn __verification_hud_last_line(last) -> string {
const freshness = last?.freshness ?? {}
const suffix = if freshness?.status == "bound_stale" {
" - STALE (" + to_string(len(freshness?.stale_files ?? [])) + " file(s) changed since)"
} else if freshness?.status == "bound_fresh" {
" (current snapshot)"
} else {
" (freshness unknown)"
}
return "last check: " + to_string(last?.rung ?? "R?")
+ " -> "
+ to_string(last?.verdict ?? "?")
+ " @"
+ to_string(last?.age ?? "0s")
+ " ago"
+ suffix
}
pub fn __verification_hud_next_line(label: string, entry) -> string {
return label + ": " + to_string(entry?.rung ?? "R?")
+ " "
+ to_string(entry?.command ?? "")
+ " - est "
+ to_string(entry?.p95 ?? "0s")
+ " (p95 "
+ to_string(entry?.timing_kind ?? "warm")
+ ")"
}
/**
* Capture current on-disk hashes for `paths` under one code-index sequence
* binding. Unknown-but-readable workspace files are included with
* `known = false`; unreadable or out-of-workspace paths return a null hash
* and appear in `missing`. The `snapshot` field is the direct path->hash map
* consumed by `verification_diagnostic_classify`.
*
* @effects: [host, fs]
* @errors: [backend, invalid_argument]
* @api_stability: experimental
* @example: verification_file_hash_snapshot(["src/main.zig", "build.zig"])
*/