Persisting In-Flight State / Reconnect Recovery#

Problem#

An "in-flight" artifact shown in the UI (LLM streaming reply, tool call, agentic function, task spawn, merge) lives only in memory and the WebSocket stream until it finishes. Refreshing the page loses it, because the msg has not been written to the SessionDB yet; it becomes visible in a new page only after the final flush to disk.

If the backend dies midway or the network drops, that artifact is gone: there is no state to query, no way to recover it, and no way to inspect its progress.

Goal#

Any "in-flight" artifact visible in chat / DAG should immediately persist a placeholder, and every subsequent incremental update is flushed to disk + pushed over WS. The frontend can refresh / switch back at any moment, see the current progress + keep receiving the live stream, with zero state loss.

Five Components#

1. Unified placeholder schema#

Every msg (whether user / assistant / tool) gains three new fields:

status:          "pending" | "running" | "done" | "error" | "aborted"
started_at:      float (epoch)
last_update_at:  float (epoch)

pending = id assigned but not started yet; running = actively producing output; one of three terminal states.

Write timing: any operation that runs for a while writes its placeholder immediately:

• LLM streaming reply: before the dispatcher calls the LLM
• tool call: before the tool dispatcher calls the function
• agentic function: before _execute/run.py runs the function
• task spawn: already in place (runner.py)
• merge: before _execute/_run_merge

2. Throttled incremental persistence#

Every status=running msg, while running, writes its current snapshot back to the SessionDB throttled at ~250ms:

• content: streaming partial / current tree dump
• metadata.context_tree: the agentic function's DAG snapshot
• metadata.partial_tokens_used: cumulative input/output tokens
• last_update_at: now

Throttling is managed by a per-msg _ThrottledSaver instance, which guarantees one final flush to disk before the backend exits (atexit / signal hook).

When done, call finalize(status="done", content=...) to write the terminal state.

3. WS subscription by msg_id#

New ws action:

{ "action": "subscribe_msg", "session_id": "...", "msg_id": "..." }

The backend maintains:

_msg_subscribers: dict[tuple[str, str], set[WebSocket]]

On each placeholder update, both persist and push to that channel:

{
  "type": "msg_update",
  "data": {
    "session_id": "...",
    "msg_id": "...",
    "content_delta": "...",
    "tree": {...},
    "status": "running"
  }
}

Subscription release: cleaned up automatically when the client sends unsubscribe_msg / disconnects / the msg enters a terminal state.

4. Frontend load_session detection + auto-resubscribe#

After session-store's feedFromConv call, scan all ChatMsg:

const running = msgs.filter(m => m.status === "running");
running.forEach(m => wsSend({
  action: "subscribe_msg",
  session_id: sid,
  msg_id: m.id,
}));

On receiving a msg_update event, patch the corresponding ChatMsg's content / tree.

UI layer:

• AssistantBubble, on seeing status=running, renders a cursor / streaming animation
• RuntimeBlock, on seeing status=running, shows the running Execution DAG (already supported, since that's how the stream tree is drawn now)
• attach card: consistent with the existing status=running behavior (already present)

5. Death detection + abort sweep#

When the worker starts:

1. Scan ~/.openprogram/sessions/*/history/*.json
2. Find msgs with status=running whose last_update_at exceeds the threshold (e.g. 5 minutes)
3. Set status=aborted, and append a line metadata.aborted_reason="worker restart"

This ensures no orphan msgs are left "running forever" after a backend restart.

Schema Changes#

openprogram/store/session/_msg_adapter.py::_node_to_msg:

• On write, reflect node.metadata.status / started_at / last_update_at into the msg dict
• Old nodes without a status field default to status="done" (backward compatible)

openprogram/context/nodes.py::Call: don't touch the dataclass; all new fields go through metadata.

Out of Scope#

• True resume after interruption — the backend dying midway and, after restart, continuing to run the remaining sub-calls. That requires checkpointing the LLM context and replaying it, several times the effort of this design. Here an interrupted msg is marked aborted, and the user reruns it with the Retry button.
• A cross-session global active-task panel showing which msgs are running right now. The _msg_subscribers keyspace makes this straightforward to add later.

Implementation Status#

The design lands in independently usable, independently revertible pieces:

Of these, the task spawn placeholder is already in place (runner.py), as is the status=running rendering in RuntimeBlock and the attach card.