| 💾 Saved. d722711 k33g 7h ago | 1 | // Package agent contains the interactive conversation loop (REPL). |
| 2 | package agent |
| 3 | |
| 4 | import ( |
| 5 | "bufio" |
| 6 | "context" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "io" |
| 10 | "os" |
| 11 | "os/signal" |
| 12 | "strings" |
| 13 | |
| 14 | "mm/internal/compact" |
| 15 | "mm/internal/config" |
| 16 | "mm/internal/detector" |
| 17 | "mm/internal/engine" |
| 18 | "mm/internal/mention" |
| 19 | "mm/internal/session" |
| 20 | "mm/internal/spinner" |
| 21 | |
| 22 | "github.com/firebase/genkit/go/ai" |
| 23 | ) |
| 24 | |
| 25 | type genResult struct { |
| 26 | resp *ai.ModelResponse |
| 27 | // history is the FULL conversation returned by engine.Generate — tool turns |
| 28 | // included. Genkit only hands back the last message; without this field the |
| 29 | // agent forgets the commands it just ran and starts telling stories instead |
| 30 | // of re-reading. |
| 31 | history []*ai.Message |
| 32 | err error |
| 33 | } |
| 34 | |
| 35 | // Run starts the loop: read user input → generate (with automatic tool calls) → |
| 36 | // print → start over, keeping the history. |
| 37 | func Run(ctx context.Context, e *engine.Engine, system string, tools []ai.ToolRef, d *detector.LoopDetector) { |
| 38 | messages := session.Fresh(system) |
| 39 | ctx, cancelAll := context.WithCancel(ctx) |
| 40 | defer cancelAll() |
| 41 | |
| 42 | // @path mentions are relative to where mm was started — the same anchor |
| 43 | // as a relative skillsDir and as the bash tool's own working directory. |
| 44 | cwd, _ := os.Getwd() |
| 45 | |
| 46 | // sigCh captures Ctrl+C (SIGINT). |
| 47 | sigCh := make(chan os.Signal, 1) |
| 48 | signal.Notify(sigCh, os.Interrupt) |
| 49 | |
| 50 | // inputCh receives commands from the reader goroutine. |
| 51 | inputCh := make(chan string) |
| 52 | |
| 53 | // startInputReader launches a goroutine that reads from Stdin. |
| 54 | // It is designed to restart if an interrupt (Ctrl+C) occurs. |
| 55 | startInputReader := func() { |
| 56 | go func() { |
| 57 | for { |
| 58 | reader := bufio.NewReader(os.Stdin) |
| 59 | line, err := reader.ReadString('\n') |
| 60 | if err != nil { |
| 61 | select { |
| 62 | case <-ctx.Done(): |
| 63 | return |
| 64 | default: |
| 65 | if err == io.EOF { |
| 66 | close(inputCh) |
| 67 | return |
| 68 | } |
| 69 | // If it's an interrupt, just continue to restart the reader. |
| 70 | continue |
| 71 | } |
| 72 | } |
| 73 | select { |
| 74 | case <-ctx.Done(): |
| 75 | return |
| 76 | case inputCh <- strings.TrimSpace(line): |
| 77 | } |
| 78 | } |
| 79 | }() |
| 80 | } |
| 81 | |
| 82 | startInputReader() |
| 83 | |
| 84 | fmt.Println(`Agent ready. Commands: "/quit" to exit, "/abort" to stop current generation, "/compact" to compress the history, "/new" to start a new session.`) |
| 85 | fmt.Println(`Shortcut: Ctrl+C to abort generation, or Ctrl+C at prompt to quit.`) |
| 86 | |
| 87 | generating := false |
| 88 | |
| 89 | for { |
| 90 | fmt.Print("\n> ") |
| 91 | |
| 92 | // 1. Wait for the next user input OR a quit signal. |
| 93 | var input string |
| 94 | var ok bool |
| 95 | |
| 96 | select { |
| 97 | case <-sigCh: |
| 98 | if !generating { |
| 99 | fmt.Println("\nGoodbye.") |
| 100 | return |
| 101 | } |
| 102 | // If generating, we let the generation loop handle the SIGINT. |
| 103 | // We fall through to wait for the next input (which will be the prompt). |
| 104 | case input, ok = <-inputCh: |
| 105 | if !ok { |
| 106 | return |
| 107 | } |
| 108 | } |
| 109 | |
| 110 | // 2. Basic Command Handling. |
| 111 | if input == "/quit" { |
| 112 | break |
| 113 | } |
| 114 | if input == "" { |
| 115 | continue |
| 116 | } |
| 117 | if input == "/abort" { |
| 118 | spinner.Stop() |
| 119 | fmt.Println("⚠️ No generation in progress.") |
| 120 | continue |
| 121 | } |
| 122 | // Manual compression. Only reachable here, between two generations: |
| 123 | // Genkit holds its own copy of the conversation while it runs, and a |
| 124 | // tool_call/tool_result pair in flight must not move under it. |
| 125 | if input == "/compact" { |
| 126 | messages = compactHistory(ctx, e, messages, true) |
| 127 | continue |
| 128 | } |
| 129 | // New session. Same place and same reason as /compact: the history |
| 130 | // only moves between two generations. Typed DURING a generation, the |
| 131 | // command goes through the "new command received" branch below, which |
| 132 | // cancels the turn and re-injects it here — so the reset always applies |
| 133 | // to the finished history, never to one Genkit still holds. |
| 134 | if session.IsNewCommand(input) { |
| 135 | messages = startNewSession(e, d, system, messages) |
| 136 | continue |
| 137 | } |
| 138 | |
| 139 | // 3. Normal message: Add to history and start generation. |
| 140 | // |
| 141 | // The automatic compression runs BEFORE the question is appended and |
| 142 | // BEFORE `before` is computed: that index is what the rollback |
| 143 | // (`messages[:before-1]`) and the ⚙ line (`res.history[before:]`) |
| 144 | // rely on, and a history that shrinks after it is taken would break |
| 145 | // both. The server's own count of the last context is preferred to |
| 146 | // the estimate when it is larger — it sees the tools' JSON and the |
| 147 | // chat template, which the estimate cannot. The window is the one the |
| 148 | // Engine knows: the yaml's contextWindow, or what the probe learned. |
| 149 | // |
| 150 | // That probe ran once, at start-up. A llama-server started AFTER bob |
| 151 | // left `ctx: unknown` for the whole session — observed — and the |
| 152 | // compression could then only trigger on maxMessages. So, while the |
| 153 | // window is unknown and compression is on, ask the server again |
| 154 | // before deciding; once known, never again. A probe that still |
| 155 | // fails prints nothing: the start-up warning already said so. |
| 156 | if c := config.Cfg.Context; c.Enabled && e.ContextWindow == 0 { |
| 157 | if w, src := e.EnsureContextWindow(ctx); w > 0 { |
| 158 | dimln(fmt.Sprintf("ctx: %d (%s)", w, src)) |
| 159 | } |
| 160 | } |
| 161 | if dec := compact.Decide(messages, e.LastInputTokens(), e.ContextWindow, config.Cfg.Context); dec.Compact { |
| 162 | messages = compactHistory(ctx, e, messages, false) |
| 163 | } |
| 164 | // "@path" in the question names a file for the model to look at. The |
| 165 | // text stays as typed; each existing path is appended as an |
| 166 | // [attached file: …] line — what an editor's @-mention becomes over |
| 167 | // ACP — and echoed here so the user sees the notation was understood. |
| 168 | input, attached := mention.Expand(input, cwd) |
| 169 | for _, a := range attached { |
| 170 | dimln("📎 " + a.Path) |
| 171 | } |
| 172 | messages = append(messages, ai.NewUserTextMessage(input)) |
| 173 | before := len(messages) |
| 174 | fmt.Println() |
| 175 | |
| 176 | // Create a cancellable context for this specific generation cycle. |
| 177 | genCtx, cancelGen := context.WithCancel(ctx) |
| 178 | generating = true |
| 179 | |
| 180 | // Launch generation in a goroutine. |
| 181 | resultCh := make(chan genResult, 1) |
| 182 | go func() { |
| 183 | resp, history, err := e.Generate(genCtx, messages, tools) |
| 184 | resultCh <- genResult{resp: resp, history: history, err: err} |
| 185 | }() |
| 186 | |
| 187 | // 4. The "Waiting Room": Listen for completion, interruption, OR new command. |
| 188 | generationRunning := true |
| 189 | for generationRunning { |
| 190 | select { |
| 191 | case res := <-resultCh: |
| 192 | generationRunning = false |
| 193 | generating = false |
| 194 | |
| 195 | // We take the FULL history returned by engine.Generate, not just |
| 196 | // resp.Message — and we take it EVEN when the generation failed |
| 197 | // or was interrupted: on an abort (Ctrl+C, /abort), a maxTurns |
| 198 | // overrun or a silence caught by the watchdog, all the work of |
| 199 | // the turn is in `history`, and Genkit itself returns nothing. |
| 200 | // (resp.History() will not do: empty on the streamed path.) |
| 201 | switch { |
| 202 | case len(res.history) > before: |
| 203 | messages = res.history |
| 204 | case res.resp != nil && res.resp.Message != nil: |
| 205 | messages = append(messages, res.resp.Message) |
| 206 | case res.err != nil: |
| 207 | messages = messages[:before-1] // question sans réponse : on l'oublie |
| 208 | } |
| 209 | |
| 210 | if res.err != nil { |
| 211 | if genCtx.Err() == context.Canceled { |
| 212 | fmt.Println("\n🛑 Generation aborted.") |
| 213 | } else { |
| 214 | // One line, in the provider's words: see engine.Explain. |
| 215 | fmt.Println("\n[error: " + e.Explain(res.err) + "]") |
| 216 | } |
| 217 | } else { |
| 218 | // --- Global Loop Detection --- |
| 219 | // We check the tool calls in the model response. |
| 220 | if res.resp != nil && res.resp.Message != nil { |
| 221 | for _, part := range res.resp.Message.Content { |
| 222 | if part.IsToolRequest() && part.ToolRequest != nil { |
| 223 | d.Record(detector.Action{ |
| 224 | ToolName: part.ToolRequest.Name, |
| 225 | Input: fmt.Sprintf("%v", part.ToolRequest.Input), |
| 226 | Output: "[tool_call]", |
| 227 | }) |
| 228 | } |
| 229 | } |
| 230 | } |
| 231 | fmt.Println() |
| 232 | } |
| 233 | |
| 234 | // The command count is the only visible proof that the model |
| 235 | // worked rather than improvised. Zero is normal on a |
| 236 | // conversational question; zero on "read this file" is not. |
| 237 | turn := res.history[min(before, len(res.history)):] |
| 238 | cmds := engine.CommandList(turn) |
| 239 | ops := engine.FileOps(turn) |
| 240 | line := fmt.Sprintf("⚙ %d command(s)", len(cmds)) |
| 241 | // The file tools have their own column: the A/B of this part |
| 242 | // is "edits through bash" versus "edits through tools", and |
| 243 | // a single number would add up what it is meant to separate. |
| 244 | if len(ops) > 0 { |
| 245 | line += fmt.Sprintf(" · 📝 %d file op(s)", len(ops)) |
| 246 | } |
| 247 | if k := engine.Skills(turn); k > 0 { |
| 248 | line += fmt.Sprintf(" · 📖 %d skill(s)", k) |
| 249 | } |
| 250 | if spinner.Styled() { |
| 251 | fmt.Printf("\033[2m%s\033[0m\n", line) |
| 252 | } else { |
| 253 | fmt.Println(line) |
| 254 | } |
| 255 | if config.Cfg.DisplayCommands { |
| 256 | printCommands(cmds) |
| 257 | printCommands(ops) |
| 258 | } |
| 259 | |
| 260 | case <-sigCh: |
| 261 | // Handle Ctrl+C during generation. |
| 262 | // Stop() first: the spinner is running during the generation, |
| 263 | // and without it this message lands on its line — the next |
| 264 | // frame would erase it. Same rule as in dmr.emit. |
| 265 | spinner.Stop() |
| 266 | fmt.Println("\n🛑 Interrupted (Ctrl+C).") |
| 267 | cancelGen() |
| 268 | // We don't set generationRunning=false; we wait for resultCh to catch the Canceled error. |
| 269 | |
| 270 | case nextInput := <-inputCh: |
| 271 | spinner.Stop() // le spinner tourne : on lui reprend la ligne |
| 272 | switch nextInput { |
| 273 | case "/abort": |
| 274 | cancelGen() |
| 275 | fmt.Println("🛑 Aborting...") |
| 276 | case "/quit": |
| 277 | cancelGen() |
| 278 | fmt.Println("Goodbye.") |
| 279 | return |
| 280 | default: |
| 281 | // A new message arrived! Cancel current task and re-inject the message. |
| 282 | fmt.Println("🔄 New command received. Cancelling current task...") |
| 283 | cancelGen() |
| 284 | |
| 285 | // Wait for current goroutine to clean up before re-injecting. |
| 286 | res := <-resultCh |
| 287 | |
| 288 | // Same principle as above: the commands already run stay in |
| 289 | // the history, otherwise the next question starts again |
| 290 | // without knowing what was done. |
| 291 | switch { |
| 292 | case len(res.history) > before: |
| 293 | messages = res.history |
| 294 | case res.err != nil: |
| 295 | messages = messages[:before-1] |
| 296 | if genCtx.Err() != context.Canceled { |
| 297 | fmt.Println("[error: " + e.Explain(res.err) + "]") |
| 298 | } |
| 299 | } |
| 300 | generationRunning = false |
| 301 | generating = false |
| 302 | |
| 303 | // Re-inject the command into the main loop. |
| 304 | go func(cmd string) { inputCh <- cmd }(nextInput) |
| 305 | } |
| 306 | } |
| 307 | } |
| 308 | // Final cleanup for this cycle. |
| 309 | cancelGen() |
| 310 | } |
| 311 | } |
| 312 | |
| 313 | // printCommands recaps the commands of the turn, in green. |
| 314 | // |
| 315 | // The count says HOW MANY, the list says WHAT: it reads at a glance once the |
| 316 | // answer is written, whereas the 🛠️ lines are scattered through the output of |
| 317 | // the commands. Green separates it from both the grey of the tools and the |
| 318 | // model's own text — and it only comes out on a terminal, like every other |
| 319 | // colour here. |
| 320 | func printCommands(cmds []string) { |
| 321 | if len(cmds) == 0 { |
| 322 | return |
| 323 | } |
| 324 | green, off := "", "" |
| 325 | if spinner.Styled() { |
| 326 | green, off = "\033[32m", "\033[0m" |
| 327 | } |
| 328 | |
| 329 | var b strings.Builder |
| 330 | for i, c := range cmds { |
| 331 | b.WriteString(green + fmt.Sprintf(" %d. %s", i+1, oneLine(c)) + off + "\n") |
| 332 | } |
| 333 | fmt.Print(b.String()) |
| 334 | } |
| 335 | |
| 336 | // maxCommandWidth is the width beyond which a command is cut. A hundred |
| 337 | // columns: that fits a demo terminal without overflowing it. |
| 338 | const maxCommandWidth = 100 |
| 339 | |
| 340 | // oneLine folds a command onto ONE line and truncates it when needed. |
| 341 | // |
| 342 | // A command can be long and span several lines — a heredoc, a trailing &&. The |
| 343 | // recap is there to scan what the agent did, not to re-read its code: spread |
| 344 | // over several lines it would grow longer than the output it summarises, and |
| 345 | // the numbering would stop being readable. |
| 346 | func oneLine(s string) string { |
| 347 | // Fields splits on ALL whitespace, line breaks included: that is exactly |
| 348 | // the folding we want, heredoc indentation and all. |
| 349 | s = strings.Join(strings.Fields(s), " ") |
| 350 | r := []rune(s) |
| 351 | if len(r) <= maxCommandWidth { |
| 352 | return s |
| 353 | } |
| 354 | return string(r[:maxCommandWidth-1]) + "…" |
| 355 | } |
| 356 | |
| 357 | // compactHistory replaces the old turns of `msgs` by a model-written summary |
| 358 | // and says so in one line. On ANY failure — server down, watchdog, empty or |
| 359 | // malformed result — it returns `msgs` unchanged: the compression exists to |
| 360 | // keep the next question possible, so it must never cost one. |
| 361 | // |
| 362 | // `forced` is the /compact command: it skips the threshold, not the "is there |
| 363 | // anything older than the kept turns" check, and it says so when there is not. |
| 364 | // The loop detector is deliberately not touched: it records what the agent |
| 365 | // DOES, and forgetting a conversation does not make a repeated command new. |
| 366 | func compactHistory(ctx context.Context, e *engine.Engine, msgs []*ai.Message, forced bool) []*ai.Message { |
| 367 | cfg := config.Cfg.Context |
| 368 | |
| 369 | // The summary is the longest prefill of the session and prints nothing |
| 370 | // while it runs: without a label the agent looks hung, exactly the case |
| 371 | // the spinner exists for. |
| 372 | spinner.Start("Compressing") |
| 373 | res, err := compact.Compact(ctx, msgs, cfg, func(ctx context.Context, request []*ai.Message) (string, error) { |
| 374 | return e.Summarize(ctx, request, cfg.SummaryMaxTokens) |
| 375 | }) |
| 376 | spinner.Stop() |
| 377 | |
| 378 | switch { |
| 379 | case errors.Is(err, compact.ErrNothingToCompact): |
| 380 | if forced { |
| 381 | dimln(fmt.Sprintf("🗜️ nothing to compact: %d message(s), no turn older than the last %d", len(msgs), cfg.KeepLastTurns)) |
| 382 | } |
| 383 | return msgs |
| 384 | case err != nil: |
| 385 | // One line, in the provider's words, like every other error here. |
| 386 | fmt.Printf("[compact: failed, history kept: %s]\n", e.Explain(err)) |
| 387 | return msgs |
| 388 | } |
| 389 | |
| 390 | // The server's last measure described the history we just replaced. |
| 391 | e.ForgetInputTokens() |
| 392 | |
| 393 | if cfg.ShowStats { |
| 394 | dimln("🗜️ " + res.Report()) |
| 395 | } |
| 396 | return res.Messages |
| 397 | } |
| 398 | |
| 399 | // startNewSession forgets the conversation and returns the history of a fresh |
| 400 | // one: the system prompt alone. Three things make up "the session" here, and |
| 401 | // all three are reset — the messages, the server's token count of the last |
| 402 | // context (it described a history that no longer exists, same rule as after a |
| 403 | // compression), and the loop detector (a new session is a new task: a command |
| 404 | // repeated in the old one must not be flagged as a loop in this one — the |
| 405 | // opposite of compactHistory's choice, where the task continues). |
| 406 | // |
| 407 | // It says in one line how much was forgotten, so a /new on an empty session |
| 408 | // visibly did nothing rather than silently nothing. |
| 409 | func startNewSession(e *engine.Engine, d *detector.LoopDetector, system string, old []*ai.Message) []*ai.Message { |
| 410 | forgotten := session.Forgotten(old) |
| 411 | e.ForgetInputTokens() |
| 412 | d.Reset() |
| 413 | dimln(fmt.Sprintf("🆕 new session: %d message(s) forgotten", forgotten)) |
| 414 | return session.Fresh(system) |
| 415 | } |
| 416 | |
| 417 | // dimln prints one line, dimmed on a terminal and plain in a pipe — the same |
| 418 | // rule as the ⚙ line: `bob | jq` must not receive escape codes. |
| 419 | func dimln(line string) { |
| 420 | if spinner.Styled() { |
| 421 | fmt.Printf("\033[2m%s\033[0m\n", line) |
| 422 | } else { |
| 423 | fmt.Println(line) |
| 424 | } |
| 425 | } |