fix(rewind): truncate by conversation-wide message boundary, invalidate live mirror

docs/logs/engineering-log.md

@@ -5906,3 +5906,77 @@ Skipped creating separate issues for Op/EventMsg protocol (already covered by SS
   the touched `INDEX.md` files, the three corrected investigation files, and
   this log against `test -e` (758 links, 0 dead). This is documentation only:
   no runtime, API, or test behavior changed.
+
+# 2026-09-05 (Issue #1370 rewind message truncation and live mirror repair)
+
+- Cause: `RestoreRewindPoint` truncated `conversation_messages` by comparing
+  the rewind point's run-local tool-call step (`RewindPoint.Step`, set in
+  `runner_step_engine.go`) against `conversation_messages.step`, a
+  conversation-wide message index shared across every run on that
+  conversation. Rewinding to a point in a conversation's second (or later)
+  run deleted the first run's later messages too. Separately, the runner's
+  in-memory conversation mirror (populated at each run's completion and
+  served by `ConversationMessagesSnapshot`/`GET /messages`) was never
+  invalidated by rewind, so the live daemon kept serving pre-rewind history
+  and the next run re-persisted it over the truncated DB.
+- Fix: `RewindPoint` gained `MessageBoundary`, the conversation-wide index of
+  the assistant message carrying the rewound tool call, recorded at capture
+  time in the step engine (see the followup entry below for a correction to
+  this value). `RestoreRewindPoint` truncates by `step >= MessageBoundary`
+  when it is recorded; points captured before this field existed
+  (`MessageBoundary == 0`) fall back to the legacy step comparison with a
+  logged warning instead of silently over-deleting.
+  `Runner.InvalidateConversationHistory` drops the in-memory mirror entry
+  for a conversation; the rewind HTTP handler calls it immediately after a
+  successful restore.
+- Regression: a store-level test proves a two-run conversation's rewind keeps
+  run 1's messages plus run 2's user prompt and tool-call message,
+  truncating only what follows; an HTTP-level test drives two real runs
+  through the handler and proves `GET /messages` reflects the truncation
+  immediately; a third test drives a real three-run `Runner` flow and proves
+  the run following a rewind does not resurrect the truncated tool result or
+  final answer in its LLM request. Related: #1303 describes the same
+  resurrection symptom from a different angle (workspace population, TUI
+  JSON tags) and remains open.
+
+# 2026-09-05 (Issue #1370 followup: dangling assistant tool_calls after restore)
+
+- Cause: the boundary above (`len(messages)` at capture time) pointed just
+  past the assistant message carrying the rewound tool call, so a restore
+  kept that assistant message while deleting only its tool result. Real
+  providers (OpenAI et al.) reject an assistant message with `tool_calls`
+  that isn't immediately followed by matching tool messages; the fake/stub
+  providers this repo's tests use do not enforce that, so the bug shipped
+  with green tests. Caught in review before merge stabilized.
+- Fix: `MessageBoundary` is now the conversation-wide index of the assistant
+  message itself (`assistantToolCallIndex`, captured immediately after that
+  message is appended in `runner_step_engine.go`), so restore deletes that
+  message and everything after it. Parallel tool calls issued in one
+  assistant turn all capture the same index, since they share one assistant
+  message. `RestoreRewindPoint`'s truncation query and its
+  unset-boundary fallback are unchanged; only the captured value changed.
+- Regression: a new test drives a real single-turn run with two parallel
+  tool calls through the actual capture path and restores using either
+  call's point, asserting both points share one boundary, the last
+  persisted message is never an assistant message with tool_calls, and no
+  tool message lacks a preceding assistant tool_calls entry for its ID.
+
+# 2026-09-05 (Issue #1370 followup: rewind_points prune deleted unrelated older points)
+
+- Cause: found by live verification on main after #1378 merged.
+  `RestoreRewindPoint`'s future-point pruning query also compared
+  `point.Step`, the same run-local tool-call counter whose misuse in message
+  truncation this issue already fixed. Step restarts every run, so run 2's
+  edit point (step 1 within run 2) and run 1's write point (also step 1
+  within run 1, an unrelated run) collided: restoring the edit point deleted
+  the write point too, and a later restore to that still-valid older point
+  returned "not found".
+- Fix: pruning now uses `MessageBoundary` when recorded -- a point is
+  superseded only by a strictly greater boundary, or an equal boundary
+  (parallel tool calls sharing one assistant message) captured later
+  (`created_at`); the target is always excluded. Falls back to the legacy
+  step predicate only when the target has no recorded boundary, matching
+  the existing message-truncation fallback.
+- Regression: `TestRestoreRewindPoint_PruneKeepsOlderPointsFromEarlierRuns`
+  drives two real runs, restores to run 2's edit point, then restores to
+  run 1's write point and asserts it still succeeds.

