feat(serve): keep the trajectory, so a reopened session still has one

The pane was built from the live stream alone, so re-entering a session showed
an empty table. Nothing was on disk to reload: trajectory.Recorder is wired
into the CLI and e2ebench, never into serve, and the .events.jsonl beside each
session is a conversation projection with no timings, usage or audit records.

Serve now mirrors the broadcast frames it already hands the SSE clients into
<session>.wire.jsonl, and GET /trajectory reads them back. Frames are what the
pane consumes, so replaying them reproduces it row for row with no second
schema in between; internal/trajectory keeps recording its richer audit stream
for offline analysis, which answers a different question.

Only kinds the pane draws are kept. Streamed text and reasoning arrive one
frame per chunk and would be nearly the whole file while contributing no rows,
and dropping them is what makes it affordable to close the handle after every
write — a retained one blocks the session directory from being removed on
Windows, which is how TestConcurrentResumeAndForkKeepAlignment caught it.

Deleting the active session is refused by the kernel with 409, but the rail
collapsed the row and reloaded regardless, so a rejected delete looked like it
had worked. The failure now restores the row and surfaces the reason.

desktop/frontend-next/src/port/mock.ts

@@ -206,6 +206,10 @@ export class MockPort implements AgentPort {
     return { ...this.state };
   }
 
+  async trajectory(): Promise<WireEvent[]> {
+    return [];
+  }
+
   async history(): Promise<HistoryMessage[]> {
     return [];
   }

desktop/frontend-next/src/port/port.ts

