A Text User Interface for the Coding Harness Agent

 1 ┌ Coding Harness ──────────────────────────────────────────────────────┐
 2 │ ╭ Conversation ─────────────────────────────╮ ╭ Session ───────────╮ │
 3 │ │ > add type hints to search.py             │ │ state     ready    │ │
 4 │ │ [intent: coding → coding tools]           │ │ provider  fireworks│ │
 5 │ │ * read_file {"path": "search.py"}         │ │ model ...          │ │
 6 │ │ --- a/search.py                           │ │ context   ~2100 tok│ │
 7 │ │ +++ b/search.py                           │ │ cost      $0.0012  │ │
 8 │ ╰───────────────────────────────────────────╯ ╰────────────────────╯ │
 9 │ > prompt or /command                                                 │
10 └──────────────────────────────────────────────────────────────────────┘

 1 [project]
 2 name = "coding-harness-text-ui"
 3 version = "0.1.0"
 4 description = "Textual terminal UI for the coding harness agent in ../coding-harness-agent"
 5 requires-python = ">=3.14"
 6 dependencies = [
 7     "textual>=8.2.4",
 8     "coding-harness-agent",
 9 ]
10 
11 [tool.uv]
12 package = false
13 
14 [tool.uv.sources]
15 coding-harness-agent = { path = "../coding-harness-agent", editable = true }

1 cd source-code/coding-harness-text-ui
2 uv sync
3 export FIREWORKS_API_KEY=your-key     # or configure a local MLX provider
4 uv run harness_tui.py                  # or: make run
5 uv run harness_tui.py --cwd ~/myproject --provider mlx

 1 def gate_write(path, full, current, new, heading):
 2     """Shared approval flow: diff, then y/n/s, then write + `make check`."""
 3     diff_text = unified_diff(current, new, "a/" + path, "b/" + path)
 4     print("")
 5     if heading:
 6         print(heading)
 7     print_colored_diff(diff_text)
 8 
 9     if dry_run:
10         return "dry-run: diff shown, file not written (use without --dry-run to apply)"
11     if auto_approve:
12         # Safety: still show diff above, then auto-apply without prompting
13         if not quiet_mode:
14             print("[auto-approve: applying change without prompt]")
15         return apply_change(path, full, new).replace(
16             "applied;", "applied (auto-approved);", 1)
17     answer = prompt_yes_no_skip()
18     if answer == "no":
19         return "user rejected the change"
20     if answer == "skip":
21         reason = prompt_reason()
22         return "user skipped: {}".format(reason)
23     return apply_change(path, full, new)

