Persist turn state + restore conversation history on cold-boot

[senamakel]

Summary

• Persist each thread's in-flight agent turn to disk so the UI can rehydrate (or surface "interrupted") after a restart.
• Fix set_agent_definition_name to also rebuild session_key — previously, persist wrote session_raw/<ts>_orchestrator.jsonl while resume searched for <ts>_orchestrator_thread-XXX.jsonl, so transcript-resume always missed and the LLM ran every cold-boot turn with empty history.
• Add a JSONL fallback: when the web channel builds a fresh agent, seed it from the authoritative conversation log via Agent::seed_resume_from_messages so existing on-disk threads (with the misnamed transcripts) recover immediately, not just future ones.
• New JSON-RPC surface (threads_turn_state_get / _list / _clear) and Redux hydration plumbing for the snapshot.

Problem

Cold-booting the desktop app and continuing a conversation behaved as if the thread had no history — the agent reflected the literal current message back instead of answering against the prior turns. Two layered bugs:

1. The web channel scopes transcripts per thread by calling set_agent_definition_name("orchestrator_thread-XXX") after build. The setter only updated agent_definition_name; session_key (used by persist_session_transcript for the filename) stayed at the builder-time <ts>_orchestrator. find_latest_transcript searches by agent_definition_name, so it could never match the on-disk file. Every cold boot ran with an empty transcript.
2. Even after fixing the rename, every existing transcript on disk still has the wrong filename, so the next cold boot would still come up empty for an entire release window.

In parallel, no per-thread runtime state (in-flight tool timeline, streaming buffer, lifecycle) survived a process restart, so a navigation away mid-turn always lost UI state.

Solution

Turn-state persistence

• New openhuman::threads::turn_state module: TurnState types, atomic JSON TurnStateStore at <workspace>/memory/conversations/turn_states/<hex(thread_id)>.json, and a TurnStateMirror that translates AgentProgress → snapshot mutations. High-frequency text/thinking/args deltas mutate memory only; flushes happen on iteration / tool / subagent boundaries to avoid filesystem thrash. Terminal TurnCompleted deletes the file; bridge exit without completion stamps Interrupted.
• Web-channel progress bridge wires the mirror alongside its existing socket emissions.
• bootstrap_skill_runtime calls mark_all_interrupted once per process so stale snapshots from a previous process surface consistently.
• Three new RPC methods on the threads namespace: threads_turn_state_get, _list, _clear.
• App: chatRuntimeSlice gains a hydrateRuntimeFromSnapshot reducer + fetchAndHydrateTurnState thunk. InferenceTurnLifecycle extended with 'interrupted'. Conversations.tsx dispatches the thunk on every thread switch.

Cold-boot history restoration

• set_agent_definition_name now rebuilds session_key's suffix from the new sanitised name (preserving the unix-timestamp prefix). Future persists land at the same path resume searches.
• Agent::seed_resume_from_messages(messages, current_user_message) builds a system prompt and primes cached_transcript_messages with the prior conversation. web::run_chat_request calls it on every cache-miss build using conversations::get_messages as the authoritative source. Trailing user message that matches the incoming message is deduped to avoid double-sending.

Submission Checklist

• Tests added or updated (happy path + at least one failure / edge case) per docs/TESTING-STRATEGY.md
• Diff coverage ≥ 80% — Rust unit tests for the store (7), mirror (7), and session_key/seed_resume regressions (5); JSON-RPC E2E lifecycle test (json_rpc_thread_turn_state_lifecycle); Vitest reducer tests for streaming + interrupted hydration. Full Vitest suite (1347) and cargo test for the affected modules pass locally.
• N/A: behaviour-only change — no new feature rows to add to docs/TEST-COVERAGE-MATRIX.md; the touched threads / agent surfaces are already represented.
• N/A: no matrix-row additions, so no feature IDs to list under ## Related.
• No new external network dependencies introduced (mock backend used per docs/TESTING-STRATEGY.md)
• N/A: not a release-cut surface — turn-state snapshot file + transcript path scoping are internal to the threads domain, no docs/RELEASE-MANUAL-SMOKE.md entries.
• Linked issue tracked via Related — see Consolidate thread message storage onto session_raw transcripts #1201 (this PR is the workaround that issue retires).

