// std/fs - file-system convenience helpers built on the host fs primitives.
import { pretty, safe_parse } from "std/json"
type WriteDataOptions = {pretty?: bool, trailing_newline?: bool, ensure_parent?: bool}
fn __fs_norm(path) -> string {
return replace(path ?? "", "\\", "/")
}
fn __fs_parent(path) -> string {
return dirname(path ?? "")
}
fn __fs_with_newline(text, trailing_newline) {
if !(trailing_newline ?? true) || ends_with(text, "\n") {
return text
}
return text + "\n"
}
/**
* Create the parent directory for `path` when it is not `.` or empty.
*
* @effects: []
* @errors: []
*/
pub fn ensure_parent_dir(path: string) {
let parent = __fs_parent(path)
if parent != "" && parent != "." {
harness.fs.mkdir(parent)
}
}
/**
* Read and parse a JSON file, returning `fallback` for missing or invalid files.
*
* @effects: []
* @errors: []
*/
pub fn read_json(path: string, fallback = nil) {
if !harness.fs.exists(path) {
return fallback
}
let parsed = safe_parse(harness.fs.read_text(path))
if parsed == nil {
return fallback
}
return parsed
}
/**
* Read and parse a JSON file, returning `{ok, value?, error?}`.
*
* @effects: []
* @errors: []
*/
pub fn read_json_result(path: string) -> dict {
if !harness.fs.exists(path) {
return {ok: false, error: "file not found: " + path}
}
let parsed = try {
json_parse(harness.fs.read_text(path))
}
if is_ok(parsed) {
return {ok: true, value: unwrap(parsed)}
}
return {ok: false, error: to_string(parsed)}
}
/**
* Write a JSON file with optional pretty formatting and parent-dir creation.
*
* @effects: []
* @errors: []
*/
pub fn write_json(path: string, value, options: WriteDataOptions = {}) {
let opts = options ?? {}
if opts.ensure_parent ?? true {
ensure_parent_dir(path)
}
let body = if opts.pretty ?? false {
pretty(value)
} else {
json_stringify(value)
}
harness.fs.write_text(path, __fs_with_newline(body, opts.trailing_newline ?? true))
}
/**
* Read and parse a YAML file, returning `fallback` for missing or invalid files.
*
* @effects: []
* @errors: []
*/
pub fn read_yaml(path: string, fallback = nil) {
if !harness.fs.exists(path) {
return fallback
}
let parsed = try {
yaml_parse(harness.fs.read_text(path))
}
if is_ok(parsed) {
return unwrap(parsed)
}
return fallback
}
/**
* Write a YAML file.
*
* @effects: []
* @errors: []
*/
pub fn write_yaml(path: string, value, options: WriteDataOptions = {}) {
let opts = options ?? {}
if opts.ensure_parent ?? true {
ensure_parent_dir(path)
}
harness.fs
.write_text(path, __fs_with_newline(yaml_stringify(value), opts.trailing_newline ?? true))
}
/**
* Read and parse a TOML file, returning `fallback` for missing or invalid files.
*
* @effects: []
* @errors: []
*/
pub fn read_toml(path: string, fallback = nil) {
if !harness.fs.exists(path) {
return fallback
}
let parsed = try {
toml_parse(harness.fs.read_text(path))
}
if is_ok(parsed) {
return unwrap(parsed)
}
return fallback
}
/**
* Write a TOML file.
*
* @effects: []
* @errors: []
*/
pub fn write_toml(path: string, value, options: WriteDataOptions = {}) {
let opts = options ?? {}
if opts.ensure_parent ?? true {
ensure_parent_dir(path)
}
harness.fs
.write_text(path, __fs_with_newline(toml_stringify(value), opts.trailing_newline ?? true))
}
/**
* Write a list of lines as a text file.
*
* @effects: []
* @errors: []
*/
pub fn write_lines(path: string, lines: list<string>, options: WriteDataOptions = {}) {
let opts = options ?? {}
if opts.ensure_parent ?? true {
ensure_parent_dir(path)
}
harness.fs
.write_text(path, __fs_with_newline(join(lines ?? [], "\n"), opts.trailing_newline ?? true))
}
/**
* Append one line to a text file.
*
* @effects: []
* @errors: []
*/
pub fn append_line(path: string, line: string) {
let text = line ?? ""
harness.fs.append(path, text + "\n")
}
/**
* Create a file if missing and update it to empty content only on first creation.
*
* @effects: []
* @errors: []
*/
pub fn touch(path: string) {
if !harness.fs.exists(path) {
ensure_parent_dir(path)
harness.fs.write_text(path, "")
}
}
/**
* Return files matching `pattern` below `root`, using the host glob implementation.
*
* @effects: []
* @errors: []
*/
pub fn find_files(root: string, pattern: string, options = {}) -> list<string> {
let matches = harness.fs.glob(pattern, {base: root, long_running: options?.long_running ?? false})
if options?.relative ?? false {
return matches.map({ path -> relative_path(root, path) }).to_list()
}
return matches
}
/**
* Return `path` relative to `root` when it is inside that root.
*
* @effects: []
* @errors: []
*/
pub fn relative_path(root: string, path: string) -> string {
let base_raw = __fs_norm(root)
let target = __fs_norm(path)
let base = if ends_with(base_raw, "/") {
base_raw
} else {
base_raw + "/"
}
if starts_with(target, base) {
return substring(target, len(base))
}
return target
}
/**
* Return whether `path` exists and is a regular file.
*
* @effects: []
* @errors: []
*/
pub fn is_file(path: string) -> bool {
let info = try {
harness.fs.stat(path)
}
if !is_ok(info) {
return false
}
return unwrap(info)?.is_file ?? false
}
/**
* Return whether `path` exists and is a directory.
*
* @effects: []
* @errors: []
*/
pub fn is_dir(path: string) -> bool {
let info = try {
harness.fs.stat(path)
}
if !is_ok(info) {
return false
}
return unwrap(info)?.is_dir ?? false
}
/**
* Return a file size in bytes, or nil when the path cannot be statted.
*
* @effects: []
* @errors: []
*/
pub fn file_size(path: string) {
let info = try {
harness.fs.stat(path)
}
if is_ok(info) {
return unwrap(info)?.size
}
return nil
}
/**
* Workspace snapshots — thin pipeline-facing wrappers over the host
* `hostlib_fs_*` snapshot builtins (crates/harn-hostlib/src/fs_snapshot.rs).
* They capture the pre-image of explicit paths so a pipeline can checkpoint
* the workspace and roll it back between independent retry attempts — the
* substrate for a verify-gated best-of-N agent loop.
*
* Snapshots are session-scoped: the session id defaults to the active agent
* session (`agent_session_current_id()`), so each conversation's snapshots
* stay isolated and are cleaned up when the host closes the session. Callers
* outside an agent session must pass an explicit `session_id`. This helper
* resolves and validates that session id.
*
* @effects: [host]
* @errors: []
*/
fn __fs_snapshot_session(opts) -> string {
let session = opts?.session_id ?? agent_session_current_id()
if session == nil || session == "" {
throw "fs_snapshot: no active agent session; pass an explicit `session_id`"
}
return session
}
/**
* Capture a workspace snapshot of `paths`, returning `{snapshot_id,
* captured_paths, byte_count}`.
*
* `opts.session_id` defaults to the active agent session. `opts.snapshot_id`
* defaults to a fresh `snapshot-<uuid>` id so independent attempts never
* collide. `paths` are captured immediately; restoring later reinstates
* their pre-image (including deleting paths that were absent at capture).
*
* @effects: [host, fs]
* @errors: [backend]
* @api_stability: experimental
* @example: let snap = fs_snapshot(["src/lib.rs"]); fs_restore(snap.snapshot_id)
*/
pub fn fs_snapshot(paths: list<string> = [], opts = {}) -> dict {
let session = __fs_snapshot_session(opts)
let scope = opts?.snapshot_id ?? ("snapshot-" + uuid_v7())
return hostlib_fs_snapshot({session_id: session, scope_id: scope, paths: paths ?? [], root: opts?.root})
}
/**
* Restore a previously-captured snapshot, returning `{snapshot_id,
* restored_paths, skipped_paths_with_reasons}`. With an empty `paths` every
* captured path is restored; otherwise only the listed subset.
*
* @effects: [host, fs]
* @errors: [backend]
* @api_stability: experimental
* @example: fs_restore(snap.snapshot_id)
*/
pub fn fs_restore(snapshot_id: string, paths: list<string> = [], opts = {}) -> dict {
let session = __fs_snapshot_session(opts)
return hostlib_fs_restore({session_id: session, snapshot_id: snapshot_id, paths: paths ?? []})
}
/**
* List the snapshots registered for the session, oldest first. Each entry is
* `{snapshot_id, scope_id, taken_at_ms, captured_paths, byte_count}`.
*
* @effects: [host]
* @errors: [backend]
* @api_stability: experimental
* @example: for snap in fs_list_snapshots() { log(snap.snapshot_id) }
*/
pub fn fs_list_snapshots(opts = {}) -> list<dict> {
let session = __fs_snapshot_session(opts)
let result = hostlib_fs_list_snapshots({session_id: session})
return result?.snapshots ?? []
}
/**
* Drop a snapshot's in-memory and on-disk state, returning `{snapshot_id,
* dropped}`. Idempotent: dropping an unknown id reports `dropped: false`.
* The best-of-N loop calls this to release each attempt's checkpoint.
*
* @effects: [host, fs]
* @errors: [backend]
* @api_stability: experimental
* @example: fs_drop_snapshot(snap.snapshot_id)
*/
pub fn fs_drop_snapshot(snapshot_id: string, opts = {}) -> dict {
let session = __fs_snapshot_session(opts)
return hostlib_fs_drop_snapshot({session_id: session, snapshot_id: snapshot_id})
}