1 from approval import unified_diff, print_colored_diff, prompt_yes_no_skip, prompt_reason

  1 # agent_bridge.py -- connect the coding-harness-agent library to a text UI.
  2 #
  3 # Copyright (C) 2026 Mark Watson <markw@markwatson.com>
  4 # Licensed under the GNU Affero General Public License v3.0 (AGPL-3.0)
  5 #
  6 # The agent library in ../coding-harness-agent was written for a terminal
  7 # REPL: it reports progress with print() and asks for edit approval with
  8 # input(). This module adapts those two channels without changing the
  9 # library:
 10 #
 11 #   * Output:   the UI captures stdout; LineBuffer turns the fragments that
 12 #               print() writes into whole lines.
 13 #   * Approval: tools.py imported print_colored_diff, prompt_yes_no_skip and
 14 #               prompt_reason by name, so replacing those names on the tools
 15 #               module reroutes every diff and approval prompt to the UI.
 16 #
 17 # Everything here is UI-toolkit agnostic. The caller supplies two callables:
 18 #   show_diff(diff_text)        -- display a unified diff (any thread)
 19 #   ask_approval(diff_text)     -- block until the user answers; returns
 20 #                                  ("yes" | "no" | "skip", reason_str)
 21 
 22 import os
 23 import threading
 24 
 25 import agent
 26 import approval
 27 import fireworks_ai
 28 import harness_config as hc
 29 import mlx_serve
 30 import tools
 31 
 32 
 33 class LineBuffer:
 34     """Accumulate printed text and emit complete lines to `emit`."""
 35 
 36     def __init__(self, emit):
 37         self._emit = emit
 38         self._partial = ""
 39         self._lock = threading.Lock()
 40 
 41     def write(self, text):
 42         with self._lock:
 43             self._partial += text
 44             *lines, self._partial = self._partial.split("\n")
 45         for line in lines:
 46             self._emit(line)
 47 
 48     def flush(self):
 49         with self._lock:
 50             rest, self._partial = self._partial, ""
 51         if rest:
 52             self._emit(rest)
 53 
 54 
 55 def install_approval_hooks(show_diff, ask_approval):
 56     """Route the edit tools' diff display and y/n/s prompts to the UI."""
 57     state = {"diff": "", "reason": ""}
 58 
 59     def print_colored_diff(diff_text):
 60         state["diff"] = diff_text
 61         show_diff(diff_text)
 62 
 63     def prompt_yes_no_skip():
 64         answer, reason = ask_approval(state["diff"])
 65         state["reason"] = reason or ""
 66         return answer
 67 
 68     def prompt_reason():
 69         # The UI collects the reason together with the "skip" answer.
 70         return state["reason"]
 71 
 72     tools.print_colored_diff = print_colored_diff
 73     tools.prompt_yes_no_skip = prompt_yes_no_skip
 74     tools.prompt_reason = prompt_reason
 75     # Diffs are colored by the UI, never with raw ANSI escapes.
 76     approval.set_color_enabled(False)
 77 
 78 
 79 def start_session(cwd=None, provider=None, model=None, auto_approve=False,
 80                   dry_run=False, debug=False):
 81     """Load config and initialize agent state, mirroring agent.cli_main.
 82 
 83     Returns an error string when the session cannot start, else None.
 84     """
 85     if cwd:
 86         if not os.path.isdir(cwd):
 87             return "--cwd directory does not exist: {}".format(cwd)
 88         os.chdir(cwd)
 89     # Load config after the chdir so .local_coding_harness.json is found in
 90     # the directory the agent will work in.
 91     hc.load_harness_config()
 92     agent.apply_harness_flags()
 93     agent.apply_env_overrides()
 94     if not hc.config_provider_names():
 95         return ("no providers configured. Add a \"providers\" section to "
 96                 "~/.coding_harness.json or .local_coding_harness.json")
 97     if provider:
 98         name = provider.strip().lower()
 99         if not hc.config_provider(name):