Impact

• Runtime: desktop only. New writes under <workspace>/memory/conversations/turn_states/. Snapshot writes are atomic (tempfile::persist) and bounded to iteration / tool boundaries; no per-delta thrash.
• Migration: existing thread message JSONLs are left untouched. Existing misnamed transcripts under session_raw/ are simply ignored — the JSONL fallback recovers history from memory/conversations/threads/.
• Performance: one extra disk read on cache-miss (conversation JSONL parse + system-prompt build), bounded by typical thread length. Hot path (cache-hit) is untouched.
• Security: snapshot files live alongside existing conversation logs and inherit the same workspace permissions; no new attack surface.

Related

• Closes:
• Follow-up PR(s)/TODOs: Consolidate thread message storage onto session_raw transcripts #1201 — consolidate the two thread-message stores onto a single session_raw-backed transcript so the JSONL fallback added here can be retired.

Summary by CodeRabbit

• New Features

  • Persisted per-thread turn snapshots with get/list/clear controls
  • UI fetches and hydrates per-thread runtime when opening a conversation
  • Fresh sessions can be seeded from conversation logs for improved resume

• Behavior

  • Startup marks stale snapshots as interrupted to enable retry paths
  • Session renaming is sanitized to avoid resume/persistence mismatches
  • Thread delete/purge now removes associated persisted turn snapshots

• Tests

  • Added unit and end-to-end tests for persistence, mirroring, recovery, and hydration

[coderabbitai]

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

📝 Walkthrough

Walkthrough

Adds persistent per-thread turn snapshots: Rust records and persists AgentProgress into TurnState files, exposes RPCs to get/list/clear snapshots, marks stale snapshots Interrupted at startup, mirrors turn-state during web-channel streaming, and the frontend fetches and hydrates persisted turn state into Redux when a thread is selected.

Changes

Turn-state persistence → RPC → frontend hydration

Layer / File(s)	Summary
Types / Wire Shapes (backend) src/openhuman/threads/turn_state/types.rs	Adds serde wire/storage types: TurnLifecycle (Started/Streaming/Interrupted), TurnPhase, ToolTimelineStatus, ToolTimelineEntry, SubagentActivity, SubagentToolCall, TurnState snapshot model, and RPC envelopes.
Store Implementation (backend) src/openhuman/threads/turn_state/store.rs, src/openhuman/threads/turn_state/store_tests.rs	Filesystem-backed JSON per-thread snapshot store; atomic overwrite via temp file; global mutex; put/get/delete/list/mark_all_interrupted APIs and tests.
Mirror / Observer (backend) src/openhuman/threads/turn_state/mirror.rs, src/openhuman/threads/turn_state/mirror_tests.rs	TurnStateMirror translates AgentProgress into TurnState, flushes at iteration/tool boundaries, deletes on TurnCompleted, finish() marks Interrupted; tests exercise lifecycle, args buffering, text deltas, subagents.
Module wiring (backend) src/openhuman/threads/turn_state/mod.rs, src/openhuman/threads/mod.rs	New turn_state submodule exposing mirror, store, types; re-exports TurnStateMirror, TurnStateStore, and types.
RPC ops / controller (backend) src/openhuman/threads/ops.rs, src/openhuman/threads/schemas.rs, src/openhuman/threads/schemas_tests.rs	Adds ops turn_state_get/list/clear, controller schemas/handlers, includes snapshot deletion during thread delete/purge; tests updated to include new controller functions.
Startup recovery (backend) src/core/jsonrpc.rs	bootstrap_skill_runtime calls mark_all_interrupted at startup to stamp non-completed snapshots as Interrupted with current timestamp.
Web-channel integration (backend) src/openhuman/channels/providers/web.rs	Session rebuild returns (agent, was_built_fresh); fresh builds seed resume from conversation JSONL via new seed_resume_from_messages; progress bridge now constructs TurnStateMirror with TurnStateStore and mirrors AgentProgress events, deleting or marking Interrupted on exit.
Session helpers & tests (backend) src/openhuman/agent/harness/session/runtime.rs, src/openhuman/agent/harness/session/tests.rs	set_agent_definition_name sanitizes and rebuilds session_key suffix; seed_resume_from_messages primes cached transcript on cold agents; tests added.
End-to-end RPC test (integration) tests/json_rpc_e2e.rs	New Tokio JSON-RPC e2e test exercising turn-state list→get→clear lifecycle.
Types / API client (frontend) app/src/types/turnState.ts, app/src/services/api/threadApi.ts	Adds TypeScript persisted turn-state types and threadApi.getTurnState, listTurnStates, clearTurnState that call RPC endpoints and normalize envelope results.
Redux slice & thunk (frontend) app/src/store/chatRuntimeSlice.ts, app/src/store/__tests__/chatRuntimeSlice.test.ts	Extends InferenceTurnLifecycle with 'interrupted'; adds conversion helpers from persisted snapshots; new hydrateRuntimeFromSnapshot reducer; fetchAndHydrateTurnState(threadId) thunk fetches snapshot and dispatches hydration; tests cover streaming and interrupted snapshot hydration.
UI wiring (frontend) app/src/pages/Conversations.tsx	On selectedThreadId change, dispatches loadThreadMessages(selectedThreadId) then fetchAndHydrateTurnState(selectedThreadId) to restore per-thread runtime state.

