feat(session): per-turn persistence for crash/interrupt-safe resume

AGENTS.md

@@ -92,6 +92,7 @@ ReAct cycle: observe → think → act → repeat.
 - **Interaction modes** — engaging (narrated), enhance (persistent), verbose (raw), off.
 - Max 300 iterations by default.
 - **Post-response async processing** — skill learning, episode extraction, and per-turn extended-memory extraction run in background goroutines tracked by `MemoryManager.RunBackground`; `Agent.Close` drains them via `WaitForBackground` (bounded, ~15s) so the work survives CLI exit without hanging `odek run`.
+- **Per-turn session persistence** — the loop fires an optional `SetMessagesPersistCallback` after each completed step (after a tool batch's result messages, and after the final assistant message) with a freshly-copied message snapshot. `run --session`, `continue`, the REPL, and `odek serve` wire it to `Store.SaveNoIndex` — an atomic save that skips the vector-index update (embedding can be a remote HTTP call and must not fire every iteration) — so an interrupted run (Ctrl-C, SIGTERM, crash) can be resumed via `odek continue` from the last completed step. Error/interrupt paths additionally persist the partial history minus any trailing assistant message with unanswered tool calls (`persistPartialMessages`). The final successful save still goes through `Save`/`Append`, updating the semantic index once per completed turn. If the loop trimmed history in place, the callback skips the save rather than overwriting richer persisted state. `odek serve` filters dynamically-injected system messages (skills/memory/episodes) out of persisted snapshots, preserving the session's original leading system message.
 - **Storage maintenance janitor** — `maintenance.Start` (internal/maintenance) runs a periodic sweep of `~/.odek` (expired sessions/audit/plans, oversized-log rotation, media sweep, skill skip-list GC) inside `odek telegram`, `odek serve`, and `odek schedule daemon` when `maintenance.enabled` is set; `odek cleanup [--dry-run]` runs the same sweep on demand. Session files are also trimmed at write time (oldest groups first, system message kept, `[Session storage limit: ...]` marker persisted into the transcript) when they would exceed `MaxSessionFileBytes`, so a session can never grow past the load cap. Operator-only config — project `./odek.json` cannot set it. See docs/MAINTENANCE.md.
 - **Artifact-aware file search** — `search_files` and `multi_grep` skip build/artifact directories (`node_modules`, `vendor`, `.git`, `__pycache__`, `.venv`, etc.) automatically, reducing noise and speeding scans.
 - **Semantic session search** — the `session_search` tool uses go-vector RandomProjections + k-NN for semantic similarity search through session content, with a two-tier pipeline: vector index (fast, ~1ms) → deepSearch fallback (exhaustive, slower).

cmd/odek/main.go

@@ -1497,6 +1497,7 @@ func run(args []string) error {
 	var sessIDCapture string
 	var currentTurn int
 	var sessionStore *session.Store
+	var runSess *session.Session
 
 	cwd, _ = os.Getwd()
 	if mm := agent.Memory(); mm != nil {
@@ -1520,6 +1521,7 @@ func run(args []string) error {
 			sess.Sandbox = resolved.Sandbox
 			store.Save(sess)
 			sessionID = sess.ID
+			runSess = sess
 			mm.SetSessionContext(sessionID, cwd)
 			fmt.Fprintf(os.Stderr, "odek: session %s created\n", sessionID)
 
@@ -1587,6 +1589,21 @@ func run(args []string) error {
 			mm.AppendBuffer("user", f.Task)
 		}
 
+		// Persist per-turn progress so an interrupted run (Ctrl-C, SIGTERM,
+		// crash) can be resumed via `odek continue` from the last completed
+		// step instead of losing the whole in-progress turn.
+		if runSess != nil {
+			agent.SetMessagesPersistCallback(func(snapshot []llm.Message) {
+				if len(snapshot) < len(runSess.Messages) {
+					// The loop trimmed history in place — keep the richer
+					// state already persisted instead of overwriting it.
+					return
+				}
+				runSess.Messages = snapshot
+				_ = sessionStore.SaveNoIndex(runSess)
+			})
+		}
+
 		result, allMessages, runErr = agent.RunWithMessages(ctx, messages)
 
 		// Append agent response to buffer
@@ -1603,13 +1620,17 @@ func run(args []string) error {
 
 		if runErr == nil {
 			// Re-load the pre-created session and append the messages produced
-			// by the run. The pre-created session contains the system + user task;
-			// append only the assistant/tool turns that follow.
+			// by the run. The per-turn persist callback above already saved the
+			// full history, so this delta is usually empty — Append still
+			// refreshes metadata and updates the vector index.
 			latest, err := sessionStore.Load(sessionID)
 			if err != nil {
 				return fmt.Errorf("load session: %w", err)
 			}
-			newMsgs := allMessages[len(latest.GetMessages()):]
+			var newMsgs []llm.Message
+			if n := len(latest.GetMessages()); n < len(allMessages) {
+				newMsgs = allMessages[n:]
+			}
 			if err := sessionStore.Append(sessionID, newMsgs); err != nil {
 				return fmt.Errorf("save session: %w", err)
 			}
@@ -1641,6 +1662,13 @@ func run(args []string) error {
 		result, allMessages, runErr = agent.RunWithMessages(ctx, messages)
 	}
 
+	if runErr != nil && runSess != nil {
+		// Interrupted/failed session run — persist the partial history so
+		// the in-progress turn survives up to the last completed step
+		// (mirrors the Telegram cancellation path).
+		persistPartialMessages(sessionStore, runSess, allMessages)
+	}
+
 	if runErr != nil {
 		return runErr
 	}
@@ -2419,6 +2447,29 @@ func expandHome(path string) string {
 
 // ── Continue (Multi-Turn) ─────────────────────────────────────────────
 
+// persistPartialMessages saves the in-flight message history of an
+// interrupted/failed run so the session keeps progress up to the last
+// completed step. A trailing assistant message with unanswered tool calls
+// is dropped first: its tool results never completed, and resuming with
+// dangling tool calls is an invalid request for OpenAI-compatible APIs.
+func persistPartialMessages(store *session.Store, sess *session.Session, messages []llm.Message) {
+	if store == nil || sess == nil || len(messages) == 0 {
+		return
+	}
+	for len(messages) > 0 &&
+		messages[len(messages)-1].Role == "assistant" &&
+		len(messages[len(messages)-1].ToolCalls) > 0 {
+		messages = messages[:len(messages)-1]
+	}
+	if len(messages) < len(sess.Messages) {
+		// The loop trimmed history in place — keep the richer state already
+		// persisted by the per-turn callback instead of overwriting it.
+		return
+	}
+	sess.Messages = messages
+	_ = store.SaveNoIndex(sess)
+}
+
 // continueCmd handles `odek continue [--id <id>] <task>`.
 // It loads an existing session (latest or by ID), appends the new task,
 // runs the agent with full history, and saves the updated session.
@@ -2589,6 +2640,10 @@ func continueCmd(args []string) error {
 	// Build message history: session messages + new user message
 	// The system message is already in the session
 	messages := sess.GetMessages()
+	// histLen is the pre-run history length used for the audit delta below;
+	// the per-turn persist callback replaces sess.Messages during the run,
+	// so it cannot be derived from sess afterwards.
+	histLen := len(messages)
 
 	// Create the run context early so that the return-after-break summary can
 	// be recorded in the audit log before the turn starts.
@@ -2625,16 +2680,33 @@ func continueCmd(args []string) error {
 	}
 
 	rend.Start(task)
+
+	// Persist per-turn progress so an interrupted run (Ctrl-C, SIGTERM,
+	// crash) can be resumed again from the last completed step instead of
+	// losing the whole in-progress turn.
+	agent.SetMessagesPersistCallback(func(snapshot []llm.Message) {
+		if len(snapshot) < len(sess.Messages) {
+			// The loop trimmed history in place — keep the richer state
+			// already persisted instead of overwriting it.
+			return
+		}
+		sess.Messages = snapshot
+		_ = store.SaveNoIndex(sess)
+	})
+
 	result, allMessages, err := agent.RunWithMessages(ctx, messages)
 	if err != nil {
+		// Persist the partial history so the interrupted turn survives up
+		// to the last completed step (mirrors the Telegram cancel path).
+		persistPartialMessages(store, sess, allMessages)
 		return err
 	}
 	_ = result
 
 	// Record per-turn divergence assessment after the turn completes.
 	// Use the original prompt so injected resources from @-refs/--ctx do
 	// not count as user-mentioned.
-	recordTurnAudit(auditStore, sessIDCapture, currentTurn, originalTask, allMessages[len(sess.GetMessages()):])
+	recordTurnAudit(auditStore, sessIDCapture, currentTurn, originalTask, allMessages[histLen:])
 
 	// Append agent response to buffer
 	if len(allMessages) > 0 {
@@ -2648,18 +2720,18 @@ func continueCmd(args []string) error {
 		}
 	}
 
-	// Save updated session — persist messages AND buffer
-	newMsgs := allMessages[len(sess.GetMessages()):]
-	if err := store.Append(sess.ID, newMsgs); err != nil {
-		return fmt.Errorf("save session: %w", err)
+	// The per-turn persist callback already saved the full message history;
+	// finish with one Save to persist the buffer and update the vector index
+	// for the completed turn.
+	updated, err := store.Load(sess.ID)
+	if err != nil {
+		return fmt.Errorf("reload session: %w", err)
 	}
-	// Re-load session to persist buffer (Append reads from disk)
 	if mm := agent.Memory(); mm != nil {
-		updated, err := store.Load(sess.ID)
-		if err == nil {
-			updated.Buffer = mm.GetBuffer()
-			store.Save(updated)
-		}
+		updated.Buffer = mm.GetBuffer()
+	}
+	if err := store.Save(updated); err != nil {
+		return fmt.Errorf("save session: %w", err)
 	}
 
 	fmt.Fprintf(os.Stderr, "odek: session %s saved (%d turns)\n", sess.ID, sess.Turns+1)

cmd/odek/repl.go

@@ -195,6 +195,20 @@ func replCmd(args []string) error {
 		sess.Sandbox = resolved.Sandbox
 		store.Save(sess)
 	}
+
+	// Persist per-turn progress so an interrupted turn (Ctrl-C) survives up
+	// to the last completed step instead of losing the whole turn.
+	agent.SetMessagesPersistCallback(func(snapshot []llm.Message) {
+		if sess == nil || len(snapshot) < len(sess.Messages) {
+			// The loop trimmed history in place — keep the richer state
+			// already persisted instead of overwriting it.
+			return
+		}
+		sess.Messages = snapshot
+		if err := store.SaveNoIndex(sess); err != nil {
+			fmt.Fprintf(os.Stderr, "odek: save error: %v\n", err)
+		}
+	})
 	cwd, _ := os.Getwd()
 	if mm := agent.Memory(); mm != nil {
 		mm.SetSessionContext(sess.ID, cwd)
@@ -260,9 +274,6 @@ func replCmd(args []string) error {
 			messages = injectReturnAfterBreak(ctx, agent.Memory(), messages)
 			resumedSession = false
 		}
-		// origLen is computed after the (ephemeral) injection so only the
-		// new user/assistant turns are persisted back to the session.
-		origLen := len(messages)
 		messages = append(messages, llm.Message{Role: "user", Content: input})
 
 		// Append user input to buffer (AppendBuffer summarizes raw text).
@@ -274,6 +285,9 @@ func replCmd(args []string) error {
 		rend.Start(input)
 		_, allMessages, err := agent.RunWithMessages(ctx, messages)
 		if err != nil {
+			// Persist the partial history so the interrupted turn survives
+			// up to the last completed step (mirrors the Telegram cancel path).
+			persistPartialMessages(store, sess, allMessages)
 			fmt.Fprintf(os.Stderr, "odek: agent error: %v\n", err)
 			continue
 		}
@@ -285,18 +299,16 @@ func replCmd(args []string) error {
 			}
 		}
 
-		// Save new messages to session
-		newMsgs := allMessages[origLen:]
-		if err := store.Append(sess.ID, newMsgs); err != nil {
-			fmt.Fprintf(os.Stderr, "odek: save error: %v\n", err)
-		}
-
-		// Reload session to get updated turn count + persist buffer
+		// The per-turn persist callback already saved the full history;
+		// reload and Save once more to persist the buffer and update the
+		// vector index for the completed turn.
 		sess, _ = store.Load(sess.ID)
 		if sess != nil {
 			if mm := agent.Memory(); mm != nil {
 				sess.Buffer = mm.GetBuffer()
-				store.Save(sess)
+			}
+			if err := store.Save(sess); err != nil {
+				fmt.Fprintf(os.Stderr, "odek: save error: %v\n", err)
 			}
 		}
 

cmd/odek/serve.go

@@ -1082,6 +1082,40 @@ func handlePrompt(
 	}
 
 	origLen := len(messages) - 1 // initial estimate: index of the user message we appended
+
+	// Persist per-turn progress so an interrupted run can be resumed from
+	// the last completed step instead of losing the whole in-progress turn.
+	// Mirror the store path below: dynamically-injected system messages
+	// (skills, memory, episodes) are filtered out so persisted snapshots
+	// don't accumulate internal injections or corrupt future origLen
+	// calculations. The session's own leading system message is preserved.
+	if sess != nil {
+		var head []llm.Message
+		if len(sess.Messages) > 0 && sess.Messages[0].Role == "system" {
+			head = sess.Messages[:1]
+		}
+		agent.SetMessagesPersistCallback(func(snapshot []llm.Message) {
+			if len(snapshot) < len(sess.Messages) {
+				// The loop trimmed history in place — keep the richer state
+				// already persisted instead of overwriting it.
+				return
+			}
+			filtered := make([]llm.Message, 0, len(snapshot))
+			filtered = append(filtered, head...)
+			for i, m := range snapshot {
+				if i == 0 && len(head) > 0 {
+					continue // replaced by the session's original head
+				}
+				if m.Role == "system" {
+					continue
+				}
+				filtered = append(filtered, m)
+			}
+			sess.Messages = filtered
+			_ = store.SaveNoIndex(sess)
+		})
+	}
+
 	start := time.Now()
 	_, allMessages, err := agent.RunWithMessages(ctx, messages)
 	latency := time.Since(start)
@@ -1183,21 +1217,16 @@ func handlePrompt(
 		"sessionOutputTokens":  *sessionOutputTokens,
 	})
 
-	// Save session — persist messages AND buffer.
-	// Filter out dynamically-injected system messages (skills, memory, episodes)
-	// so they are not stored in the session and don't corrupt future origLen
-	// calculations on subsequent turns. Only user/assistant/tool messages persist.
+	// Save session — persist buffer and update the vector index.
+	// The message history was already persisted per-turn by the persist
+	// callback above (which filters dynamically-injected system messages —
+	// skills, memory, episodes — so they are not stored in the session and
+	// don't corrupt future origLen calculations on subsequent turns).
 	if sess != nil {
 		if mm := agent.Memory(); mm != nil {
 			sess.Buffer = mm.GetBuffer()
 		}
-		var toStore []llm.Message
-		for _, m := range newMsgs {
-			if m.Role != "system" {
-				toStore = append(toStore, m)
-			}
-		}
-		store.Append(sess.ID, toStore)
+		store.Save(sess)
 	}
 
 	// ── Learn loop: run self-improvement heuristics ──

internal/loop/loop.go

@@ -103,6 +103,14 @@ type IterationInfo struct {
 // of the agent loop. Used by Telegram/WebUI for progress reporting.
 type IterationCallback func(info IterationInfo)
 
+// MessagesPersistCallback is an optional callback invoked after each
+// completed step of the agent loop (after a tool batch's result messages
+// are appended, and after the final assistant message). It receives a
+// freshly-allocated copy of the current message history so callers can
+// persist per-turn progress; an interrupted run can then be resumed from
+// the last completed step instead of losing the whole in-progress turn.
+type MessagesPersistCallback func(messages []llm.Message)
+
 // Engine runs the agent loop: observe → think → act → repeat.
 type Engine struct {
 	client         *llm.Client
@@ -138,6 +146,11 @@ type Engine struct {
 	// iterationCallback is an optional callback fired after each iteration.
 	iterationCallback IterationCallback
 
+	// messagesPersistCallback is an optional callback fired after each
+	// completed step with a copy of the current message history, so callers
+	// can persist per-turn progress for crash/interrupt recovery.
+	messagesPersistCallback MessagesPersistCallback
+
 	// memoryPromptFunc is called before each LLM invocation to get fresh
 	// memory content. This ensures memory mutations during a session
 	// are visible to the agent on the next turn.
@@ -283,6 +296,12 @@ func (e *Engine) SetNarrator(n *narrate.Narrator) { e.narrator = n }
 // If nil, no callback is fired.
 func (e *Engine) SetIterationCallback(cb IterationCallback) { e.iterationCallback = cb }
 
+// SetMessagesPersistCallback sets the per-step message persistence callback.
+// If nil, no callback is fired.
+func (e *Engine) SetMessagesPersistCallback(cb MessagesPersistCallback) {
+	e.messagesPersistCallback = cb
+}
+
 // SetMaxToolParallel sets the maximum concurrency for tool execution per
 // iteration. 0 or negative = use default (4).
 func (e *Engine) SetMaxToolParallel(n int) { e.MaxToolParallel = n }
@@ -1253,6 +1272,7 @@ func (e *Engine) runLoop(ctx context.Context, messages []llm.Message) (string, [
 				Content:          result.Content,
 				ReasoningContent: result.ReasoningContent,
 			})
+			e.emitMessagesPersist(messages)
 			return result.Content, messages, nil
 		}
 
@@ -1574,6 +1594,10 @@ func (e *Engine) runLoop(ctx context.Context, messages []llm.Message) (string, [
 			})
 		}
 
+		// Persist per-turn progress now that the tool batch's result
+		// messages are appended — an interrupted run can resume from here.
+		e.emitMessagesPersist(messages)
+
 		// Fire iteration callback with tool call results
 		if e.iterationCallback != nil {
 			e.iterationCallback(IterationInfo{
@@ -1597,6 +1621,19 @@ func (e *Engine) runLoop(ctx context.Context, messages []llm.Message) (string, [
 
 // ── Helpers ───────────────────────────────────────────────────────────
 
+// emitMessagesPersist fires the per-step persistence callback with a
+// freshly-allocated copy of the message history. The copy is required
+// because trimContext mutates the loop's slice in place — a handed-out
+// snapshot must not change under the caller. Nil callback = no-op.
+func (e *Engine) emitMessagesPersist(messages []llm.Message) {
+	if e.messagesPersistCallback == nil {
+		return
+	}
+	snapshot := make([]llm.Message, len(messages))
+	copy(snapshot, messages)
+	e.messagesPersistCallback(snapshot)
+}
+
 // lastUserMessage returns the content of the most recent user message.
 func lastUserMessage(messages []llm.Message) string {
 	for i := len(messages) - 1; i >= 0; i-- {