100             return "unknown provider '{}' (available: {})".format(
101                 provider, ", ".join(hc.config_provider_names()))
102         agent.switch_provider(name)
103     if model:
104         agent.set_current_model(model.strip())
105     tools.auto_approve = auto_approve
106     tools.dry_run = dry_run
107     if debug:
108         fireworks_ai.debug_log = True
109     tools.register_all()
110     agent.reset_conversation()
111     return None
112 
113 
114 def run_line(line):
115     """Run one line of user input: a slash command or a model turn.
116 
117     Blocks (call it from a worker thread). Returns "quit" when the user asked
118     to leave, else None. Output arrives through print().
119     """
120     cmd = agent.handle_slash_command(line)
121     if cmd == "quit":
122         return "quit"
123     if cmd is None:
124         agent.send_to_model(line)
125     return None
126 
127 
128 def status():
129     """-> dict describing the session, for a status panel."""
130     if agent.using_mlx():
131         with mlx_serve._stats_lock:
132             pt = mlx_serve._session_prompt_tokens
133             ct = mlx_serve._session_completion_tokens
134         cost = 0.0
135     else:
136         pt, _cached, ct, _total = fireworks_ai.session_snapshot()
137         cost = fireworks_ai.session_cost()
138     chars = sum(agent.message_char_size(m) for m in agent.messages)
139     if tools.dry_run:
140         mode = "dry-run"
141     elif tools.auto_approve:
142         mode = "auto-approve"
143     else:
144         mode = "ask"
145     return {
146         "cwd": os.getcwd(),
147         "provider": agent.current_provider_name_or_legacy(),
148         "model": agent.current_model_id(),
149         "edits": mode,
150         "search": "{} ({})".format("on" if agent.search_enabled else "off",
151                                    agent.search_engine),
152         "debug": "on" if fireworks_ai.debug_log else "off",
153         "messages": len(agent.messages),
154         "context_tokens": chars // 4,
155         "prompt_tokens": pt,
156         "completion_tokens": ct,
157         "cost": cost,
158     }
159 
160 
161 def completion_words():
162     """Slash commands and skill names, for input suggestions."""
163     return agent.SLASH_COMMANDS + ["/" + s for s in agent.list_skills()]

  1 # harness_tui.py -- Textual terminal UI for the coding harness agent.
  2 #
  3 # Copyright (C) 2026 Mark Watson <markw@markwatson.com>
  4 # Licensed under the GNU Affero General Public License v3.0 (AGPL-3.0)
  5 #
  6 # All agent behavior (providers, the tool-calling loop, tools, slash
  7 # commands, skills, compaction) comes from the coding-harness-agent library
  8 # in ../coding-harness-agent. This file is only the UI:
  9 #
 10 #   +-------------------------------------------+----------------+
 11 #   | Conversation log (agent output, diffs)    | Session status |
 12 #   |                                           |                |
 13 #   +-------------------------------------------+----------------+
 14 #   | > prompt or /command                                       |
 15 #   +------------------------------------------------------------+
 16 #
 17 # A model turn runs in a background thread so the UI stays responsive.
 18 # When the agent proposes an edit, a modal shows the diff and asks
 19 # Yes / No / Skip (with a reason), replacing the library's input() prompt.
 20 
 21 import argparse
 22 import sys
 23 import threading
 24 from concurrent.futures import Future
 25 
 26 from rich.text import Text
 27 
 28 from textual import events
 29 from textual.app import App, ComposeResult
 30 from textual.binding import Binding
 31 from textual.containers import Horizontal, Vertical
 32 from textual.message import Message
 33 from textual.screen import ModalScreen
 34 from textual.suggester import SuggestFromList
 35 from textual.widgets import Button, Footer, Header, Input, Label, RichLog, Static
 36 
 37 import agent_bridge
 38 
 39 
 40 # ---------------------------------------------------------------------------
 41 # Rendering helpers
 42 
 43 def diff_renderable(diff_text):
 44     """A unified diff as Rich Text: cyan headers, green adds, red removals."""
 45     out = Text()
 46     for line in diff_text.rstrip("\n").split("\n"):
 47         if line.startswith(("+++", "---", "@@")):
 48             style = "bold cyan"
 49         elif line.startswith("+"):
 50             style = "green"
 51         elif line.startswith("-"):
 52             style = "red"
 53         else:
 54             style = ""
 55         out.append(line + "\n", style=style)
 56     return out
 57 
 58 
 59 def output_line_style(line):
 60     """Pick a style for one line the agent printed."""
 61     s = line.lstrip()
 62     if s.startswith("* "):
 63         return "cyan"                      # tool call
 64     if s.startswith("[intent:") or s.startswith("[auto-approve"):
 65         return "dim"
 66     if s.startswith("[DEBUG]"):
 67         return "dim yellow"
 68     if s.startswith(("Error", "[Search error", "[Classifier")):
 69         return "bold red"
 70     if s.startswith("(new file:"):
 71         return "bold"
 72     return ""
 73 
 74 
 75 # ---------------------------------------------------------------------------
 76 # Widgets
 77 
 78 class HistoryInput(Input):
 79     """An Input with up/down command history."""
 80 
 81     def __init__(self, **kwargs):
 82         super().__init__(**kwargs)
 83         self.history = []
 84         self.history_index = 0
 85         self.draft = ""
 86 
 87     def remember(self, line):
 88         if line and (not self.history or self.history[-1] != line):
 89             self.history.append(line)
 90         self.history_index = len(self.history)
 91         self.draft = ""
 92 
 93     def on_key(self, event: events.Key) -> None:
 94         if event.key not in ("up", "down"):
 95             return
 96         event.prevent_default()
 97         if event.key == "up":
 98             if self.history_index == len(self.history):
 99                 self.draft = self.value