@@ -89,6 +89,9 @@ export interface AgentPort {
   deleteSession(name: string): Promise<void>;
   status(): Promise<SessionStatus>;
   history(): Promise<HistoryMessage[]>;
+  // Replaying the persisted wire frames rebuilds the trajectory pane row for
+  // row; the live stream only ever covers the current connection.
+  trajectory(): Promise<WireEvent[]>;
   subscribe(onEvent: (ev: WireEvent) => void): () => void;
 
   submit(text: string): Promise<void>;

desktop/frontend-next/src/port/sse.ts

@@ -66,6 +66,10 @@ export class SsePort implements AgentPort {
     return this.post("/delete-session", { name });
   }
 
+  trajectory() {
+    return this.get<WireEvent[]>("/trajectory");
+  }
+
   history() {
     return this.get<HistoryMessage[]>("/history");
   }

desktop/frontend-next/src/ui/App.tsx

@@ -46,6 +46,7 @@ export function App({ port }: { port: AgentPort }) {
 
   useEffect(() => {
     let alive = true;
+    port.trajectory().then((evs) => alive && evs.forEach((e) => trajDispatch(e))).catch(() => {});
     Promise.all([port.history(), port.status()]).then(([msgs, st]) => {
       if (!alive) return;
       setStatus(st);
@@ -175,10 +176,12 @@ export function App({ port }: { port: AgentPort }) {
             status={status}
             run={run}
             cost={`${s.metrics.currency}${s.metrics.cost.toFixed(2)}`}
+            onError={fail}
             onFold={() => setRail(false)}
             onSwitched={() => {
               refreshStatus();
               trajDispatch({ kind: "__clear" });
+              port.trajectory().then((evs) => evs.forEach((e) => trajDispatch(e))).catch(() => {});
               port.history().then((msgs) => {
                 const r = fromHistory(msgs);
                 dispatch({ kind: "__restore", items: r.items, plan: r.plan, hit: 0, miss: 0 } as never);

desktop/frontend-next/src/ui/Sessions.tsx

@@ -6,6 +6,7 @@ interface Props {
   status: SessionStatus | null;
   run: string;
   cost: string;
+  onError: (e: unknown) => void;
   onFold: () => void;
   onSwitched: () => void;
 }
@@ -13,7 +14,7 @@ interface Props {
 const RUN_ST: Record<string, string> = { running: "运行中", halt: "等你", done: "已完成", idle: "空闲" };
 const WS_STATE: Record<string, string> = { running: "running", halt: "awaiting", done: "done" };
 
-export function Sessions({ port, status, run, cost, onFold, onSwitched }: Props) {
+export function Sessions({ port, status, run, cost, onError, onFold, onSwitched }: Props) {
   const [list, setList] = useState<SessionEntry[]>([]);
   const [busy, setBusy] = useState("");
   const [confirm, setConfirm] = useState("");
@@ -56,12 +57,18 @@ export function Sessions({ port, status, run, cost, onFold, onSwitched }: Props)
     setGone(e.path);
     try {
       await port.deleteSession(e.name);
-    } finally {
-      setTimeout(() => {
-        setGone("");
-        load();
-      }, 220);
+    } catch (err) {
+      // The kernel refuses to delete the session it is driving. Put the row
+      // back rather than let the collapse imply it worked.
+      if (row) row.style.height = "";
+      setGone("");
+      onError(err);
+      return;
     }
+    setTimeout(() => {
+      setGone("");
+      load();
+    }, 220);
   };
 
   return (

internal/serve/serve.go

@@ -68,6 +68,7 @@ type Server struct {
 	titleUsageSink    event.Sink
 	titles            *titleCache
 	fill              *titleFiller
+	wire              *wireLog
 	auth              *authGate // nil when auth is disabled
 	providerSetupMu   sync.RWMutex
 	providerSetup     providerSetupState
@@ -88,6 +89,7 @@ func New(ctrl control.SessionAPI, bc *Broadcaster, serveCfg config.ServeConfig)
 		ctrl:   ctrl,
 		bc:     bc,
 		titles: newTitleCache(ctrl.SessionDir()),
+		wire:   &wireLog{},
 		fill:   newTitleFiller(),
 		auth:   newAuthGate(serveCfg),
 	}
@@ -95,6 +97,7 @@ func New(ctrl control.SessionAPI, bc *Broadcaster, serveCfg config.ServeConfig)
 		bc.SetDisplayCurrency(cfg.ExplicitDisplayCurrency())
 	}
 	s.initTitleProvider()
+	s.attachWireLog()
 	return s
 }
 
@@ -517,6 +520,7 @@ func (s *Server) handler() http.Handler {
 	mux.HandleFunc("POST /extensions/reload", s.reloadExtensionsHTTP)
 	mux.HandleFunc("GET /status", s.status)
 	mux.HandleFunc("GET /sessions", s.sessions)
+	mux.HandleFunc("GET /trajectory", s.trajectory)
 	mux.HandleFunc("GET /skills", s.skills)
 	mux.HandleFunc("GET /todos", s.todos)
 	mux.HandleFunc("POST /delete-session", s.deleteSession)

internal/serve/wirelog.go

@@ -0,0 +1,105 @@
+package serve
+
+import (
+	"encoding/json"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+)
+
+// The trajectory pane is built from wire frames, so the frames are what gets
+// kept: replaying them reproduces the pane row for row, with no second schema
+// to map through. internal/trajectory records a richer audit stream for offline
+// analysis and answers a different question.
+const wireLogMaxBytes = 8 << 20
+
+// Only the kinds the pane draws a row for. Streamed text and reasoning arrive
+// one frame per chunk and would be most of the file while contributing nothing
+// to read back; keeping them out is what lets each write close its handle.
+var wireLogKinds = map[string]bool{
+	"turn_started": true, "turn_done": true, "message": true,
+	"tool_dispatch": true, "tool_result": true, "usage": true,
+	"guardian_assessment": true, "approval_request": true, "ask_request": true,
+	"compaction_started": true, "compaction_done": true, "retrying": true,
+	"steer": true, "context_maintenance": true, "completion_summary": true,
+	"notice": true,
+}
+
+type wireLog struct {
+	mu sync.Mutex
+}
+
+func wireLogPath(sessionPath string) string {
+	if strings.TrimSpace(sessionPath) == "" {
+		return ""
+	}
+	return strings.TrimSuffix(sessionPath, filepath.Ext(sessionPath)) + ".wire.jsonl"
+}
+
+// attachWireLog mirrors qualifying broadcast frames to the current session's
+// log. Subscribing is how the SSE handler reads the same stream, so the log
+// sees exactly what a connected client would have.
+func (s *Server) attachWireLog() {
+	ch, unsubscribe := s.bc.Subscribe()
+	go func() {
+		defer unsubscribe()
+		for frame := range ch {
+			// Path is read per frame, not cached: /resume and /new swap it
+			// underneath and the next row belongs to the new session.
+			s.wire.write(wireLogPath(s.ctl().SessionPath()), frame)
+		}
+	}()
+}
+
+func (w *wireLog) write(path string, frame []byte) {
+	if path == "" {
+		return
+	}
+	var head struct {
+		Kind string `json:"kind"`
+	}
+	if err := json.Unmarshal(frame, &head); err != nil || !wireLogKinds[head.Kind] {
+		return
+	}
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	// Reopened per write rather than held: a retained handle blocks the session
+	// file's directory from being removed on Windows, and the filtered volume
+	// makes the cost irrelevant.
+	f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
+	if err != nil {
+		return
+	}
+	defer f.Close()
+	// A runaway turn must not fill the disk; drop rather than truncate, so what
+	// is on disk stays a prefix of what happened.
+	if st, err := f.Stat(); err == nil && st.Size() > wireLogMaxBytes {
+		return
+	}
+	_, _ = f.Write(append(frame, '\n'))
+}
+
+func (s *Server) trajectory(w http.ResponseWriter, _ *http.Request) {
+	path := wireLogPath(s.ctl().SessionPath())
+	if path == "" {
+		writeJSON(w, []any{})
+		return
+	}
+	s.wire.mu.Lock()
+	data, err := os.ReadFile(path)
+	s.wire.mu.Unlock()
+	if err != nil {
+		writeJSON(w, []any{})
+		return
+	}
+	out := []json.RawMessage{}
+	for _, line := range strings.Split(strings.TrimRight(string(data), "\n"), "\n") {
+		if strings.TrimSpace(line) == "" {
+			continue
+		}
+		out = append(out, json.RawMessage(line))
+	}
+	writeJSON(w, out)
+}