Struct ServerState 

pub struct ServerState {

    pub journal_dir: PathBuf,
    pub sessions: Mutex<HashMap<String, Arc<ClientSession>>>,
    pub inference: OnceLock<Arc<InferenceEngine>>,
    pub host: Arc<HostState>,
    pub shared_memgine: Option<Arc<Mutex<MemgineEngine>>>,
    pub voice_sessions: Arc<VoiceSessionRegistry>,
    pub meetings: Arc<MeetingRegistry>,
    pub a2ui: A2uiSurfaceStore,
    pub ui_agent: Arc<UIImprovementAgent>,
    pub ui_agent_oscillation: Arc<OscillationDetector>,
    pub ui_agent_budget: Arc<IterationBudget>,
    pub admission: Arc<InferenceAdmission>,
    pub a2ui_route_auth: Mutex<HashMap<String, A2aRouteAuth>>,
    pub supervisor: OnceLock<Arc<Supervisor>>,
    pub observer_manifest_path: OnceLock<PathBuf>,
    pub a2a_dispatcher: OnceLock<Arc<A2aDispatcher>>,
    pub a2ui_subscribers: Mutex<HashMap<String, Arc<WsChannel>>>,
    pub auth_token: OnceLock<String>,
    pub host_token: OnceLock<String>,
    pub parslee_session: OnceLock<ParsleeSession>,
    pub attached_agents: Mutex<HashMap<String, String>>,
    pub agent_memgines: Mutex<HashMap<String, Arc<Mutex<MemgineEngine>>>>,
    pub chat_sessions: Mutex<HashMap<String, ChatSession>>,
    pub runs: Mutex<HashMap<String, RunMeta>>,
    pub run_subscribers: Mutex<HashMap<(String, String), RunTraceSubscriber>>,
    pub run_store: RunStore,
    pub mcp_url: OnceLock<String>,
    pub mcp_sessions: OnceLock<Arc<SessionMap>>,
    pub approval_gate: ApprovalGate,
    /* private fields */
}

Global server state shared across all connections.

Fields

journal_dir: PathBuf

sessions: Mutex<HashMap<String, Arc<ClientSession>>>

inference: OnceLock<Arc<InferenceEngine>>

host: Arc<HostState>

shared_memgine: Option<Arc<Mutex<MemgineEngine>>>

When Some, create_session clones this handle into every new ClientSession.memgine — embedders that want a single shared memgine across all WS sessions set this. Standalone car-server leaves it None, which gives each session its own engine (preserving today’s behavior).

voice_sessions: Arc<VoiceSessionRegistry>

Process-wide voice session registry. Each voice.transcribe_stream.start call registers its own per-client WsVoiceEventSink so events route back to the originating WS connection only.

meetings: Arc<MeetingRegistry>

Process-wide meeting registry. Meeting ids are global; each meeting binds to the originating client’s WS for upstream events but persists transcripts to the resolved .car/meetings/<id>/ regardless of which client started it.

a2ui: A2uiSurfaceStore

Process-wide A2UI surface store. Agent-produced surfaces are visible to every host UI subscriber, independent of the WebSocket session that applied the update.

ui_agent: Arc<UIImprovementAgent>

In-process UI-improvement agent. Invoked from handle_a2ui_render_report with each inbound report; returned Decision::Patch envelopes are applied via the standard apply_a2ui_envelope path so all subscribers see the patch. Arc so the agent’s interior DashMap state survives across handler calls even when ServerState is cheap-cloned.

ui_agent_oscillation: Arc<OscillationDetector>

Per-surface oscillation detector for the UI-improvement loop. Sits between the agent’s Decision::Patch and the apply path so A→B→A patch cycles get cooled down without the agent itself having to track history. neo’s review: “controllers use workqueue backoff; reconcilers stay stateless.”

ui_agent_budget: Arc<IterationBudget>

Per-surface iteration budget. Backstop against runaway loops the oscillation detector misses — caps total agent- driven patches per surface at DEFAULT_MAX_ITERATIONS.

admission: Arc<InferenceAdmission>