100             if self.history_index > 0:
101                 self.history_index -= 1
102                 self.value = self.history[self.history_index]
103         elif self.history_index < len(self.history):
104             self.history_index += 1
105             self.value = (self.draft if self.history_index == len(self.history)
106                           else self.history[self.history_index])
107         self.cursor_position = len(self.value)
108 
109 
110 class ConversationLog(RichLog):
111     """The main log. It receives the agent's printed output.
112 
113     Printed text, diffs and end-of-turn markers all arrive as messages
114     posted to this widget, so they are displayed in the order the agent
115     produced them even though the agent runs in another thread.
116     """
117 
118     class ShowDiff(Message):
119         def __init__(self, diff_text):
120             super().__init__()
121             self.diff_text = diff_text
122 
123     class TurnFinished(Message):
124         def __init__(self, quit_requested):
125             super().__init__()
126             self.quit_requested = quit_requested
127 
128     def __init__(self, **kwargs):
129         super().__init__(wrap=True, markup=False, highlight=False, **kwargs)
130         self.buffer = agent_bridge.LineBuffer(self.write_agent_line)
131 
132     def write_agent_line(self, line):
133         self.write(Text(line, style=output_line_style(line)))
134 
135     def on_print(self, event: events.Print) -> None:
136         self.buffer.write(event.text)
137 
138     def on_conversation_log_show_diff(self, message: ShowDiff) -> None:
139         self.buffer.flush()
140         self.write(diff_renderable(message.diff_text))
141 
142     def on_conversation_log_turn_finished(self, message: TurnFinished) -> None:
143         self.buffer.flush()
144 
145 
146 class ApprovalScreen(ModalScreen):
147     """Show a proposed edit and return ("yes" | "no" | "skip", reason)."""
148 
149     BINDINGS = [
150         Binding("y", "answer('yes')", "Apply"),
151         Binding("n", "answer('no')", "Reject"),
152         Binding("s", "answer('skip')", "Skip + reason"),
153         Binding("escape", "answer('no')", "Reject", show=False),
154     ]
155 
156     def __init__(self, diff_text):
157         super().__init__()
158         self.diff_text = diff_text
159 
160     def compose(self) -> ComposeResult:
161         with Vertical(id="approval-dialog"):
162             yield Label("The agent proposes this change. Apply it?",
163                         id="approval-title")
164             yield RichLog(id="approval-diff", wrap=False, markup=False)
165             yield Input(placeholder="Why are you skipping it? (Enter to send)",
166                         id="approval-reason")
167             with Horizontal(id="approval-buttons"):
168                 yield Button("Yes, apply (y)", variant="success", id="yes")
169                 yield Button("No (n)", variant="error", id="no")
170                 yield Button("Skip + reason (s)", variant="warning", id="skip")
171 
172     def on_mount(self) -> None:
173         self.query_one("#approval-diff", RichLog).write(
174             diff_renderable(self.diff_text))
175         self.query_one("#approval-reason", Input).display = False
176         self.query_one("#yes", Button).focus()
177 
178     def on_button_pressed(self, event: Button.Pressed) -> None:
179         self.action_answer(event.button.id)
180 
181     def action_answer(self, answer):
182         if answer != "skip":
183             self.dismiss((answer, ""))
184             return
185         reason = self.query_one("#approval-reason", Input)
186         reason.display = True
187         reason.focus()
188 
189     def on_input_submitted(self, event: Input.Submitted) -> None:
190         self.dismiss(("skip", event.value.strip()))
191 
192 
193 # ---------------------------------------------------------------------------
194 # The application
195 
196 class HarnessApp(App):
197     TITLE = "Coding Harness"
198 
199     CSS = """
200     #main { height: 1fr; }
201     #conversation {
202         width: 1fr;
203         border: round $primary;
204         padding: 0 1;
205     }
206     #status {
207         width: 38;
208         border: round $secondary;
209         padding: 0 1;
210     }
211     #prompt { margin: 0 1; }
212     ApprovalScreen { align: center middle; }
213     #approval-dialog {
214         width: 90%;
215         height: 85%;
216         border: thick $warning;
217         background: $surface;
218         padding: 1 2;
219     }
220     #approval-title { text-style: bold; margin-bottom: 1; }
221     #approval-diff { height: 1fr; border: round $primary; }
222     #approval-reason { margin-top: 1; }
223     #approval-buttons { height: auto; align-horizontal: right; margin-top: 1; }
224     #approval-buttons Button { margin-left: 2; }
225     """
226 
227     BINDINGS = [
228         Binding("ctrl+q", "quit", "Quit"),
229         Binding("ctrl+l", "clear_log", "Clear log"),
230         Binding("escape", "focus_prompt", "Focus prompt", show=False),
231     ]
232 
233     def __init__(self):
234         super().__init__()
235         self.busy = False
236         self.closing = threading.Event()
237 
238     def compose(self) -> ComposeResult:
239         yield Header()
240         with Horizontal(id="main"):
241             log = ConversationLog(id="conversation")
242             log.border_title = "Conversation"
243             yield log
244             status = Static(id="status")
245             status.border_title = "Session"
246             yield status
247         yield HistoryInput(
248             id="prompt",
249             placeholder="Ask the agent, or type /help for commands",
250             suggester=SuggestFromList(agent_bridge.completion_words(),
251                                       case_sensitive=False))
252         yield Footer()
253 
254     @property
255     def conversation(self) -> ConversationLog:
256         return self.query_one("#conversation", ConversationLog)
257 
258     def on_mount(self) -> None:
259         agent_bridge.install_approval_hooks(self.show_diff, self.ask_approval)
260         self.begin_capture_print(self.conversation)
261         info = agent_bridge.status()
262         self.sub_title = info["cwd"]
263         log = self.conversation
264         log.write(Text("Coding Agent.  /help for agent commands, /clear to "
265                        "clear this log, Ctrl+Q to quit.", style="bold"))
266         log.write(Text("Up/Down recalls history; Right arrow accepts a "
267                        "suggested /command.", style="dim"))
268         self.refresh_status()
269         self.set_interval(1.0, self.refresh_status)
270         self.query_one("#prompt").focus()
271 
272     def on_unmount(self) -> None:
273         self.closing.set()
274 
275     # --- status panel ---
276 
277     def refresh_status(self) -> None:
278         info = agent_bridge.status()
279         t = Text()
280 
281         def row(label, value, style=""):
282             t.append("{:<10}".format(label), style="bold")
283             t.append(str(value) + "\n", style=style)
284 
285         row("state", "working…" if self.busy else "ready",
286             "bold yellow" if self.busy else "bold green")
287         row("provider", info["provider"])
288         t.append("model\n", style="bold")
289         t.append("  " + info["model"] + "\n", style="cyan")
290         row("edits", info["edits"],
291             "" if info["edits"] == "ask" else "bold magenta")
292         row("search", info["search"])
293         row("debug", info["debug"])
294         t.append("\n")
295         row("messages", info["messages"])
296         row("context", "~{} tok".format(info["context_tokens"]))
297         row("prompt", "{} tok".format(info["prompt_tokens"]))
298         row("output", "{} tok".format(info["completion_tokens"]))
299         cost = info["cost"]
300         row("cost", "n/a" if cost is None else "${:.4f}".format(cost))
301         t.append("\ncwd\n", style="bold")
302         t.append(info["cwd"] + "\n", style="dim")
303         self.query_one("#status", Static).update(t)
304 
305     # --- input ---
306 
307     def on_input_submitted(self, event: Input.Submitted) -> None:
308         if not isinstance(event.input, HistoryInput):
309             return
310         line = event.value.strip()
311         prompt = event.input
312         prompt.remember(line)
313         prompt.value = ""
314         if not line:
315             return
316         if line in ("/quit", "/exit"):
317             self.exit()
318             return
319         if line == "/clear":
320             self.action_clear_log()
321             return
322         if self.busy:
323             self.notify("The agent is still working on the previous request.",
324                         severity="warning")
325             return
326         self.conversation.write(Text("\n> " + line, style="bold"))
327         self.set_busy(True)
328         threading.Thread(target=self.run_turn, args=(line,), daemon=True).start()
329 
330     def set_busy(self, busy):
331         self.busy = busy
332         prompt = self.query_one("#prompt", HistoryInput)
333         prompt.placeholder = ("Working… (wait for the reply)" if busy
334                               else "Ask the agent, or type /help for commands")
335         self.refresh_status()
336 
337     def run_turn(self, line):
338         """Worker thread: run one line through the agent library.
339 
340         A daemon thread rather than a Textual worker, so quitting does not
341         have to wait for an in-flight HTTP request to finish."""
342         quit_requested = False
343         try:
344             quit_requested = agent_bridge.run_line(line) == "quit"
345         except Exception as e:  # noqa: BLE001
346             print("\nError talking to model: {}".format(e))
347         if self.closing.is_set():
348             return
349         self.conversation.post_message(ConversationLog.TurnFinished(quit_requested))
350         try:
351             self.call_from_thread(self.finish_turn, quit_requested)
352         except RuntimeError:
353             pass  # app already shut down
354 
355     def finish_turn(self, quit_requested):
356         self.set_busy(False)
357         if quit_requested:
358             self.exit()
359 
360     # --- approval bridge (called from the worker thread) ---
361 
362     def show_diff(self, diff_text):
363         self.conversation.post_message(ConversationLog.ShowDiff(diff_text))
364 
365     def ask_approval(self, diff_text):
366         """Block the worker thread until the user answers the modal."""
367         answer = Future()
368         try:
369             self.call_from_thread(self.push_screen, ApprovalScreen(diff_text),
370                                   answer.set_result)
371         except RuntimeError:
372             return ("no", "")
373         while not self.closing.is_set():
374             try:
375                 result = answer.result(timeout=0.25)
376             except TimeoutError:
377                 continue
378             verdict, reason = result
379             label = {"yes": "approved", "no": "rejected",
380                      "skip": "skipped: " + reason}[verdict]
381             print("[change {}]".format(label))
382             return result
383         return ("no", "")
384 
385     # --- actions ---
386 
387     def action_clear_log(self) -> None:
388         self.conversation.clear()
389 
390     def action_focus_prompt(self) -> None:
391         self.query_one("#prompt").focus()
392 
393 
394 # ---------------------------------------------------------------------------
395 # CLI entry
396 
397 def build_arg_parser():
398     parser = argparse.ArgumentParser(
399         prog="harness-tui",
400         description="Text UI for the coding harness agent.")
401     parser.add_argument("--provider", metavar="NAME",
402                         help="provider profile from the harness config")
403     parser.add_argument("--model", metavar="ID",
404                         help="override the provider's model")
405     parser.add_argument("--cwd", metavar="DIR",
406                         help="directory the agent works in")
407     parser.add_argument("-y", "--yes", action="store_true",
408                         help="auto-approve edits (diffs are still shown)")
409     parser.add_argument("--dry-run", action="store_true",
410                         help="show diffs but never write files")
411     parser.add_argument("--debug", action="store_true",
412                         help="log raw requests and responses")
413     return parser
414 
415 
416 def main(argv=None):
417     args = build_arg_parser().parse_args(argv)
418     error = agent_bridge.start_session(
419         cwd=args.cwd, provider=args.provider, model=args.model,
420         auto_approve=args.yes, dry_run=args.dry_run, debug=args.debug)
421     if error:
422         sys.stderr.write("error: {}\n".format(error))
423         sys.exit(5)
424     HarnessApp().run()
425 
426 
427 if __name__ == "__main__":
428     main()