Sequence Diagram(s)

sequenceDiagram
    participant UI as UI (Conversations)
    participant Redux as Redux (chatRuntimeSlice)
    participant API as Frontend API (threadApi)
    participant RPC as RPC Server (threads/ops)
    participant Store as TurnStateStore (Filesystem)

    UI->>Redux: dispatch fetchAndHydrateTurnState(threadId)
    Redux->>API: getTurnState(threadId)
    API->>RPC: threads_turn_state_get
    RPC->>Store: get(threadId)
    Store-->>RPC: TurnState | null
    RPC-->>API: GetTurnStateResponse
    API-->>Redux: PersistedTurnState | null
    Redux->>Redux: dispatch hydrateRuntimeFromSnapshot(snapshot)
    Redux->>UI: Update runtime slices (lifecycle/status/streaming/toolTimeline)

Loading

sequenceDiagram
    participant Web as Web Channel (run_chat_task)
    participant Conv as Conversation JSONL (Memory)
    participant Agent as Agent Session
    participant Bridge as Progress Bridge (TurnStateMirror)
    participant Store as TurnStateStore (Filesystem)

    Web->>Agent: build_session_agent() or reuse from cache
    alt fresh agent built
        Web->>Conv: get_messages(thread_id)
        Conv-->>Web: [(role, content), ...]
        Web->>Agent: seed_resume_from_messages(pairs, current_msg)
        Agent->>Agent: populate cached_transcript_messages
    end
    Web->>Bridge: spawn_progress_bridge(TurnStateStore,...)
    Bridge->>Store: TurnStateMirror::new() flush started snapshot
    loop agent progress
        Agent->>Bridge: AgentProgress event
        Bridge->>Bridge: observe(event) update TurnState
        alt flush boundary (iteration/tool events)
            Bridge->>Store: flush() persist snapshot
        end
    end
    alt TurnCompleted
        Bridge->>Store: delete snapshot
    else early exit
        Bridge->>Store: finish() mark Interrupted + flush
    end

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Possibly related issues

• Consolidate thread message storage onto session_raw transcripts #1201: Related to restart/resume transcript behavior; this PR implements turn-state persistence and recovery referenced by that issue.

Possibly related PRs

• tinyhumansai/openhuman#590: Overlaps thread/Conversations and threadApi surface changes; closely related.
• tinyhumansai/openhuman#1159: Modifies ToolTimeline/subagent runtime types and chatRuntimeSlice; related to persisted → runtime mapping.
• tinyhumansai/openhuman#514: Related Conversations.tsx inference/runtime state handling and AgentProgress wiring.

Poem

🐰 I stitched each turn to disk tonight,

