PromptRunner.Control.Store (PromptRunnerSDK v0.12.1)

Copy Markdown View Source

The file transport for one control-plane state root.

control/
  requests/     one file per command, retained until an outcome is durable
  outcomes/     immutable request outcome receipts
  log.jsonl     append-only: every command, who, when, outcome
  snapshot.json atomically replaced on each event batch
  events.jsonl  append-only canonical event stream for subscribers
  events/       append-only streams archived by run id

A directory rather than a socket, for the first transport: no daemon, no port, no supervision tree to get wrong; it works under tee, nohup, tmux, and with no terminal at all; it survives the runner dying, because the requests are just sitting there; and it is trivially inspectable when something goes wrong.

It also forces the API to be serialisable and asynchronous from day one, which is the discipline that keeps the CLI from quietly becoming privileged.

Summary

Types

A legacy packet root or an explicit runtime state root.

Functions

Persists an immutable request outcome, then removes the pending request.

Reads every pending request without deleting it.

Prepares the subscriber event stream for run_id without truncating bytes.

Legacy compatibility hook.

Tags an operator-owned runtime directory for use as the control state root.

Reads and deletes every pending request, in arrival order.

Writes one request file.

Rewrites snapshot.json.

Types

root()

@type root() :: String.t() | {:state_root, String.t()}

A legacy packet root or an explicit runtime state root.

Functions

append_event(root, event)

@spec append_event(root(), map()) :: :ok | {:error, term()}

append_log(root, entry)

@spec append_log(root(), PromptRunner.Control.Entry.t()) :: :ok | {:error, term()}

archived_events_path(root, run_id)

@spec archived_events_path(root(), String.t()) :: String.t()

complete_request(root, name, outcome)

@spec complete_request(root(), String.t(), map()) :: :ok | {:error, term()}

Persists an immutable request outcome, then removes the pending request.

The ordering is the durability guarantee: a crash may leave both files, but can never remove the request before its outcome exists.

control_dir(packet_dir)

@spec control_dir(root()) :: String.t()

events_path(root)

@spec events_path(root()) :: String.t()

init(root)

@spec init(root()) :: :ok | {:error, term()}

log_path(root)

@spec log_path(root()) :: String.t()

outcomes_dir(root)

@spec outcomes_dir(root()) :: String.t()

pending_requests(root)

@spec pending_requests(root()) :: [{String.t(), {:ok, map()} | {:error, term()}}]

Reads every pending request without deleting it.

Requests with an existing durable outcome receipt are acknowledged and skipped. This closes the crash window between writing the receipt and removing the inbox file without applying the request twice.

prepare_event_stream(root, run_id)

@spec prepare_event_stream(root(), String.t()) :: :ok | {:error, term()}

Prepares the subscriber event stream for run_id without truncating bytes.

Reopening the same run leaves the current stream in place and continues appending. Opening a different run atomically archives the current stream by its previous run id. A subscriber reading the canonical path therefore sees only the current run, while prior streams remain available for audit.

read_log(root)

@spec read_log(root()) :: {:ok, [PromptRunner.Control.Entry.t()]}

read_snapshot(root)

@spec read_snapshot(root()) ::
  {:ok, PromptRunner.Control.Snapshot.t()} | {:error, term()}

request_outcome_path(root, name)

@spec request_outcome_path(root(), String.t()) :: String.t()

requests_dir(root)

@spec requests_dir(root()) :: String.t()

reset_events(root)

@spec reset_events(root()) :: :ok | {:error, term()}

Legacy compatibility hook.

Event streams are append-only now, so resetting is intentionally a no-op. New code should call prepare_event_stream/2.

snapshot_path(root)

@spec snapshot_path(root()) :: String.t()

state_root(path)

@spec state_root(String.t()) :: {:state_root, String.t()}

Tags an operator-owned runtime directory for use as the control state root.

take_requests(root)

@spec take_requests(root()) :: [{String.t(), {:ok, map()} | {:error, term()}}]

Reads and deletes every pending request, in arrival order.

This preserves the legacy Store API. The control plane no longer uses it, because it cannot provide durable request outcomes. New consumers should use pending_requests/1 and complete_request/3.

write_request(root, request)

@spec write_request(root(), map()) :: {:ok, String.t()} | {:error, term()}

Writes one request file.

The name carries a sortable timestamp and a unique suffix, so requests are consumed in arrival order and two writers racing cannot collide on a name.

write_snapshot(root, snapshot)

@spec write_snapshot(root(), PromptRunner.Control.Snapshot.t()) ::
  :ok | {:error, term()}

Rewrites snapshot.json.

Written to a temporary file and renamed, because a reader polling this file must never observe a half-written one. A rename within a directory is atomic on every filesystem this runs on.