Process-wide concurrency gate for inference RPC handlers. Sized from host RAM at startup, overridable via crate::admission::ENV_MAX_CONCURRENT. Without this, N concurrent users multiply KV-cache and activation memory and take the host out (#114-adjacent: filed alongside the daemon always-on rework). The semaphore lives on ServerState so it is shared across every WebSocket session in the same process.

a2ui_route_auth: Mutex<HashMap<String, A2aRouteAuth>>

Server-side A2A continuation auth keyed by A2UI surface id. Kept out of A2uiSurface.owner so host renderers never see bearer/API-key material.

supervisor: OnceLock<Arc<Supervisor>>

Lifecycle-managed agents — declarative manifest at ~/.car/agents.json driving spawn/restart/stop. Closes Parslee-ai/car-releases#27. Lazy-initialized so embedders that don’t want process supervision don’t pay the disk-touch cost at server start.

observer_manifest_path: OnceLock<PathBuf>

Manifest path this daemon is observing but does NOT own. Set by car-server when boot-time supervisor construction fails with car_registry::supervisor::SupervisorError::AlreadyRunning — another car-server process on the host holds the exclusive lock on this manifest. In that state, supervisor() returns a clear “observe-only” error so mutation handlers refuse (preventing the duplicate-spawn bug from Parslee-ai/car-releases#44), while read-only handlers (agents.list, agents.health) fall back to car_registry::supervisor::Supervisor::list_from_manifest / car_registry::supervisor::Supervisor::health_from_manifest so operators can still inspect what the primary daemon is supervising.

a2a_dispatcher: OnceLock<Arc<A2aDispatcher>>

In-core A2A dispatcher — embedders that consume car-server-core get A2A reachability “for free” without standing up a separate HTTP listener. Closes Parslee-ai/car-releases#28. Lazy-init so the embedder can override the runtime / task store / agent card via ServerStateConfig::with_a2a_runtime etc. before the first dispatch.

a2ui_subscribers: Mutex<HashMap<String, Arc<WsChannel>>>

WS clients subscribed to A2UI envelope events. After every successful a2ui.apply / a2ui.ingest, the resulting A2uiApplyResult is broadcast to every subscriber as an a2ui.event JSON-RPC notification. Closes Parslee-ai/car-releases#29. Subscribers register via the a2ui/subscribe method and are auto-cleaned on WS disconnect.

auth_token: OnceLock<String>

Per-launch auth token. When Some, the WS dispatcher rejects non-auth methods on unauthenticated sessions until the client calls session.auth with the matching value. When None, auth is disabled and every connection works as before. Set at startup by car-server unless --no-auth is passed (default flipped 2026-05); embedders that want to enable auth call ServerState::install_auth_token. Closes Parslee-ai/car-releases#32.

host_token: OnceLock<String>

Per-launch host token — a credential distinct from auth_token, granting the host-management role (host-class reads like cross-agent run traces). Critically it is never served over GET /auth-token: a session becomes host-role only by presenting this via session.auth { host_token }, and the only way to obtain it is reading the 0600 host-token file, which a different local user cannot. This is what stops any authenticated local client from self-elevating to host and reading every agent’s run traces (Parslee-ai/car#254). When None, host-role can’t be granted (no host reads).

parslee_session: OnceLock<ParsleeSession>

Parslee cloud identity loaded from the user’s OS keychain at daemon startup when car auth login has been completed.

attached_agents: Mutex<HashMap<String, String>>

agent_id -> client_id map of currently-attached lifecycle agents (#169). Populated by the session.auth handler when a supervised child presents its agent_id + per-agent token; drained on disconnect by remove_session. Single-claim: a second connection presenting the same agent_id is rejected so the daemon-side per-agent state stays unambiguous.

agent_memgines: Mutex<HashMap<String, Arc<Mutex<MemgineEngine>>>>

agent_id -> persistent memgine map (#170). Lazy-loaded on first connection per id from ~/.car/memory/agents/<id>.jsonl, retained across daemon restart, surviving any single disconnect/reconnect of the supervised child. Connections that auth without an agent_id (browser, host, ad-hoc CLI) keep the per-WS ephemeral memgine on ClientSession.memgine — no behaviour change.

chat_sessions: Mutex<HashMap<String, ChatSession>>

In-flight agents.chat sessions keyed by session_id. See ChatSession for shape. Populated by agents.chat, cleared on terminal agent.chat.event or agents.chat.cancel. Disconnect cleanup happens in remove_session — any in-flight session bound to either the disconnecting host or agent client is dropped so subsequent stray notifications from a respawned agent fall on the floor rather than racing into a stale stream.

runs: Mutex<HashMap<String, RunMeta>>

Agent runs keyed by run_id (agent run tracing, U1). Process- wide (not per-session) so a run’s record outlives the WS connection that produced it — the durable, connection- independent boundary client_id cannot be (R1). Populated by runs.start, made terminal by runs.complete, and swept to Incomplete on a mid-run disconnect past the grace window (R5). U2/U3 build the per-turn recorder and disk store on top of this registry.

run_subscribers: Mutex<HashMap<(String, String), RunTraceSubscriber>>

Live runs.trace.event subscribers keyed by (run_id, host_client_id) (agent run tracing, U4). Each value is a crate::host::RunTraceSubscriber — the producer side of a bounded channel whose dedicated drain task writes frames to that connection’s WS. Two CarHost windows on one run register two distinct entries (explicit fanout — the built-in notification registry is single-subscriber-per-method).

Lock contract (invariant #1): runs.subscribe snapshots the run’s turns AND inserts the subscriber while holding the runs lock; the recorder (record_run_turns), start_run, complete_run, and mark_run_incomplete append to runs and push to run_subscribers while holding the SAME runs lock — so snapshot/register and append/notify are serialized. No turn appended in the snapshot/register window is dropped (gap) or double-delivered (dup). The lock order is always runs → run_subscribers; never the reverse.

run_store: RunStore

Disk-backed run-trace store (agent run tracing, U3). Source of truth for REPLAY (U5) — persists each run’s RunStarted, turns, and terminal record as JSONL under ~/.car/runs/{agent_id}/ so a run survives daemon restarts (R4), bounded by retention/GC (R6) and protected at rest with 0600/0700 perms + backup exclusion (R14). The in-memory runs buffer stays the source for the LIVE stream (U4); this store mirrors what was recorded. Derived from journal_dir at construction (sibling runs/).

mcp_url: OnceLock<String>

Bound MCP HTTP-streamable URL (e.g. "http://127.0.0.1:9102/mcp") — car-server installs this after binding the listener. Used by the agents.invoke_external handler to default InvokeOptions.mcp_endpoint so external agents (Claude Code today) load the daemon’s CAR namespace via --mcp-config automatically. None when MCP isn’t bound (e.g. --mcp-bind disabled).

mcp_sessions: OnceLock<Arc<SessionMap>>

Registry of connected MCP SSE sessions. Populated alongside [mcp_url] when car-server boots the MCP listener. Public so handlers can call crate::mcp::push_to_session to send server-initiated requests to a specific MCP-connected client (MCP-3 foundation; MCP-3b will wire host-owned tool dispatch through this).

approval_gate: ApprovalGate

Approval gate for high-risk WS methods (audit 2026-05). The gate intercepts automation.run_applescript, automation.shortcuts.run, messages.send, mail.send, and vision.ocr before they dispatch, raises a host.create_approval for the user to act on, and waits (with a timeout) for host.resolve_approval. Approve → dispatch continues; deny / timeout → JSON-RPC error code -32003. The set of gated methods and the wait timeout are embedder-overridable via ServerStateConfig::with_approval_gate.

Implementations

impl ServerState

pub fn standalone(journal_dir: PathBuf) -> Self

Constructor for the standalone car-server binary. Each WS connection gets its own per-session memgine — matches the pre-extraction default and is correct for a single-process daemon serving one user at a time.

Embedders must not call this. It silently leaves shared_memgine = None, which re-introduces the dual-memgine bug U7 was created to prevent (one engine in the embedder, a fresh one inside every WS session). Embedders use ServerState::embedded instead, which makes the shared engine handle a required argument so it cannot be forgotten.

pub fn embedded( journal_dir: PathBuf, shared_memgine: Arc<Mutex<MemgineEngine>>, ) -> Self

Constructor for embedders (e.g. tokhn-daemon). The shared memgine handle is required: every WS session created by this state will reuse the same engine, preventing the dual-memgine bug.

For embedders that also want to inject a pre-warmed inference engine or other advanced wiring, build a ServerStateConfig directly and call ServerState::with_config.

pub fn with_config(cfg: ServerStateConfig) -> Self

Build a ServerState from a ServerStateConfig — the path embedders use when they need to inject a shared memgine and a pre-warmed inference engine, or any other advanced wiring the convenience constructors don’t cover.

pub fn install_auth_token(&self, token: String) -> Result<(), String>

Enable the per-launch auth handshake. After this call, every new WS connection must call session.auth with token as the first frame; otherwise the connection is closed. Called by car-server at startup unless --no-auth is set (default flipped 2026-05); embedders supply their own token if they want the same posture. Returns Err(token) when auth was already installed.

pub fn install_host_token(&self, token: String) -> Result<(), String>

Install the per-launch host token (Parslee-ai/car#254). A session that later presents this via session.auth { host_token } is granted the host-management role (ClientSession::is_host), which authorize_run_access requires for cross-agent run-trace reads. Set by car-server at startup (mints + writes the 0600 host-token file) unless --no-auth is set. Returns Err(token) when a host token was already installed.

pub fn install_parslee_session( &self, session: ParsleeSession, ) -> Result<(), ParsleeSession>

pub fn install_mcp_url(&self, url: String) -> Result<(), String>

Install the bound MCP URL after car-server’s listener is up. Idempotent on the first call; subsequent calls are accepted silently (matches the supervisor / a2a_dispatcher install idiom). Returns Err(()) when an MCP URL was already installed — embedders should treat this as “another component beat us to it” and use whichever value is now set.

pub fn install_mcp_sessions( &self, sessions: Arc<SessionMap>, ) -> Result<(), Arc<SessionMap>>

Install the MCP SSE session registry. Pairs with [install_mcp_url] — both come from the same start_mcp call and either both get installed or neither does (the daemon binds them together).

pub fn supervisor(&self) -> Result<Arc<Supervisor>, String>

Lazy-initialize and return the agent supervisor. The first call constructs a car_registry::supervisor::Supervisor backed by ~/.car/agents.json + ~/.car/logs/. Embedders that need a non-default location should call ServerState::install_supervisor before any handler runs.

In observer mode (set via [install_observer_manifest]), returns a clear error mentioning the manifest path the primary daemon owns. This prevents the second daemon from re-attempting user_default() (which would also fail with AlreadyRunning) on every WS call, and gives mutation handlers a stable refusal path. Read-only handlers (agents.list, agents.health) should call Self::observer_manifest_path first and fall back to car_registry::supervisor::Supervisor::list_from_manifest / health_from_manifest when set. Closes Parslee-ai/car-releases#44.

pub fn install_supervisor( &self, supervisor: Arc<Supervisor>, ) -> Result<(), Arc<Supervisor>>

Replace the lazy default with a caller-supplied supervisor. Returns Err(()) when a supervisor was already installed. Used by the standalone car-server binary to call start_all() on a known-good handle without paying the lazy-init lookup cost.

pub fn supervisor_if_installed(&self) -> Option<Arc<Supervisor>>

Non-acquiring read of the currently-installed supervisor. Unlike supervisor, this does NOT lazy- init via user_default() — it returns None instead of constructing a fresh Supervisor and acquiring the <manifest>.lock as a side effect. Use this from read-only metadata paths (host.subscribe identity, status surfaces) where causing lock acquisition on observation would be a Heisenberg subscribe — the act of asking “do you own the lock?” must not be the act of taking it.

pub fn install_observer_manifest(&self, path: PathBuf) -> Result<(), PathBuf>

Mark this daemon as observing a manifest owned by another car-server process. After this call, supervisor() returns an “observe-only” error and read-only handlers (agents.list, agents.health) fall back to the static Supervisor::list_from_manifest / health_from_manifest paths. Idempotent — subsequent calls with the same path are no-ops; a different path returns Err(()). Closes Parslee-ai/car-releases#44.

pub fn observer_manifest_path(&self) -> Option<&PathBuf>

Path of the manifest this daemon is observing but not supervising. None when this daemon owns the supervisor (the normal case) or when no manifest is configured at all (no HOME, embedder didn’t install one).

pub async fn a2a_dispatcher(&self) -> Arc<A2aDispatcher>

Lazy-initialize and return the in-core A2A dispatcher. The first call constructs an car_a2a::A2aDispatcher from either the embedder’s overrides (set via ServerStateConfig::with_a2a_runtime / with_a2a_store / with_a2a_card_source) or sensible defaults: a fresh Runtime with register_agent_basics registered, an InMemoryTaskStore, and a card built from the runtime’s tool schemas advertising ws://127.0.0.1:9100/ as the public URL. Closes Parslee-ai/car-releases#28.

pub async fn start_run(&self, meta: RunMeta)

pub async fn complete_run( &self, run_id: &str, termination: RunTermination, ) -> Result<(), String>

Make a run terminal with a harness-reported outcome (runs.complete). Returns Err if the run_id is unknown or already terminal — the handler maps that to a JSON-RPC error so a double-complete or stale id is visible, not silently swallowed.

U3: also appends the terminal RunEnded line to the run’s JSONL file so REPLAY (U5) reports the right status (Completed) after a restart. The disk write happens after the lock is released; disk failures are logged, never fatal.

pub async fn mark_run_incomplete(&self, run_id: &str) -> bool

Mark a run Incomplete (R5) — used by disconnect cleanup when a harness drops without runs.complete. No-op if the run is already terminal (the common healthy-close case where runs.complete won the race). Returns true if it actually wrote the Incomplete marker.

U3: on the transition to Incomplete, appends the terminal RunEnded { Incomplete } line to disk so an orphaned run reports Incomplete (not InProgress) on REPLAY after a restart. Disk failures are logged, never fatal.

pub async fn run_meta(&self, run_id: &str) -> Option<RunMeta>

Non-acquiring read of a run’s current metadata (clone). Used by tests and by the U2 recorder to learn a run’s owning agent_id.

NOTE: this clones the ENTIRE RunMeta, including its turns buffer (full prompts + CLI output — up to the per-run caps). Hot callers that need only the run’s header facts must use Self::run_header instead; cloning the whole buffer per batch RPC is a ~512 MB worst-case copy at this PR’s caps (ADV-2).

pub async fn run_header(&self, run_id: &str) -> Option<(String, bool, usize)>

Lightweight, non-cloning read of a run’s header facts under the runs lock: (agent_id, terminal, turn_count). The owning agent id is the only heap allocation; the turn BUFFER is never cloned (ADV-2).

This is what handle_runs_record_turns needs — owner for the write authz check, terminality for the distinguishable run_terminal drop, and the current turn count for the (fast-path) ceiling pre-check — without the run_meta deep clone of every recorded turn’s payload.

pub async fn record_run_turns( &self, run_id: &str, records: Vec<RunRecord>, ) -> RecordRunTurnsOutcome

Append per-turn trace records to a run’s in-memory buffer (agent run tracing, U2). The recorder calls this after every proposal.submit on a session with a current run, passing the RunRecord::Turns the recorder produced for that proposal.

Returns a RecordRunTurnsOutcome:

• Appended { new_total } — the batch landed; new_total is the run’s new total turn count (the caller computes its next start_index from it).
• RefusedCeiling — the batch would take the run past RECORD_TURNS_RUN_CEILING and was refused WHOLE (the under-lock runaway backstop, ADV-1). Nothing was appended.
• UnknownOrTerminal — the run_id is unknown (the bracket was never opened) or already terminal (a turn arriving after runs.complete is dropped — the run is closed). Nothing was appended.

U3 flushes the in-memory buffer to disk; U4 broadcasts it. Both read via Self::run_turns.

U3: the same turn records are appended to the run’s JSONL file so REPLAY (U5) sees them after a restart. The disk append happens for exactly the records that were accepted into the in-memory buffer (so disk and memory stay in lock-step), keyed by the run’s owning agent_id. Disk failures are logged, never fatal.

pub async fn run_turn_count(&self, run_id: &str) -> usize

Number of turns already recorded for a run — the start_index the recorder passes so per-turn index stays monotonic across the run’s proposals. 0 for an unknown run.

pub async fn run_turns(&self, run_id: &str) -> Vec<RunRecord>

Clone of a run’s ordered per-turn trace (agent run tracing, U2). This is the accessor U3 reads to flush turns to disk and U4 reads to broadcast them. Empty Vec for an unknown run or a run with no turns yet.

pub async fn subscribe_run( &self, run_id: &str, host_client_id: &str, channel: Arc<WsChannel>, ) -> Option<RunSubscribeResponse>

Atomically snapshot a run’s turns AND register a live runs.trace.event subscriber for (run_id, host_client_id) — the load-bearing invariant #1 (agent run tracing, U4).

Holds the runs lock across BOTH (a) reading the run’s current turn buffer as the snapshot at cursor C = turns.len() and (b) inserting the subscriber into run_subscribers. Because the lifecycle path (record_run_turns / complete_run / mark_run_incomplete) appends-and-notifies under the SAME lock, the snapshot contains exactly the turns ≤ C and every turn appended after registration is delivered — no turn in the snapshot/register window is dropped (gap) or double-delivered (dup).

Spawns the subscriber’s drain task (bounded channel → WS) so the producer never writes the socket directly (invariant #2). Returns the snapshot response shape. None if the run_id is unknown (the handler maps that to a not-found error). A re-subscribe by the same (run_id, host_client_id) replaces the prior subscriber (its drain task ends when the old sender drops) and re-snapshots — the R8 reconnect path: the fresh snapshot covers any turns emitted during the gap with no dup.

Terminal-run disk fallback: a completed/incomplete run has its in-memory turns evicted for heap hygiene (see Self::clear_terminal_run_turns). When the run is terminal AND its in-memory buffer is empty, the snapshot is re-sourced from the disk store (crate::run_store::RunStore::get_run_trace, the same source runs.get_trace reads), filtered to RunRecord::Turn so the shape matches the live in-memory snapshot. A terminal run takes no further appends, so the disk trail is final and there is no gap/dup concern — the snapshot/register atomicity the runs lock provides is only load-bearing for the in-progress path, which is unchanged.

pub async fn unsubscribe_run(&self, run_id: &str, host_client_id: &str) -> bool

Remove a live run-trace subscriber for (run_id, host_client_id) (agent run tracing, U4). Returns true if a subscription existed. Dropping the crate::host::RunTraceSubscriber drops its channel sender, which ends the drain task.

pub async fn drop_run_subscribers_for_client(&self, host_client_id: &str)

Drop every live run-trace subscription owned by host_client_id (agent run tracing, U4 — R8 cleanup). Called from [remove_session] on disconnect so a CarHost that drops doesn’t leave dangling drain tasks. Reconnect-durability is client-side: the run stays subscribable while it lives, and the CarHost re-issues runs.subscribe {run_id} on its new connection — the server never synthesizes a failure on subscriber drop.

pub async fn create_session( &self, client_id: &str, channel: Arc<WsChannel>, ) -> Arc<ClientSession>

pub async fn remove_session( &self, client_id: &str, ) -> Option<Arc<ClientSession>>

Remove a per-client session from the registry on disconnect. Returns the removed session if present so callers can drop any remaining strong refs (e.g. drain pending tool callbacks). Fix for MULTI-4 / WS-3 — without this, state.sessions retains Arc<ClientSession> for every connection that ever existed.