Deltas tucked safe in JSON light,
Interrupted flags for sleepy ends,
Hydrate and hop—resume, dear friends,
Threads awake, the story mends.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title 'Persist turn state + restore conversation history on cold-boot' accurately and specifically describes the main changes in the PR.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 5

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@app/src/store/chatRuntimeSlice.ts`:
- Around line 260-282: The code currently writes snapshot.lifecycle into
inferenceTurnLifecycleByThread but still hydrates inferenceStatusByThread and
streamingAssistantByThread from snapshot even when the lifecycle is
'interrupted'; change the logic in the reducer handling snapshot to first check
if snapshot.lifecycle === 'interrupted' (or equivalent interrupted value) and,
when true, skip populating inferenceStatusByThread and
streamingAssistantByThread (instead delete/leave them absent) so that no
live-progress UI is restored for interrupted turns; update the branches around
inferenceTurnLifecycleByThread, inferenceStatusByThread, and
streamingAssistantByThread to apply this lifecycle guard before using
snapshot.iteration, snapshot.streamingText, or snapshot.thinking.

In `@src/openhuman/channels/providers/web.rs`:
- Around line 595-600: TurnStateMirror::observe() returns a boolean indicating a
flush-boundary snapshot must be persisted, but the code currently drops that
value at the call site (turn_state.observe(&event)); update the loop to capture
the return and, when true, immediately persist the snapshot using the same
persistence mechanism used by finish() (i.e., write the snapshot into the
turn_state_store) so boundary updates are durable before TurnCompleted/finish()
runs; reference TurnStateMirror::observe(), the turn_state.observe(&event) call,
and the existing finish()/TurnCompleted persistence path to implement the
immediate flush.

In `@src/openhuman/threads/ops.rs`:
- Around line 462-499: thread lifecycle cleanup currently doesn't remove
persisted turn-state snapshots; update the thread deletion and purge code paths
(e.g., the functions thread_delete and threads_purge in this module) to also
remove turn-state files from the workspace. After acquiring the same
workspace_dir() used by turn_state_get/list/clear, call the turn_state store
APIs (e.g., turn_state::store::delete(dir, &thread_id) for single-thread
deletion and iterate/delete all relevant entries when purging) and
handle/propagate any errors consistently with existing delete/purge behavior so
no stale turn_state remains.

In `@src/openhuman/threads/turn_state/mirror.rs`:
- Around line 79-100: AgentProgress::ToolCallStarted currently always appends a
new ToolTimelineEntry even if a placeholder row with the same call_id was
created earlier (e.g. by ToolCallArgsDelta), causing duplicate entries; modify
the ToolCallStarted handling to search self.state.tool_timeline for an existing
entry with id == call_id and reuse it by updating its fields (name, round,
status -> ToolTimelineStatus::Running, active_tool,
display_name/detail/subagent/source_tool_name as needed) instead of pushing a
new entry, and only push a new ToolTimelineEntry if no existing entry is found;
apply the same reuse logic to the analogous handler referenced around lines
259-289.

In `@src/openhuman/threads/turn_state/store.rs`:
- Around line 39-63: The put() method currently fsyncs the tempfile but not the
parent directory after tmp.persist(&path), so add an explicit fsync of the
snapshot directory to ensure the directory entry (rename) is durable: after
tmp.persist(&path) (in the put function in store.rs, which uses ensure_dir(),
snapshot_path(), NamedTempFile and TURN_STATE_LOCK), open the directory (e.g.,
std::fs::File::open(&dir) or use Dir::open) and call sync_all() and map any
error into the same Result<String, _> error path (similar formatting as the
other map_err messages) so a failure to sync the directory returns an Err.
Ensure the directory fsync occurs only after persist() succeeds.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 352827a8-bd65-42ee-a25a-3493d773a4ee

📥 Commits

Reviewing files that changed from the base of the PR and between 62516ba and 255f307.

⛔ Files ignored due to path filters (2)

• Cargo.lock is excluded by !**/*.lock
• app/src-tauri/Cargo.lock is excluded by !**/*.lock

📒 Files selected for processing (19)

• app/src/pages/Conversations.tsx
• app/src/services/api/threadApi.ts
• app/src/store/__tests__/chatRuntimeSlice.test.ts
• app/src/store/chatRuntimeSlice.ts
• app/src/types/turnState.ts
• src/core/jsonrpc.rs
• src/openhuman/agent/harness/session/runtime.rs
• src/openhuman/agent/harness/session/tests.rs
• src/openhuman/channels/providers/web.rs
• src/openhuman/threads/mod.rs
• src/openhuman/threads/ops.rs
• src/openhuman/threads/schemas.rs
• src/openhuman/threads/turn_state/mirror.rs
• src/openhuman/threads/turn_state/mirror_tests.rs
• src/openhuman/threads/turn_state/mod.rs
• src/openhuman/threads/turn_state/store.rs
• src/openhuman/threads/turn_state/store_tests.rs
• src/openhuman/threads/turn_state/types.rs
• tests/json_rpc_e2e.rs

[coderabbitai]

Actionable comments posted: 1

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@src/openhuman/threads/ops.rs`:
- Around line 441-446: The current code calls turn_state::store::delete(dir,
&request.thread_id) and only logs a warning on Err, which allows thread_delete /
threads_purge to return success while persisted turn snapshots remain; change
the flow to propagate the deletion failure instead of silencing it: in the
thread_delete and threads_purge handlers detect Err from
turn_state::store::delete and either return an RPC error (propagate the Err) or
include a partial-failure field in the RPC response indicating which thread_ids
failed to delete (using request.thread_id to identify them); update any call
sites that expect unconditional success accordingly so the RPC result reflects
the cleanup failure rather than only logging it.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: bf1a0520-18ac-4ba3-b8e7-ed34010bf7d4

