PromptRunner.Control.Store (PromptRunnerSDK v0.10.0)

Copy Markdown View Source

The file transport under .prompt_runner/control/.

control/
  requests/     one file per command, consumed and deleted
  log.jsonl     append-only: every command, who, when, outcome
  snapshot.json rewritten on each event batch
  events.jsonl  append-only canonical event stream for subscribers

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

Functions

append_event(packet_dir, event)
append_log(packet_dir, entry)
control_dir(packet_dir)
events_path(packet_dir)
init(packet_dir)
log_path(packet_dir)
read_log(packet_dir)
read_snapshot(packet_dir)
requests_dir(packet_dir)
reset_events(packet_dir)

Truncates the subscriber event stream.

snapshot_path(packet_dir)
take_requests(packet_dir)

Reads and deletes every pending request, in arrival order.

write_request(packet_dir, request)

Writes one request file.

write_snapshot(packet_dir, snapshot)

Rewrites snapshot.json.

Functions

append_event(packet_dir, event)

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

append_log(packet_dir, entry)

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

control_dir(packet_dir)

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

events_path(packet_dir)

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

init(packet_dir)

@spec init(String.t()) :: :ok

log_path(packet_dir)

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

read_log(packet_dir)

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

read_snapshot(packet_dir)

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

requests_dir(packet_dir)

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

reset_events(packet_dir)

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

Truncates the subscriber event stream.

Called once when a run starts. The stream is scoped to the run in flight, so a subscriber reading from the beginning gets this run rather than a concatenation of every run the packet has ever had.

snapshot_path(packet_dir)

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

take_requests(packet_dir)

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

Reads and deletes every pending request, in arrival order.

Deleting before returning is deliberate: a request that cannot be parsed, or whose handler raises, must not be retried on the next boundary forever. It is logged and gone.

write_request(packet_dir, request)

@spec write_request(String.t(), 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(packet_dir, snapshot)

@spec write_snapshot(String.t(), 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.