| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "net/http" |
| 8 | "net/http/httptest" |
| 9 | "path/filepath" |
| 10 | "strings" |
| 11 | "testing" |
| 12 | "time" |
| 13 | |
| 14 | "reasonix/internal/agent" |
| 15 | "reasonix/internal/control" |
| 16 | "reasonix/internal/event" |
| 17 | "reasonix/internal/provider" |
| 18 | ) |
| 19 | |
| 20 | func TestMarkLocalTakeoverSpectatorPublishesAndPersistsReclaimState(t *testing.T) { |
| 21 | app := NewApp() |
| 22 | tab := &WorkspaceTab{ID: "spectator", Scope: "global", ReadOnly: true} |
| 23 | app.mu.Lock() |
| 24 | app.tabs[tab.ID] = tab |
| 25 | app.tabOrder = []string{tab.ID} |
| 26 | app.mu.Unlock() |
| 27 | |
| 28 | app.markLocalTakeoverSpectator(tab) |
| 29 | meta := app.tabMeta(tab, true) |
| 30 | if !tab.Takeover.Spectator || !meta.TakenOver || !meta.ReadOnly { |
| 31 | t.Fatalf("spectator state = tab=%v meta=%+v", tab.Takeover.Spectator, meta) |
| 32 | } |
| 33 | app.mu.Lock() |
| 34 | _, entries, _, _ := app.saveTabsCollectLocked() |
| 35 | app.mu.Unlock() |
| 36 | if len(entries) != 1 || !entries[0].TakeoverSpectator || !entries[0].ReadOnly { |
| 37 | t.Fatalf("persisted spectator entries = %+v", entries) |
| 38 | } |
| 39 | } |
| 40 | |
| 41 | func TestTakeoverSessionPromotesLocalSpectatorFromFreshDiskState(t *testing.T) { |
| 42 | isolateDesktopUserDirs(t) |
| 43 | dir := t.TempDir() |
| 44 | path := filepath.Join(dir, "round-trip.jsonl") |
| 45 | session := agent.NewSession("system") |
| 46 | session.Add(provider.Message{Role: provider.RoleUser, Content: "remote latest"}) |
| 47 | if err := session.Save(path); err != nil { |
| 48 | t.Fatal(err) |
| 49 | } |
| 50 | sourceLease, err := agent.TryAcquireSessionLease(path) |
| 51 | if err != nil { |
| 52 | t.Fatal(err) |
| 53 | } |
| 54 | sourceWriterID := agent.SessionWriterID() |
| 55 | |
| 56 | oldCtrl := control.New(control.Options{ |
| 57 | Executor: agent.New(nil, nil, agent.NewSession("stale"), agent.Options{}, event.Discard), |
| 58 | SessionDir: dir, |
| 59 | SessionPath: path, |
| 60 | }) |
| 61 | tab := &WorkspaceTab{ |
| 62 | ID: "local-spectator", Scope: "global", SessionPath: path, |
| 63 | ReadOnly: true, Ctrl: oldCtrl, Ready: true, |
| 64 | } |
| 65 | tab.Takeover.Spectator = true |
| 66 | app := NewApp() |
| 67 | app.ctx = context.Background() |
| 68 | tab.sink = &tabEventSink{tabID: tab.ID, app: app, ctx: app.ctx} |
| 69 | app.mu.Lock() |
| 70 | app.tabs[tab.ID] = tab |
| 71 | app.tabOrder = []string{tab.ID} |
| 72 | app.activeTabID = tab.ID |
| 73 | app.newSessionRuntimeLocked(tab, sessionRuntimeKey(path)) |
| 74 | app.advanceSessionRuntimeEpochLocked(tab) |
| 75 | app.mu.Unlock() |
| 76 | t.Cleanup(func() { |
| 77 | if mirror := app.takeoverMirrorForKey(sessionRuntimeKey(path)); mirror != nil { |
| 78 | mirror.stopAndFinalize(false) |
| 79 | } |
| 80 | app.mu.RLock() |
| 81 | ctrl := tab.Ctrl |
| 82 | app.mu.RUnlock() |
| 83 | if ctrl != nil { |
| 84 | ctrl.Close() |
| 85 | } |
| 86 | tab.releaseSessionLease() |
| 87 | sourceLease.Release() |
| 88 | }) |
| 89 | |
| 90 | srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 91 | switch r.URL.Path { |
| 92 | case "/handoff": |
| 93 | if err := sourceLease.ReleaseForHandoff(sourceWriterID, "forward"); err != nil { |
| 94 | t.Error(err) |
| 95 | http.Error(w, err.Error(), http.StatusInternalServerError) |
| 96 | return |
| 97 | } |
| 98 | _ = json.NewEncoder(w).Encode(takeoverGrant{ |
| 99 | SessionPath: path, MirrorID: "mirror", HandoffID: "forward", ReturnHandoffID: "return", |
| 100 | SourceWriterID: sourceWriterID, TargetWriterID: agent.SessionWriterID(), |
| 101 | }) |
| 102 | case "/external/frames": |
| 103 | _ = json.NewEncoder(w).Encode(map[string]bool{"reclaimRequested": false}) |
| 104 | case "/mirror-end": |
| 105 | w.WriteHeader(http.StatusNoContent) |
| 106 | default: |
| 107 | w.WriteHeader(http.StatusNotFound) |
| 108 | } |
| 109 | })) |
| 110 | defer srv.Close() |
| 111 | |
| 112 | originalFind := takeoverFindTargetForTest |
| 113 | takeoverFindTargetForTest = func(context.Context, *App, string) (takeoverServeRecord, *http.Client, SessionTakeoverView, error) { |
| 114 | return takeoverServeRecord{base: srv.URL}, srv.Client(), SessionTakeoverView{Holder: "serve"}, nil |
| 115 | } |
| 116 | originalBuild := takeoverBuildLocalSpectatorCandidateForTest |
| 117 | loadedFreshState := false |
| 118 | takeoverBuildLocalSpectatorCandidateForTest = func(a *App, candidateTab *WorkspaceTab, source tabRuntimeSnapshot, candidatePath string, loaded *agent.Session) (*sessionRebindCandidate, error) { |
| 119 | for _, message := range loaded.Snapshot() { |
| 120 | if message.Role == provider.RoleUser && message.Content == "remote latest" { |
| 121 | loadedFreshState = true |
| 122 | } |
| 123 | } |
| 124 | ctrl := control.New(control.Options{ |
| 125 | Executor: agent.New(nil, nil, loaded, agent.Options{}, event.Discard), |
| 126 | SessionDir: dir, SessionPath: candidatePath, |
| 127 | }) |
| 128 | return &sessionRebindCandidate{ |
| 129 | app: a, ctrl: ctrl, sink: &tabEventSink{tabID: candidateTab.ID, app: a}, |
| 130 | model: source.model, runtime: source.normalizedRuntime(), |
| 131 | }, nil |
| 132 | } |
| 133 | t.Cleanup(func() { |
| 134 | takeoverFindTargetForTest = originalFind |
| 135 | takeoverBuildLocalSpectatorCandidateForTest = originalBuild |
| 136 | }) |
| 137 | |
| 138 | if err := app.TakeoverSession(tab.ID, "wait"); err != nil { |
| 139 | t.Fatal(err) |
| 140 | } |
| 141 | if !loadedFreshState { |
| 142 | t.Fatal("replacement did not reload the transcript after targeted lease acquisition") |
| 143 | } |
| 144 | app.mu.RLock() |
| 145 | replacement := tab.Ctrl |
| 146 | readOnly, spectator := tab.ReadOnly, tab.Takeover.Spectator |
| 147 | app.mu.RUnlock() |
| 148 | if replacement == nil || replacement == oldCtrl || readOnly || spectator { |
| 149 | t.Fatalf("promoted tab = ctrl %p old %p readOnly=%v spectator=%v", replacement, oldCtrl, readOnly, spectator) |
| 150 | } |
| 151 | if got := tab.sessionLeaseRuntimeKey(); got != sessionRuntimeKey(path) { |
| 152 | t.Fatalf("promoted lease = %q, want %q", got, sessionRuntimeKey(path)) |
| 153 | } |
| 154 | if meta := app.tabMeta(tab, true); meta.TakenOver || meta.ReadOnly { |
| 155 | t.Fatalf("promoted meta remained read-only: %+v", meta) |
| 156 | } |
| 157 | } |
| 158 | |
| 159 | func TestTakeoverSessionBuildFailureKeepsLocalSpectatorAndReturnsLease(t *testing.T) { |
| 160 | isolateDesktopUserDirs(t) |
| 161 | dir := t.TempDir() |
| 162 | path := filepath.Join(dir, "failed-round-trip.jsonl") |
| 163 | session := agent.NewSession("system") |
| 164 | if err := session.Save(path); err != nil { |
| 165 | t.Fatal(err) |
| 166 | } |
| 167 | sourceLease, err := agent.TryAcquireSessionLease(path) |
| 168 | if err != nil { |
| 169 | t.Fatal(err) |
| 170 | } |
| 171 | sourceWriterID := agent.SessionWriterID() |
| 172 | oldCtrl := control.New(control.Options{ |
| 173 | Executor: agent.New(nil, nil, session, agent.Options{}, event.Discard), |
| 174 | SessionDir: dir, SessionPath: path, |
| 175 | }) |
| 176 | tab := &WorkspaceTab{ |
| 177 | ID: "failed-local-spectator", Scope: "global", SessionPath: path, |
| 178 | ReadOnly: true, Ctrl: oldCtrl, Ready: true, |
| 179 | } |
| 180 | tab.Takeover.Spectator = true |
| 181 | app := NewApp() |
| 182 | app.ctx = context.Background() |
| 183 | tab.sink = &tabEventSink{tabID: tab.ID, app: app, ctx: app.ctx} |
| 184 | app.mu.Lock() |
| 185 | app.tabs[tab.ID] = tab |
| 186 | app.newSessionRuntimeLocked(tab, sessionRuntimeKey(path)) |
| 187 | app.advanceSessionRuntimeEpochLocked(tab) |
| 188 | app.mu.Unlock() |
| 189 | |
| 190 | srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 191 | switch r.URL.Path { |
| 192 | case "/handoff": |
| 193 | if err := sourceLease.ReleaseForHandoff(sourceWriterID, "forward"); err != nil { |
| 194 | http.Error(w, err.Error(), http.StatusInternalServerError) |
| 195 | return |
| 196 | } |
| 197 | _ = json.NewEncoder(w).Encode(takeoverGrant{ |
| 198 | SessionPath: path, MirrorID: "failed-mirror", HandoffID: "forward", ReturnHandoffID: "return", |
| 199 | SourceWriterID: sourceWriterID, TargetWriterID: agent.SessionWriterID(), |
| 200 | }) |
| 201 | case "/mirror-end": |
| 202 | w.WriteHeader(http.StatusNoContent) |
| 203 | default: |
| 204 | _ = json.NewEncoder(w).Encode(map[string]bool{"reclaimRequested": false}) |
| 205 | } |
| 206 | })) |
| 207 | defer srv.Close() |
| 208 | |
| 209 | originalFind := takeoverFindTargetForTest |
| 210 | takeoverFindTargetForTest = func(context.Context, *App, string) (takeoverServeRecord, *http.Client, SessionTakeoverView, error) { |
| 211 | return takeoverServeRecord{base: srv.URL}, srv.Client(), SessionTakeoverView{Holder: "serve"}, nil |
| 212 | } |
| 213 | originalBuild := takeoverBuildLocalSpectatorCandidateForTest |
| 214 | buildErr := errors.New("injected rebuild failure") |
| 215 | takeoverBuildLocalSpectatorCandidateForTest = func(*App, *WorkspaceTab, tabRuntimeSnapshot, string, *agent.Session) (*sessionRebindCandidate, error) { |
| 216 | return nil, buildErr |
| 217 | } |
| 218 | t.Cleanup(func() { |
| 219 | takeoverFindTargetForTest = originalFind |
| 220 | takeoverBuildLocalSpectatorCandidateForTest = originalBuild |
| 221 | if mirror := app.takeoverMirrorForKey(sessionRuntimeKey(path)); mirror != nil { |
| 222 | mirror.stopAndFinalize(false) |
| 223 | } |
| 224 | oldCtrl.Close() |
| 225 | sourceLease.Release() |
| 226 | }) |
| 227 | |
| 228 | if err := app.TakeoverSession(tab.ID, "wait"); !errors.Is(err, buildErr) || !strings.Contains(err.Error(), "injected rebuild failure") { |
| 229 | t.Fatalf("TakeoverSession error = %v, want injected failure", err) |
| 230 | } |
| 231 | app.mu.RLock() |
| 232 | gotCtrl, readOnly, spectator := tab.Ctrl, tab.ReadOnly, tab.Takeover.Spectator |
| 233 | app.mu.RUnlock() |
| 234 | if gotCtrl != oldCtrl || !readOnly || !spectator || tab.sessionLeaseRuntimeKey() != "" { |
| 235 | t.Fatalf("failed promotion changed spectator: ctrl=%p old=%p readOnly=%v spectator=%v lease=%q", gotCtrl, oldCtrl, readOnly, spectator, tab.sessionLeaseRuntimeKey()) |
| 236 | } |
| 237 | |
| 238 | deadline := time.Now().Add(5 * time.Second) |
| 239 | for { |
| 240 | info, loadErr := agent.LoadSessionLeaseInfo(path) |
| 241 | if loadErr == nil && info != nil && info.HandoffTo == sourceWriterID && info.HandoffID == "return" { |
| 242 | break |
| 243 | } |
| 244 | if time.Now().After(deadline) { |
| 245 | t.Fatalf("failed promotion did not publish reverse reservation: info=%+v err=%v", info, loadErr) |
| 246 | } |
| 247 | time.Sleep(10 * time.Millisecond) |
| 248 | } |
| 249 | } |
| 250 |