📥 Commits

Reviewing files that changed from the base of the PR and between 256ecf4 and 68d40d5.

📒 Files selected for processing (6)

• app/src/store/__tests__/chatRuntimeSlice.test.ts
• app/src/store/chatRuntimeSlice.ts
• src/openhuman/threads/ops.rs
• src/openhuman/threads/turn_state/mirror.rs
• src/openhuman/threads/turn_state/mirror_tests.rs
• src/openhuman/threads/turn_state/store.rs

✅ Files skipped from review due to trivial changes (3)

• app/src/store/tests/chatRuntimeSlice.test.ts
• src/openhuman/threads/turn_state/mirror_tests.rs
• src/openhuman/threads/turn_state/mirror.rs

🚧 Files skipped from review as they are similar to previous changes (1)

• src/openhuman/threads/turn_state/store.rs

[coderabbitai]

Actionable comments posted: 1

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@src/openhuman/threads/ops.rs`:
- Around line 445-451: turn_state::store::delete may return early and prevent
web_channel::invalidate_thread_sessions from running, leaving active sessions
for a deleted thread; ensure
invalidate_thread_sessions(&request.thread_id).await always runs even when
delete fails by calling web_channel::invalidate_thread_sessions before returning
the error (or by capturing the delete result, awaiting
invalidate_thread_sessions, then propagating the formatted error). Refer to
turn_state::store::delete, web_channel::invalidate_thread_sessions, and
request.thread_id to locate and implement this reorder/try-finally style flow.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: cf9473ec-adb8-4ff6-b1f7-42e1364aef2c

📥 Commits

Reviewing files that changed from the base of the PR and between 68d40d5 and f33cf10.

📒 Files selected for processing (1)

• src/openhuman/threads/ops.rs

[coderabbitai]

Actionable comments posted: 1

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@src/openhuman/threads/ops.rs`:
- Around line 475-484: threads_purge currently calls turn_state::store::list
which skips unreadable/corrupted snapshot files, so purging can falsely succeed;
add and call a deserialize-independent API on the store (e.g.,
turn_state::store::clear_all or TurnStateStore::clear_all) from threads_purge to
remove all snapshot artifacts regardless of readability, and update
threads_purge to use that new API instead of iterating list/delete so corrupted
files are also deleted.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 186b4805-17b1-4dc9-bab6-0d518b6df15b

📥 Commits

Reviewing files that changed from the base of the PR and between f33cf10 and 179abfc.

📒 Files selected for processing (1)

• src/openhuman/threads/ops.rs