docs/runbooks/session-rewind.md

@@ -34,3 +34,5 @@ The confirmation token is required. Before confirming, ensure uncommitted work m
 Snapshots are captured before addressable `write`, `edit`, and `apply_patch` targets. Files over the per-file cap and points exceeding the per-conversation cap are listed as skipped and cannot be restored. Snapshot records are deleted automatically when their conversation is deleted or removed by retention.
 
 If restore returns an external-modification refusal, inspect or commit the current file first; use `force` only when losing that current content is intentional. Restoring an older point after a *later agent* edit to the same file needs no `force`: the expected hash for every earlier point sharing that path is kept current as the agent writes it, so `force` is only needed when the on-disk content actually diverges from the last agent-written state (issue #1371).
+
+Message truncation is keyed on the conversation-wide index of the assistant message that made the rewound tool call, not the tool-call step within its own run, so rewinding to a point from a later run in a multi-run conversation keeps every earlier run's messages and the chosen point's user prompt, deleting that assistant message and everything after it -- history never ends with an assistant message whose tool call has no response, which real providers reject on the next turn (issue #1370). Pruning of superseded rewind points uses the same conversation-wide boundary, so a point from an earlier, unrelated run is never deleted by a later run's restore. A rewind point captured before this fix has no recorded boundary and falls back to the old step-based truncation and pruning, logging a warning. A successful restore also invalidates the daemon's in-memory conversation history for that conversation, so `GET /messages` and the next run reflect the truncation immediately rather than only after a restart.

internal/harness/conversation_store_sqlite.go

@@ -5,6 +5,7 @@ import (
 	"database/sql"
 	"encoding/json"
 	"fmt"
+	"log"
 	"os"
 	"path/filepath"
 	"strings"
@@ -204,6 +205,15 @@ func (s *SQLiteConversationStore) Migrate(ctx context.Context) error {
 		}
 	}
 
+	// Idempotent migration: add message_boundary column to rewind_points if it
+	// doesn't exist (issue #1370). Existing rows default to 0, which restore
+	// treats as "not recorded" and falls back to the legacy step comparison.
+	if !s.columnExists(ctx, "rewind_points", "message_boundary") {
+		if _, err := s.db.ExecContext(ctx, `ALTER TABLE rewind_points ADD COLUMN message_boundary INTEGER NOT NULL DEFAULT 0`); err != nil {
+			return fmt.Errorf("migrate add message_boundary column: %w", err)
+		}
+	}
+
 	// Idempotent migration: create FTS5 triggers if they don't exist.
 	// Triggers keep conversation_messages_fts in sync with conversation_messages.
 	triggers := []string{
@@ -559,7 +569,7 @@ func (s *SQLiteConversationStore) SaveRewindPoint(ctx context.Context, point Rew
 	if _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO conversations (id, created_at, updated_at) VALUES (?, ?, ?)`, point.ConversationID, created.Format(time.RFC3339Nano), created.Format(time.RFC3339Nano)); err != nil {
 		return fmt.Errorf("create rewind conversation: %w", err)
 	}
-	if _, err := tx.ExecContext(ctx, `INSERT INTO rewind_points (id, conversation_id, step, tool, created_at) VALUES (?, ?, ?, ?, ?)`, point.ID, point.ConversationID, point.Step, point.Tool, created.Format(time.RFC3339Nano)); err != nil {
+	if _, err := tx.ExecContext(ctx, `INSERT INTO rewind_points (id, conversation_id, step, tool, created_at, message_boundary) VALUES (?, ?, ?, ?, ?, ?)`, point.ID, point.ConversationID, point.Step, point.Tool, created.Format(time.RFC3339Nano), point.MessageBoundary); err != nil {
 		return fmt.Errorf("insert rewind point: %w", err)
 	}
 	stmt, err := tx.PrepareContext(ctx, `INSERT INTO rewind_file_snapshots (point_id, path, content, existed, skipped, skip_reason, expected_hash) VALUES (?, ?, ?, ?, ?, ?, ?)`)
@@ -655,7 +665,7 @@ func (s *SQLiteConversationStore) FinalizeRewindPoint(ctx context.Context, point
 
 // ListRewindPoints returns newest rewind points first with their captured files.
 func (s *SQLiteConversationStore) ListRewindPoints(ctx context.Context, convID string) ([]RewindPoint, error) {
-	rows, err := s.db.QueryContext(ctx, `SELECT p.id, p.step, p.tool, p.created_at, f.path, f.content, COALESCE(f.existed,0), COALESCE(f.skipped,0), COALESCE(f.skip_reason,''), COALESCE(f.expected_hash,'') FROM rewind_points p LEFT JOIN rewind_file_snapshots f ON f.point_id=p.id WHERE p.conversation_id=? ORDER BY p.step DESC, p.created_at DESC, f.id ASC`, convID)
+	rows, err := s.db.QueryContext(ctx, `SELECT p.id, p.step, p.tool, p.created_at, COALESCE(p.message_boundary,0), f.path, f.content, COALESCE(f.existed,0), COALESCE(f.skipped,0), COALESCE(f.skip_reason,''), COALESCE(f.expected_hash,'') FROM rewind_points p LEFT JOIN rewind_file_snapshots f ON f.point_id=p.id WHERE p.conversation_id=? ORDER BY p.step DESC, p.created_at DESC, f.id ASC`, convID)
 	if err != nil {
 		return nil, fmt.Errorf("list rewind points: %w", err)
 	}
@@ -665,17 +675,17 @@ func (s *SQLiteConversationStore) ListRewindPoints(ctx context.Context, convID s
 	for rows.Next() {
 		var id, tool, created, reason, expected string
 		var path sql.NullString
-		var step, existed, skipped int
+		var step, messageBoundary, existed, skipped int
 		var content []byte
-		if err := rows.Scan(&id, &step, &tool, &created, &path, &content, &existed, &skipped, &reason, &expected); err != nil {
+		if err := rows.Scan(&id, &step, &tool, &created, &messageBoundary, &path, &content, &existed, &skipped, &reason, &expected); err != nil {
 			return nil, fmt.Errorf("scan rewind point: %w", err)
 		}
 		i, ok := byID[id]
 		if !ok {
 			t, _ := time.Parse(time.RFC3339Nano, created)
 			i = len(points)
 			byID[id] = i
-			points = append(points, RewindPoint{ID: id, ConversationID: convID, Step: step, Tool: tool, CreatedAt: t})
+			points = append(points, RewindPoint{ID: id, ConversationID: convID, Step: step, Tool: tool, CreatedAt: t, MessageBoundary: messageBoundary})
 		}
 		if path.Valid && path.String != "" {
 			points[i].Files = append(points[i].Files, RewindFileSnapshot{Path: path.String, Content: content, Exists: existed == 1, Skipped: skipped == 1, SkipReason: reason, ExpectedHash: expected})
@@ -757,13 +767,43 @@ func (s *SQLiteConversationStore) RestoreRewindPoint(ctx context.Context, convID
 		return result, fmt.Errorf("rewind begin tx: %w", err)
 	}
 	defer tx.Rollback()
-	res, err := tx.ExecContext(ctx, `DELETE FROM conversation_messages WHERE conversation_id=? AND step>?`, convID, point.Step)
+	// point.Step is a run-local tool-call counter (see runner_step_engine.go),
+	// not comparable to conversation_messages.step, which is a
+	// conversation-wide message index shared across every run on this
+	// conversation (issue #1370). point.MessageBoundary records that
+	// conversation-wide index at capture time and is the only field safe to
+	// truncate against. Points captured before this field existed have
+	// MessageBoundary==0 ("not recorded"); rather than silently deleting
+	// everything, fall back to the legacy (imperfect, multi-run-unsafe)
+	// step comparison and log a warning so operators can see it happened.
+	var res sql.Result
+	if point.MessageBoundary > 0 {
+		res, err = tx.ExecContext(ctx, `DELETE FROM conversation_messages WHERE conversation_id=? AND step>=?`, convID, point.MessageBoundary)
+	} else {
+		log.Printf("rewind: point %q (conversation %q) has no recorded message boundary; falling back to step-based truncation, which can delete unrelated messages in a multi-run conversation", point.ID, convID)
+		res, err = tx.ExecContext(ctx, `DELETE FROM conversation_messages WHERE conversation_id=? AND step>?`, convID, point.Step)
+	}
 	if err != nil {
 		return result, fmt.Errorf("rewind truncate messages: %w", err)
 	}
 	n, _ := res.RowsAffected()
 	result.MessagesTruncated = int(n)
-	if _, err := tx.ExecContext(ctx, `DELETE FROM rewind_points WHERE conversation_id=? AND (step>? OR (step=? AND id<>?))`, convID, point.Step, point.Step, point.ID); err != nil {
+	// point.Step is run-local and restarts every run, so it collides across
+	// runs (run 2's first mutating call and run 1's first mutating call are
+	// both step 0/1 within their own run but capture entirely unrelated
+	// points): pruning by it deleted an older, still-valid point from an
+	// earlier run whenever a later run's point happened to share a step
+	// number. Prune by MessageBoundary instead when it is recorded: a point
+	// is superseded only if it comes after the target in conversation order
+	// (a strictly greater boundary), or shares the same boundary (parallel
+	// tool calls in one assistant turn) but was captured later. The target
+	// itself is always excluded. Fall back to the legacy step predicate only
+	// when the target has no recorded boundary.
+	if point.MessageBoundary > 0 {
+		if _, err := tx.ExecContext(ctx, `DELETE FROM rewind_points WHERE conversation_id=? AND id<>? AND (message_boundary>? OR (message_boundary=? AND created_at>?))`, convID, point.ID, point.MessageBoundary, point.MessageBoundary, point.CreatedAt.Format(time.RFC3339Nano)); err != nil {
+			return result, fmt.Errorf("rewind delete future points: %w", err)
+		}
+	} else if _, err := tx.ExecContext(ctx, `DELETE FROM rewind_points WHERE conversation_id=? AND (step>? OR (step=? AND id<>?))`, convID, point.Step, point.Step, point.ID); err != nil {
 		return result, fmt.Errorf("rewind delete future points: %w", err)
 	}
 	// The files just restored now hold this content on disk, so every

internal/harness/rewind.go

@@ -73,6 +73,14 @@ type RewindPoint struct {
 	Tool           string               `json:"tool"`
 	CreatedAt      time.Time            `json:"created_at"`
 	Files          []RewindFileSnapshot `json:"files"`
+	// MessageBoundary is the conversation-wide message count captured at the
+	// moment this point was recorded (the number of conversation_messages
+	// rows that must survive a restore). Step is a run-local tool-call
+	// counter and is NOT comparable to conversation_messages.step across
+	// runs (issue #1370); MessageBoundary is. Zero means "not recorded"
+	// (points captured before this field existed), in which case restore
+	// falls back to the legacy step-based comparison.
+	MessageBoundary int `json:"message_boundary,omitempty"`
 }
 
 // RewindStore is deliberately optional so existing ConversationStore adapters