| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "bufio" |
| 5 | "context" |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "net/http" |
| 9 | "net/http/httptest" |
| 10 | "os" |
| 11 | "reasonix/internal/control" |
| 12 | "reasonix/internal/proc" |
| 13 | "reasonix/internal/provider" |
| 14 | "strings" |
| 15 | "testing" |
| 16 | "time" |
| 17 | ) |
| 18 | |
| 19 | type modelObservedOutput struct { |
| 20 | writer io.Writer |
| 21 | wrote chan<- struct{} |
| 22 | } |
| 23 | |
| 24 | func (w modelObservedOutput) Write(p []byte) (int, error) { |
| 25 | n, err := w.writer.Write(p) |
| 26 | select { |
| 27 | case w.wrote <- struct{}{}: |
| 28 | default: |
| 29 | } |
| 30 | return n, err |
| 31 | } |
| 32 | |
| 33 | func TestModelGatewayProcess(t *testing.T) { |
| 34 | if os.Getenv("REASONIX_TEST_GATEWAY_PROCESS") != "1" { |
| 35 | return |
| 36 | } |
| 37 | scanner := bufio.NewScanner(os.Stdin) |
| 38 | for scanner.Scan() { |
| 39 | fmt.Println(scanner.Text()) |
| 40 | } |
| 41 | os.Exit(0) |
| 42 | } |
| 43 | |
| 44 | func modelApplicationFixture(t *testing.T) (*App, *WorkspaceTab, ProviderView, <-chan struct{}) { |
| 45 | t.Helper() |
| 46 | isolateDesktopUserDirs(t) |
| 47 | called := make(chan struct{}, 16) |
| 48 | server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 49 | io.Copy(io.Discard, r.Body) |
| 50 | called <- struct{}{} |
| 51 | w.Header().Set("Content-Type", "text/event-stream") |
| 52 | fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{\"content\":\"ok\"},\"finish_reason\":\"stop\"}]}\n\ndata: [DONE]\n\n") |
| 53 | })) |
| 54 | t.Cleanup(server.Close) |
| 55 | a := NewApp() |
| 56 | t.Cleanup(a.closeSessionServices) |
| 57 | view := ProviderView{Name: "gateway-model", Kind: "openai", BaseURL: server.URL, Models: []string{"m"}, NoProxy: true} |
| 58 | if _, err := a.SaveProviderWithKey(view, "fixture-key"); err != nil { |
| 59 | t.Fatal(err) |
| 60 | } |
| 61 | a.ctx = t.Context() |
| 62 | a.readyHook = func() {} |
| 63 | tab := modelSettingsBootTab(t, a, "gateway", t.TempDir(), "gateway-model/m") |
| 64 | tab.toolApprovalMode = tab.Ctrl.ToolApprovalMode() |
| 65 | a.activeTabID = tab.ID |
| 66 | return a, tab, view, called |
| 67 | } |
| 68 | |
| 69 | func TestModelSettingsGatewaySurvivesReplacementAndFailure(t *testing.T) { |
| 70 | a, tab, view, called := modelApplicationFixture(t) |
| 71 | old := tab.Ctrl.(*control.Controller) |
| 72 | scope := old.BackgroundScope() |
| 73 | ref, _ := old.SessionRef() |
| 74 | type process struct { |
| 75 | pid int |
| 76 | input io.WriteCloser |
| 77 | } |
| 78 | ready := make(chan process, 1) |
| 79 | exited := make(chan struct{}) |
| 80 | wrote := make(chan struct{}, 16) |
| 81 | job, err := scope.Manager.TryStartSessionProcess(ref.SessionID, "bash", "gateway", func(ctx context.Context, out io.Writer) (string, error) { |
| 82 | defer close(exited) |
| 83 | cmd := proc.CommandContext(ctx, os.Args[0], "-test.run=^TestModelGatewayProcess$") |
| 84 | cmd.Env = append(os.Environ(), "REASONIX_TEST_GATEWAY_PROCESS=1") |
| 85 | cmd.Stdout = modelObservedOutput{writer: out, wrote: wrote} |
| 86 | cmd.Stderr = out |
| 87 | input, err := cmd.StdinPipe() |
| 88 | if err != nil { |
| 89 | return "", err |
| 90 | } |
| 91 | if err = cmd.Start(); err != nil { |
| 92 | return "", err |
| 93 | } |
| 94 | ready <- process{cmd.Process.Pid, input} |
| 95 | return "", cmd.Wait() |
| 96 | }) |
| 97 | if err != nil { |
| 98 | t.Fatal(err) |
| 99 | } |
| 100 | var p process |
| 101 | select { |
| 102 | case p = <-ready: |
| 103 | case <-time.After(10 * time.Second): |
| 104 | t.Fatal("process failed to start") |
| 105 | } |
| 106 | defer p.input.Close() |
| 107 | for i := range 2 { |
| 108 | // Query changes leave the fixture endpoint valid while changing the resolver. |
| 109 | view.BaseURL = strings.Split(view.BaseURL, "?")[0] + fmt.Sprintf("?generation=%d", i) |
| 110 | if _, err := a.SaveProviderWithKey(view, "fixture-key"); err != nil { |
| 111 | t.Fatal(err) |
| 112 | } |
| 113 | before := tab.Ctrl |
| 114 | admission, current, err := a.beginTabTurn(tab.ID, false) |
| 115 | if err != nil { |
| 116 | t.Fatal(err) |
| 117 | } |
| 118 | admission.abort() |
| 119 | next := current.(*control.Controller) |
| 120 | nextRef, _ := next.SessionRef() |
| 121 | if current == before || next.BackgroundScope() != scope || next.RuntimeStatus().BackgroundJobs != 1 { |
| 122 | output, status, _ := scope.Manager.OutputForSession(ref.SessionID, job.ID) |
| 123 | t.Logf("gateway status=%s output=%s", status, output) |
| 124 | t.Fatalf("replacement lost live gateway: changed=%v sameScope=%v status=%+v all=%+v ref=%+v next=%+v", current != before, next.BackgroundScope() == scope, next.RuntimeStatus(), scope.Manager.Running(), ref, nextRef) |
| 125 | } |
| 126 | select { |
| 127 | case <-exited: |
| 128 | t.Fatal("gateway exited during replacement") |
| 129 | default: |
| 130 | } |
| 131 | if _, err := fmt.Fprintf(p.input, "pid=%d generation=%d\n", p.pid, i); err != nil { |
| 132 | t.Fatal(err) |
| 133 | } |
| 134 | marker := fmt.Sprintf("pid=%d generation=%d", p.pid, i) |
| 135 | deadline := time.After(3 * time.Second) |
| 136 | for { |
| 137 | output, _, _ := scope.Manager.OutputForSession(ref.SessionID, job.ID) |
| 138 | if strings.Contains(output, marker) { |
| 139 | break |
| 140 | } |
| 141 | select { |
| 142 | case <-wrote: |
| 143 | case <-deadline: |
| 144 | t.Fatal("gateway output was lost after replacement") |
| 145 | } |
| 146 | } |
| 147 | if err := next.RunTurn(t.Context(), "answer briefly"); err != nil { |
| 148 | t.Fatal(err) |
| 149 | } |
| 150 | select { |
| 151 | case <-called: |
| 152 | case <-time.After(time.Second): |
| 153 | t.Fatal("new model endpoint was not called") |
| 154 | } |
| 155 | users := 0 |
| 156 | for _, message := range next.History() { |
| 157 | if message.Role == provider.RoleUser && message.Content == "answer briefly" { |
| 158 | users++ |
| 159 | } |
| 160 | } |
| 161 | if users != i+1 { |
| 162 | t.Fatalf("history lost across replacement: users=%d", users) |
| 163 | } |
| 164 | } |
| 165 | if !tab.Ctrl.(*control.Controller).CancelJob(job.ID) { |
| 166 | t.Fatal("replacement cannot stop original job") |
| 167 | } |
| 168 | select { |
| 169 | case <-exited: |
| 170 | case <-time.After(10 * time.Second): |
| 171 | t.Fatal("gateway did not exit") |
| 172 | } |
| 173 | scope.Manager.WaitForSession(t.Context(), ref.SessionID, []string{job.ID}, 10) |
| 174 | if got := scope.Manager.RunningForSession(ref.SessionID); len(got) != 0 { |
| 175 | t.Fatalf("running jobs after exit: %+v", got) |
| 176 | } |
| 177 | } |
| 178 | |
| 179 | func TestModelSettingsRuntimeTaskChoiceIsSingleUseAndRevocationSafe(t *testing.T) { |
| 180 | a, tab, view, _ := modelApplicationFixture(t) |
| 181 | c := tab.Ctrl.(*control.Controller) |
| 182 | ref, _ := c.SessionRef() |
| 183 | started := make(chan struct{}) |
| 184 | job := c.BackgroundScope().Manager.StartForSession(ref.SessionID, "task", "dependent", func(ctx context.Context, _ io.Writer) (string, error) { |
| 185 | close(started) |
| 186 | <-ctx.Done() |
| 187 | return "", ctx.Err() |
| 188 | }) |
| 189 | <-started |
| 190 | view.BaseURL = strings.Replace(view.BaseURL, "127.0.0.1", "localhost", 1) |
| 191 | if _, err := a.SaveProviderWithKey(view, "fixture-key"); err != nil { |
| 192 | t.Fatal(err) |
| 193 | } |
| 194 | if admission, _, err := a.beginTabTurn(tab.ID, false); err == nil { |
| 195 | admission.abort() |
| 196 | t.Fatal("dependent task allowed rebuild") |
| 197 | } |
| 198 | d := modelApplicationDetails(c) |
| 199 | if !d.CanUseApplied || len(d.BlockingJobs) != 1 { |
| 200 | t.Fatalf("recovery details: %+v", d) |
| 201 | } |
| 202 | choice := control.ModelApplicationChoice{Mode: "applied_once", ExpectedAppliedRevision: d.AppliedRevision, ExpectedDesiredRevision: d.DesiredRevision, ExpectedRuntimeIdentity: d.RuntimeIdentity} |
| 203 | admission, current, err := a.beginRuntimeTurnWithModelChoice(tab.ID, false, false, nil, &choice) |
| 204 | if err != nil { |
| 205 | t.Fatal(err) |
| 206 | } |
| 207 | admission.abort() |
| 208 | if current != c { |
| 209 | t.Fatal("choice changed controller") |
| 210 | } |
| 211 | if admission, _, err := a.beginTabTurn(tab.ID, false); err == nil { |
| 212 | admission.abort() |
| 213 | t.Fatal("choice leaked to next submit") |
| 214 | } |
| 215 | if _, err := a.SaveProviderWithKey(view, "rotated-key"); err != nil { |
| 216 | t.Fatal(err) |
| 217 | } |
| 218 | if err := c.ValidateModelApplicationChoice(choice); err == nil { |
| 219 | t.Fatal("stale choice accepted") |
| 220 | } |
| 221 | if modelApplicationDetails(c).CanUseApplied { |
| 222 | t.Fatal("rotated credentials allowed old snapshot") |
| 223 | } |
| 224 | c.CancelJob(job.ID) |
| 225 | c.BackgroundScope().Manager.WaitForSession(t.Context(), ref.SessionID, []string{job.ID}, 10) |
| 226 | a.deferredRebuildTick(false) |
| 227 | if tab.Ctrl == c { |
| 228 | t.Fatal("task completion did not permit deferred configuration application") |
| 229 | } |
| 230 | } |
| 231 |