Building a Coding Harness Agent

 1 user input --> agent.py --> intent classifier --> chat_loop.py (agentic loop)
 2                                                     |
 3                                 fireworks_ai.py or mlx_serve.py  (HTTP transport)
 4                                                     |
 5                                        provider endpoint (hosted or local)
 6                                                     |
 7                                        tool_calls? --> tools.py executes
 8                                                     |
 9                                        approval.py gates writes
10                                                     |
11                                        results appended as tool messages, loop repeats

 1 {
 2   "role": "assistant",
 3   "content": "Let me read the file first.",
 4   "tool_calls": [
 5     {
 6       "id": "call_abc123",
 7       "type": "function",
 8       "function": {
 9         "name": "read_file",
10         "arguments": "{\"path\": \"search.py\"}"
11       }
12     }
13   ]
14 }

1 {
2   "role": "tool",
3   "tool_call_id": "call_abc123",
4   "name": "read_file",
5   "content": "def normalize(s):\n    return s.strip().lower()\n"
6 }

 1 {
 2   "default_provider": "fireworks",
 3   "providers": {
 4     "fireworks": {
 5       "type": "openai",
 6       "endpoint": "https://api.fireworks.ai/inference/v1/chat/completions",
 7       "api_key_env": "FIREWORKS_API_KEY",
 8       "model": "accounts/fireworks/models/deepseek-v4p1-flash",
 9       "generation": { "temperature": 0.6, "max_tokens": 32768 },
10       "pricing": { "input": 0.14, "cached_input": 0.028, "output": 0.28 }
11     },
12     "mlx": {
13       "type": "mlx",
14       "endpoint": "http://localhost:11434/v1/chat/completions",
15       "model": "mlx-community/gemma-4-26B-A4B-it-OptiQ-4bit",
16       "generation": { "temperature": 0.6, "max_tokens": 32768 }
17     }
18   },
19   "search": { "engine": "brave", "enabled": false },
20   "debug": false,
21   "quiet": false,
22   "plain": false
23 }

1 {
2   "default_provider": "mlx",
3   "search": { "enabled": true },
4   "providers": {
5     "fireworks": {
6       "model": "accounts/fireworks/models/kimi-k2-instruct"
7     }
8   }
9 }

  1 # coding_harness -- hierarchical JSON configuration for the coding harness,
  2 # in the rough style of the Pi coding harness config.  Python port of
  3 # harness-config.rkt.
  4 #
  5 # Copyright (C) 2026 Mark Watson <markw@markwatson.com>
  6 # Licensed under the GNU Affero General Public License v3.0 (AGPL-3.0)
  7 # See LICENSE file for details
  8 #
  9 # Two config layers are merged (local wins over global on conflicts):
 10 #
 11 #   Global:  ~/.coding_harness.json            (base configuration)
 12 #   Local:   .local_coding_harness.json        (optional per-project override,
 13 #                                              loaded from the current
 14 #                                              directory at startup)
 15 #
 16 # Rough format (all sections optional):
 17 #
 18 # {
 19 #   "default_provider": "mlx",
 20 #   "providers": {
 21 #     "mlx":       { "type": "mlx", "endpoint": "http://localhost:11434/v1/chat/completions",
 22 #                    "model": "mlx-community/gemma-4-26B-A4B-it-OptiQ-4bit",
 23 #                    "generation": { "temperature": 0.6, "max_tokens": 32768 } },
 24 #     "fireworks": { "type": "openai", "endpoint": "https://api.fireworks.ai/inference/v1/chat/completions",
 25 #                    "api_key_env": "FIREWORKS_API_KEY",
 26 #                    "model": "accounts/fireworks/models/deepseek-v4p1-flash",
 27 #                    "generation": { "temperature": 0.6, "max_tokens": 32768 },
 28 #                    "pricing": { "input": 0.14, "cached_input": 0.028, "output": 0.28 } }
 29 #   },
 30 #   "search": { "engine": "brave", "enabled": false },
 31 #   "debug": false, "quiet": false, "plain": false
 32 # }
 33 #
 34 # Provider "type" is either "mlx" (the local mlx-serve backend -- OpenAI-style
 35 # /v1/chat/completions served by mlx_lm.server on localhost:11434, oMLX on
 36 # port 8000, sushi on port 12345) or "openai" (OpenAI-style chat completions;
 37 # Fireworks.ai and any compatible endpoint). The type strings "ollama",
 38 # "omlx", and "sushi" are also accepted and mapped to "mlx".
 39 #
 40 # Every provider-specific value lives here: endpoint, model, api_key_env,
 41 # generation parameters, and the per-1M-token USD "pricing" rates (input,
 42 # cached_input, output) that /tokens uses. Nothing provider-specific is
 43 # compiled into the Python code, so adding or changing a provider needs no
 44 # code change. A profile with no "pricing" block reports token counts without
 45 # a cost estimate rather than guessing a rate.
 46 #
 47 # Merge rules: nested dicts merge recursively, local keys override global
 48 # keys; anything that is not a dict (strings, numbers, booleans, lists) is
 49 # replaced wholesale by the local value when present.
 50 
 51 import json
 52 import os
 53 
 54 GLOBAL_CONFIG_PATH = os.path.expanduser("~/.coding_harness.json")
 55 LOCAL_CONFIG_BASENAME = ".local_coding_harness.json"
 56 
 57 # The merged config, loaded once at startup (reloadable via load_harness_config).
 58 harness_config = {}
 59 
 60 # Mutable cell: the name of the provider section currently in use.
 61 _active_provider_name = None
 62 
 63 
 64 # ---------------------------------------------------------------------------
 65 # JSON loading helpers
 66 
 67 def _read_json_file(path):
 68     """-> config dict, or None if the file is missing / unreadable / not an object."""
 69     try:
 70         if not os.path.isfile(path):
 71             return None
 72         with open(path, "r", encoding="utf-8") as f:
 73             v = json.load(f)
 74         return v if isinstance(v, dict) else None
 75     except Exception:
 76         return None
 77 
 78 
 79 # ---------------------------------------------------------------------------
 80 # Recursive dict merge (local overrides global)
 81 
 82 def deep_merge(global_cfg, local_cfg):
 83     if not isinstance(global_cfg, dict):
 84         return local_cfg
 85     if not isinstance(local_cfg, dict):
 86         return local_cfg
 87     acc = dict(global_cfg)
 88     for k, v in local_cfg.items():
 89         if isinstance(acc.get(k), dict) and isinstance(v, dict):
 90             acc[k] = deep_merge(acc[k], v)
 91         else:
 92             acc[k] = v
 93     return acc
 94 
 95 
 96 def local_config_path():
 97     """The local override file, always looked up in the *current* directory
 98     (--cwd may have changed it since this module was imported)."""
 99     return os.path.join(os.getcwd(), LOCAL_CONFIG_BASENAME)
100 
101 
102 def load_harness_config():
103     """Load global then local, deep-merge, store, and return the result."""
104     global harness_config, _active_provider_name
105     global_cfg = _read_json_file(GLOBAL_CONFIG_PATH) or {}
106     local_cfg = _read_json_file(local_config_path()) or {}
107     harness_config = deep_merge(global_cfg, local_cfg)
108     _active_provider_name = None  # re-resolve the default profile
109     return harness_config
110 
111 
112 # ---------------------------------------------------------------------------
113 # Providers
114 
115 def config_providers():
116     p = harness_config.get("providers", {})
117     return p if isinstance(p, dict) else {}
118 
119 
120 def provider_names(cfg):
121     """-> sorted profile names declared in `cfg` (the caller's dict, not the
122     module global, so loading and defaulting can be reasoned about separately)."""
123     p = cfg.get("providers") if isinstance(cfg, dict) else None
124     return sorted(p.keys()) if isinstance(p, dict) else []
125 
126 
127 def config_provider_names():
128     return provider_names(harness_config)
129 
130 
131 def config_provider(name):
132     """-> provider dict for profile `name`, or None."""
133     if name is None:
134         return None
135     return config_providers().get(str(name))
136 
137 
138 def _pick_default_provider_name(cfg):
139     """-> name of the profile to start with, or None.
140 
141     The declared "default_provider" wins; otherwise "fireworks" if present,
142     otherwise the first name alphabetically.
143     """
144     names = provider_names(cfg)
145     declared = cfg.get("default_provider") if isinstance(cfg, dict) else None
146     if isinstance(declared, str) and declared in names:
147         return declared
148     if "fireworks" in names:
149         return "fireworks"
150     if names:
151         return names[0]
152     return None
153 
154 
155 def config_active_provider_name():
156     global _active_provider_name
157     if _active_provider_name is None:
158         _active_provider_name = _pick_default_provider_name(harness_config)
159     return _active_provider_name
160 
161 
162 def config_set_active_provider(name):
163     """Switch the active profile. Returns the resulting active name."""
164     global _active_provider_name
165     if name is not None and config_provider(name):
166         _active_provider_name = str(name)
167     return _active_provider_name
168 
169 
170 def config_active_provider():
171     """-> provider dict of the active profile, or None when there is no config."""
172     n = config_active_provider_name()
173     return config_provider(n) if n else None
174 
175 
176 # ---------------------------------------------------------------------------
177 # Provider field accessors (all tolerant of missing keys)
178 
179 def provider_type(provider):
180     """-> 'mlx' | 'openai' -- defaults to 'openai'.
181 
182     "mlx" selects the local mlx-serve backend (formerly "ollama"); "ollama",
183     "omlx", and "sushi" are also accepted and mapped to 'mlx for compatibility.
184     """
185     t = provider.get("type") if provider else None
186     if isinstance(t, str):
187         low = t.strip().lower()
188         if low in ("mlx", "ollama", "omlx", "sushi"):
189             return "mlx"
190     return "openai"
191 
192 
193 def provider_endpoint(provider):
194     e = provider.get("endpoint") if provider else None
195     return e if isinstance(e, str) and e != "" else None
196 
197 
198 def provider_model(provider):
199     m = provider.get("model") if provider else None
200     return m if isinstance(m, str) and m != "" else None
201 
202 
203 def provider_api_key_env(provider):
204     """Name of the env var holding the Bearer key for this endpoint, or None.
205 
206     Absent/empty/null means "no key" (plain local MLX).
207     """
208     k = provider.get("api_key_env") if provider else None
209     return k if isinstance(k, str) and k != "" else None
210 
211 
212 def provider_generation(provider):
213     g = provider.get("generation") if provider else None
214     return g if isinstance(g, dict) else {}
215 
216 
217 def provider_pricing(provider):
218     """-> dict of per-1M-token USD rates ('input', 'cached_input', 'output'),
219     or an empty dict when the profile declares none."""
220     g = provider.get("pricing") if provider else None
221     return g if isinstance(g, dict) else {}
222 
223 
224 def pricing_ref(pricing, key):
225     """-> number, or None when the profile does not declare that rate.
226 
227     The None result means "unknown", which callers report rather than
228     guessing a value.
229     """
230     if not isinstance(pricing, dict):
231         return None
232     if key in pricing:
233         return pricing[key]
234     return None
235 
236 
237 def generation_ref(generation, key, default=None):
238     """Fetch a generation parameter ("temperature", "max_tokens", "think", ...)."""
239     if not isinstance(generation, dict):
240         return default
241     return generation.get(key, default)
242 
243 
244 # ---------------------------------------------------------------------------
245 # Debug helper
246 
247 def print_config_summary():
248     def loaded(path):
249         return "(loaded)" if os.path.isfile(path) else "(absent)"
250     print("Config files: {} {} / {} {}".format(
251         GLOBAL_CONFIG_PATH, loaded(GLOBAL_CONFIG_PATH),
252         local_config_path(), loaded(local_config_path())))
253     print("Providers:    {}".format(", ".join(config_provider_names())))
254     print("Active:       {}".format(config_active_provider_name() or "(defaults)"))

 1 {
 2   "type": "function",
 3   "function": {
 4     "name": "read_file",
 5     "description": "Read and return the contents of a file. Refuses hidden/internal paths (~, #...#, dotfiles) and paths outside the working directory.",
 6     "parameters": {
 7       "type": "object",
 8       "properties": {
 9         "path": {
10           "type": "string",
11           "description": "File path relative to the working directory."
12         }
13       },
14       "required": ["path"]
15     }
16   }
17 }

  1 # tools.py -- tool registry and coding-agent tools
  2 # Python port of tools.rkt (originally py-coding-agent/tools.py).
  3 #
  4 # Copyright (C) 2026 Mark Watson <markw@markwatson.com>
  5 # Licensed under the GNU Affero General Public License v3.0 (AGPL-3.0)
  6 # See LICENSE file for details
  7 #
  8 # Six tools: read_file, list_dir, grep, run_shell, propose_edit,
  9 # replace_in_file. The two edit tools show a colored diff and ask y/n/s;
 10 # propose_edit then gates on `make check`.
 11 
 12 import json
 13 import os
 14 import re
 15 import shlex
 16 import subprocess
 17 
 18 from approval import unified_diff, print_colored_diff, prompt_yes_no_skip, prompt_reason
 19 
 20 # ---------------------------------------------------------------------------
 21 # Registry
 22 
 23 registry = {}
 24 
 25 SHELL_WHITELIST = {"make", "ls", "pwd", "cat", "uv"}
 26 MAX_CHECK_OUTPUT_CHARS = 2000
 27 MAX_SHELL_OUTPUT_CHARS = 20000
 28 GREP_MAX_MATCHES = 200
 29 GREP_MAX_LINE_CHARS = 300
 30 SHELL_TIMEOUT = 300
 31 CHECK_TIMEOUT = 600
 32 TIMEOUT_EXIT_CODE = 124
 33 
 34 # CLI-controlled modes (mutated by agent.py, read by the tools)
 35 auto_approve = False
 36 dry_run = False
 37 quiet_mode = False
 38 
 39 # Set by tool_propose_edit / tool_replace_in_file when an applied change leaves
 40 # `make check` failing. agent.py reads it to pick the process exit code for a
 41 # one-shot run instead of re-scanning old tool output.
 42 make_check_failed = False
 43 
 44 
 45 def define_tool(name, params, description, handler):
 46     """params: list of (pname, ptype, pdesc) tuples."""
 47     registry[name] = {
 48         "name": name,
 49         "description": description,
 50         "parameters": params,
 51         "handler": handler,
 52     }
 53 
 54 
 55 def render_tools(names):
 56     rendered = []
 57     for name in names:
 58         tool = registry.get(name)
 59         if tool is None:
 60             raise ValueError("Undefined tool: {}".format(name))
 61         props = {}
 62         required = []
 63         for (pname, ptype, pdesc) in tool["parameters"]:
 64             props[pname] = {"type": ptype, "description": pdesc}
 65             required.append(pname)
 66         rendered.append({
 67             "type": "function",
 68             "function": {
 69                 "name": tool["name"],
 70                 "description": tool["description"],
 71                 "parameters": {
 72                     "type": "object",
 73                     "properties": props,
 74                     "required": required,
 75                 },
 76             },
 77         })
 78     return rendered
 79 
 80 
 81 # ---------------------------------------------------------------------------
 82 # Tool dispatch
 83 
 84 def call_tool(name, args):
 85     tool = registry.get(name)
 86     if tool is None:
 87         return "Error: unknown tool '{}'. Available tools: {}".format(
 88             name, ", ".join(sorted(registry)))
 89     params = tool["parameters"]
 90     # Missing required args? Return an actionable error describing the expected
 91     # argument list -- small models frequently emit malformed/truncated
 92     # arguments, and silently receiving None tends to send them into retry loops.
 93     # Match the Racket semantics: only a missing key or a JSON null counts as
 94     # missing -- the empty string is a VALID value (propose_edit passes ""
 95     # and replace_in_file may pass an empty replacement).
 96     missing = [p[0] for p in params if args.get(p[0]) is None]
 97     if missing:
 98         return ("Error: tool '{}' missing required argument(s): {}. "
 99                 "Expected arguments (JSON object): {}".format(
100                     name, ", ".join(missing),
101                     ", ".join(p[0] for p in params)))
102     positional = [args.get(p[0]) for p in params]
103     try:
104         result = tool["handler"](*positional)
105         return str(result) if result is not None else ""
106     except Exception as e:  # noqa: BLE001
107         return "Error: tool '{}' raised: {}  (check argument types/values)".format(name, e)
108 
109 
110 def execute_tool_calls(tool_calls):
111     """tool-calls: list of dicts with 'id', 'function' {name, arguments}.
112 
113     Returns list of (call_id, name, result_str) tuples.
114     """
115     results = []
116     for call in tool_calls:
117         call_id = call.get("id", "")
118         func = call.get("function") or {}
119         name = func.get("name", "")
120         name = name if isinstance(name, str) else str(name)
121         args_json = func.get("arguments", "{}")
122         if args_json is None or not isinstance(args_json, str):
123             args_json = "{}" if args_json is None else json.dumps(args_json)
124         short = args_json if len(args_json) <= 120 else args_json[:117] + "..."
125         if not quiet_mode:
126             print("* {} {}".format(name, short))
127         bad_json = None
128         try:
129             parsed = json.loads(args_json)
130             args_parsed = parsed if isinstance(parsed, dict) else "NOT-OBJECT"
131         except Exception as e:  # noqa: BLE001
132             # The decode error itself ("Expecting ',' delimiter: line 3 ...")
133             # tells the model what to fix, so pass it through.
134             bad_json = str(e)
135             args_parsed = "BAD-JSON"
136         if isinstance(args_parsed, dict):
137             # Coerce non-string values to strings so handlers behave like the
138             # Racket version's format-based rendering.
139             args_parsed = {k: (v if isinstance(v, str) else json.dumps(v))
140                            for k, v in args_parsed.items()}
141         if str(name).strip() == "":
142             # Truncated tool call -- the model stopped mid-generation, so no
143             # function name survived. Feed that back instead of crashing.
144             result = ("Error: the model's tool call was truncated mid-generation "
145                       "(no function name provided). Received arguments: {}".format(short))
146         elif bad_json is not None:
147             result = "Error: invalid JSON in arguments for tool '{}': {}. Received: {}".format(
148                 name, bad_json, short)
149         elif args_parsed == "NOT-OBJECT":
150             result = "Error: arguments for tool '{}' must be a JSON object. Received: {}".format(name, short)
151         else:
152             result = call_tool(name, args_parsed)
153         # Guarantee a usable id: some local servers omit tool_call ids, and the
154         # matching role:"tool" message needs one to line up.
155         if not isinstance(call_id, str) or call_id == "":
156             call_id = "call_{}".format(len(results))
157         results.append((call_id, name, result))
158     return results
159 
160 
161 # ---------------------------------------------------------------------------
162 # Helpers: run subprocess and capture combined output
163 
164 def run_external(exe, args, timeout=SHELL_TIMEOUT):
165     """exe: str, args: list of str -> (combined_output, exit_code).
166 
167     A timeout is reported as exit code 124 rather than raised, so a hung
168     command becomes tool feedback instead of an exception.
169     """
170     try:
171         proc = subprocess.run([exe] + list(args),
172                               stdin=subprocess.DEVNULL,
173                               stdout=subprocess.PIPE, stderr=subprocess.PIPE,
174                               text=True, errors="replace", timeout=timeout)
175     except subprocess.TimeoutExpired:
176         return ("command timed out after {}s: {} {}".format(
177             timeout, exe, " ".join(str(a) for a in args)), TIMEOUT_EXIT_CODE)
178     return (proc.stdout + proc.stderr), proc.returncode
179 
180 
181 def truncate_string(s, max_len):
182     if len(s) > max_len:
183         return s[:max_len] + "\n... (truncated, {} total chars)".format(len(s))
184     return s
185 
186 
187 def strip_shell_quotes(s):
188     if len(s) >= 2 and ((s[0] == '"' and s[-1] == '"') or (s[0] == "'" and s[-1] == "'")):
189         return s[1:-1]
190     return s
191 
192 
193 # Hidden files (ignored from listings and rejected on read/search/run):
194 #   - names ending in ~  (e.g. foo.rkt~)
195 #   - names wrapped in #...#  (e.g. #foo.rkt#)
196 #   - names starting with .  (e.g. .git, .gitignore, .env)
197 def hidden_file(name):
198     return (name.endswith("~")
199             or (name.startswith("#") and name.endswith("#"))
200             or name.startswith("."))
201 
202 
203 def hidden_in_path(path):
204     """True when any component of `path` names a hidden/internal file.
205 
206     Checking every component (not just the basename) is what keeps the tools
207     out of directories such as .git/ or ~/.ssh/. "." and ".." are ordinary
208     path syntax, not hidden names, so they are skipped here.
209     """
210     for part in re.split(r"[\\/]+", str(path)):
211         if part in ("", ".", ".."):
212             continue
213         if hidden_file(part):
214             return True
215     return False
216 
217 
218 def hidden_arg(s):
219     cleaned = strip_shell_quotes(s)
220     return (not cleaned.startswith("-")) and hidden_in_path(cleaned)
221 
222 
223 def working_dir():
224     """The directory the agent is allowed to touch (established by --cwd)."""
225     return os.path.realpath(os.getcwd())
226 
227 
228 def path_within_working_dir(path):
229     """True when `path` resolves to the working directory or something inside it."""
230     root = working_dir()
231     full = os.path.realpath(os.path.join(root, os.path.expanduser(path)))
232     return full == root or full.startswith(root + os.sep)
233 
234 
235 def resolve_tool_path(path):
236     """-> (absolute path, None) or (None, refusal message).
237 
238     Every tool that touches the filesystem funnels through this: hidden
239     components are refused, and anything outside the working directory (../,
240     absolute paths, symlinks that escape) is refused as well.
241     """
242     if not isinstance(path, str) or path.strip() == "":
243         return None, "refusing: empty path"
244     if hidden_in_path(path):
245         return None, "refusing to touch hidden/internal path: {}".format(path)
246     full = os.path.realpath(os.path.join(working_dir(), os.path.expanduser(path)))
247     if full != working_dir() and not full.startswith(working_dir() + os.sep):
248         return None, ("refusing: {} is outside the working directory {}. "
249                       "Use a path relative to it.").format(path, working_dir())
250     return full, None
251 
252 
253 def looks_like_long_ls_line(line):
254     """True for `ls -l` rows: permissions, links, owner, group, size, date, name."""
255     parts = line.split(None, 1)
256     if not parts:
257         return False
258     mode = parts[0]
259     return len(mode) >= 10 and mode[0] in "-dlbcps" and set(mode[1:10]) <= set("rwxSsTt-")
260 
261 
262 def filter_ls_output(out):
263     """Drop `total N` headers, the . / .. rows, and hidden names.
264 
265     Long-format rows are split into at most eight fields so that a filename
266     containing spaces survives intact; short-format rows are split into one
267     name per whitespace-separated token.
268     """
269     result_lines = []
270     for line in out.split("\n"):
271         t = line.strip()
272         if t == "" or t.startswith("total "):
273             continue
274         if looks_like_long_ls_line(t):
275             fields = t.split(None, 8)
276             fname = fields[8] if len(fields) > 8 else fields[-1]
277             if fname in (".", "..") or hidden_file(fname):
278                 continue
279         else:
280             names = [n for n in t.split()
281                      if n not in (".", "..") and not hidden_file(n)]
282             if not names:
283                 continue
284             line = " ".join(names)
285         result_lines.append(line)
286     return "\n".join(result_lines)
287 
288 
289 # ---------------------------------------------------------------------------
290 # Tool implementations
291 
292 def tool_read_file(path):
293     full, refusal = resolve_tool_path(path)
294     if refusal:
295         return refusal
296     try:
297         with open(full, "r", encoding="utf-8", errors="replace") as f:
298             return f.read()
299     except Exception as e:  # noqa: BLE001
300         return "Error reading {}: {}".format(path, e)
301 
302 
303 def tool_list_dir(path):
304     full, refusal = resolve_tool_path(path)
305     if refusal:
306         return refusal
307     try:
308         entries = sorted(os.listdir(full))
309         lines = []
310         for e in entries:
311             if hidden_file(e):
312                 continue
313             child = os.path.join(full, e)
314             lines.append(e + "/" if os.path.isdir(child) else e)
315         return "\n".join(lines)
316     except Exception as e:  # noqa: BLE001
317         return "Error listing {}: {}".format(path, e)
318 
319 
320 def tool_grep(pattern, path):
321     """Recursive regex search that skips hidden files and directories.
322 
323     Implemented with os.walk + re instead of shelling out to `grep -rnE` for
324     three reasons: hidden directories (.git/) stay out of the model's context,
325     a pattern beginning with "-" cannot be mistaken for a flag, and the result
326     can be capped and truncated instead of flooding the window.
327     """
328     root, refusal = resolve_tool_path(path)
329     if refusal:
330         return refusal
331     try:
332         rx = re.compile(pattern)
333     except re.error as e:
334         return "Error: invalid regular expression '{}': {}".format(pattern, e)
335 
336     targets = [root] if os.path.isfile(root) else None
337     if targets is None and not os.path.isdir(root):
338         return "Error: no such file or directory: {}".format(path)
339 
340     matches = []
341     truncated = False
342 
343     def add_match(text):
344         nonlocal truncated
345         if len(matches) >= GREP_MAX_MATCHES:
346             truncated = True
347             return
348         matches.append(truncate_string(text, GREP_MAX_LINE_CHARS))
349 
350     if targets:
351         files = targets
352     else:
353         files = []
354         for dirpath, dirnames, filenames in os.walk(root):
355             dirnames[:] = sorted(d for d in dirnames if not hidden_file(d))
356             for fname in sorted(filenames):
357                 if not hidden_file(fname):
358                     files.append(os.path.join(dirpath, fname))
359 
360     for fname in files:
361         try:
362             with open(fname, "r", encoding="utf-8", errors="replace") as f:
363                 for lineno, line in enumerate(f, 1):
364                     if rx.search(line):
365                         rel = os.path.relpath(fname, working_dir())
366                         add_match("{}:{}:{}".format(rel, lineno, line.rstrip("\n")))
367                         if truncated:
368                             break
369         except (OSError, UnicodeError):
370             continue
371         if truncated:
372             break
373     if not matches:
374         return "no matches for {!r} under {}".format(pattern, os.path.relpath(root, working_dir()))
375     if truncated:
376         matches.append("... (stopped after {} matches; narrow the pattern or path)".format(
377             GREP_MAX_MATCHES))
378     return "\n".join(matches)
379 
380 
381 def tool_run_shell(command):
382     command = command.strip()
383     try:
384         tokens = shlex.split(command)
385     except ValueError:
386         tokens = command.split()
387     if not tokens:
388         return "empty command"
389     cmd = tokens[0]
390     if cmd not in SHELL_WHITELIST:
391         return "Command '{}' not whitelisted. Allowed: {}".format(
392             cmd, ", ".join(sorted(SHELL_WHITELIST)))
393     for a in tokens[1:]:
394         if hidden_arg(a):
395             return "refusing to run command referencing hidden/internal file: {}".format(a)
396         # Note: the whitelist limits WHICH programs run. It does not sandbox
397         # what they do -- `uv run` and `make` execute arbitrary code by design.
398         try:
399             inside = path_within_working_dir(strip_shell_quotes(a))
400         except (OSError, ValueError):
401             inside = False
402         if not inside:
403             return ("refusing to run command referencing a path outside the working "
404                     "directory: {}".format(a))
405     try:
406         out, code = run_external(cmd, tokens[1:], timeout=SHELL_TIMEOUT)
407         filtered = filter_ls_output(out) if cmd == "ls" else out
408         return "{}(exit {})".format(truncate_string(filtered, MAX_SHELL_OUTPUT_CHARS), code)
409     except Exception as e:  # noqa: BLE001
410         return "Error running command: {}".format(e)
411 
412 
413 def run_make_check():
414     try:
415         return run_external("make", ["check"], timeout=CHECK_TIMEOUT)
416     except Exception as e:  # noqa: BLE001
417         return "make check error: {}".format(e), 1
418 
419 
420 def apply_change(path, full, new):
421     """Write the approved contents and report the `make check` verdict."""
422     global make_check_failed
423     parent = os.path.dirname(full)
424     if parent:
425         os.makedirs(parent, exist_ok=True)
426     with open(full, "w", encoding="utf-8") as f:
427         f.write(new)
428     out, status = run_make_check()
429     if status == 0:
430         return "applied; make check passed"
431     make_check_failed = True
432     return "applied; make check FAILED (exit {}):\n{}".format(
433         status, truncate_string(out, MAX_CHECK_OUTPUT_CHARS))
434 
435 
436 def gate_write(path, full, current, new, heading):
437     """Shared approval flow: diff, then y/n/s, then write + `make check`."""
438     diff_text = unified_diff(current, new, "a/" + path, "b/" + path)
439     print("")
440     if heading:
441         print(heading)
442     print_colored_diff(diff_text)
443 
444     if dry_run:
445         return "dry-run: diff shown, file not written (use without --dry-run to apply)"
446     if auto_approve:
447         # Safety: still show diff above, then auto-apply without prompting
448         if not quiet_mode:
449             print("[auto-approve: applying change without prompt]")
450         return apply_change(path, full, new).replace(
451             "applied;", "applied (auto-approved);", 1)
452     answer = prompt_yes_no_skip()
453     if answer == "no":
454         return "user rejected the change"
455     if answer == "skip":
456         reason = prompt_reason()
457         return "user skipped: {}".format(reason)
458     return apply_change(path, full, new)
459 
460 
461 def tool_propose_edit(path, old, new):
462     """Whole-file edit: `old` must match the current contents exactly ("" to create)."""
463     full, refusal = resolve_tool_path(path)
464     if refusal:
465         return refusal
466     exists = os.path.isfile(full)
467     if exists:
468         try:
469             with open(full, "r", encoding="utf-8", errors="replace") as f:
470                 current = f.read()
471         except Exception as e:  # noqa: BLE001
472             return "Error reading {}: {}".format(path, e)
473     else:
474         current = ""
475 
476     if exists and current != old:
477         return ("stale base: on-disk contents of {} do not match the 'old' you "
478                 "provided. Read the file again and retry.".format(path))
479     if exists and current == new:
480         return "no changes (proposed content matches current file)"
481     if not exists and new == "":
482         return "refused: cannot create an empty file"
483 
484     heading = "(new file: {})".format(path) if not exists else None
485     return gate_write(path, full, current, new, heading)
486 
487 
488 def tool_replace_in_file(path, old_string, new_string):
489     """Compact edit: replace one unique snippet instead of echoing the whole file."""
490     full, refusal = resolve_tool_path(path)
491     if refusal:
492         return refusal
493     if old_string == "":
494         return "refused: old_string is empty; use propose_edit to create a file"
495     try:
496         with open(full, "r", encoding="utf-8", errors="replace") as f:
497             current = f.read()
498     except Exception as e:  # noqa: BLE001
499         return "Error reading {}: {}".format(path, e)
500 
501     hits = current.count(old_string)
502     if hits == 0:
503         return ("old_string not found in {}. Read the file again and copy a unique "
504                 "snippet exactly, including indentation.".format(path))
505     if hits > 1:
506         return ("old_string appears {} times in {}. Include more surrounding "
507                 "context so the snippet is unique.".format(hits, path))
508     new = current.replace(old_string, new_string, 1)
509     if new == current:
510         return "no changes (replacement is identical to the matched text)"
511     return gate_write(path, full, current, new, None)
512 
513 
514 # ---------------------------------------------------------------------------
515 # Registration
516 
517 def register_all():
518     define_tool(
519         "read_file",
520         [("path", "string", "File path relative to the working directory.")],
521         "Read and return the contents of a file. Refuses hidden/internal paths (~, #...#, dotfiles) and paths outside the working directory.",
522         tool_read_file)
523     define_tool(
524         "list_dir",
525         [("path", "string", 'Directory path. Use "." for the working directory.')],
526         "List files and subdirectories (with trailing /) in a directory. Hidden/internal files (~, #...#, and dotfiles) are excluded.",
527         tool_list_dir)
528     define_tool(
529         "grep",
530         [("pattern", "string", "Python regular expression to search for."),
531          ("path", "string", "File or directory path to search, relative to the working directory.")],
532         "Recursively search files for PATTERN and return matching lines as path:line:text. Hidden/internal files and directories (dotfiles, .git) are skipped, and the result is capped at {} matches.".format(GREP_MAX_MATCHES),
533         tool_grep)
534     define_tool(
535         "run_shell",
536         [("command", "string", "Shell command. Only whitelisted commands may run: make, ls, pwd, cat, uv.")],
537         "Run a whitelisted shell command and return its combined output. Refuses hidden/internal paths and paths outside the working directory. The whitelist limits which programs run, not what they do: make and uv execute project code.",
538         tool_run_shell)
539     define_tool(
540         "propose_edit",
541         [("path", "string", "Path to the file to edit or create."),
542          ("old", "string", "For an existing file: the exact current contents. For a new file: pass empty string."),
543          ("new", "string", "The proposed new contents of the file, in full.")],
544         "Propose a whole-file edit or new-file creation. The user is shown a unified diff and asked to approve. On approval the file is written and `make check` is run. Prefer replace_in_file for small changes to an existing file.",
545         tool_propose_edit)
546     define_tool(
547         "replace_in_file",
548         [("path", "string", "Path to an existing file inside the working directory."),
549          ("old_string", "string", "The exact snippet to replace. It must appear exactly once in the file."),
550          ("new_string", "string", "The replacement text. Pass empty string to delete the snippet.")],
551         "Propose a surgical edit: replace one unique snippet of an existing file. Far cheaper than propose_edit when the file is large. The user is shown a unified diff and asked to approve; on approval the file is written and `make check` is run.",
552         tool_replace_in_file)
553 
554 
555 ENABLED_TOOLS = ["read_file", "list_dir", "grep", "run_shell", "propose_edit",
556                  "replace_in_file"]

1 user skipped: do not touch the public API, add a wrapper instead

 1 # approval.py -- colored diffs and y/n/s approval prompts
 2 # Python port of approval.rkt (mirrors py-coding-agent/approval.py).
 3 #
 4 # Copyright (C) 2026 Mark Watson <markw@markwatson.com>
 5 # Licensed under the GNU Affero General Public License v3.0 (AGPL-3.0)
 6 # See LICENSE file for details
 7 
 8 import difflib
 9 import sys
10 
11 # ANSI colours
12 
13 ANSI_RED = "\033[31m"
14 ANSI_GREEN = "\033[32m"
15 ANSI_CYAN = "\033[36m"
16 ANSI_RESET = "\033[0m"
17 
18 # When False, print diffs without ANSI (for --plain / --no-color / piped output)
19 color_enabled = True
20 
21 
22 def set_color_enabled(enabled):
23     """Turn ANSI colors on/off (used by --plain and the plain/quiet config)."""
24     global color_enabled
25     color_enabled = bool(enabled)
26 
27 
28 # ---------------------------------------------------------------------------
29 # unified_diff : str str str str -> str
30 # difflib-based replacement for running `diff -u` on two temporary files.
31 
32 def unified_diff(old_content, new_content, old_label, new_label):
33     old_lines = old_content.splitlines(keepends=True)
34     new_lines = new_content.splitlines(keepends=True)
35     # difflib omits the trailing newline on the last line; add one so the
36     # rendered text matches `diff -u` output line for line.
37     diff = difflib.unified_diff(old_lines, new_lines,
38                                 fromfile=old_label, tofile=new_label)
39     text = "".join(diff)
40     if text and not text.endswith("\n"):
41         text += "\n"
42     return text
43 
44 
45 # ---------------------------------------------------------------------------
46 # print_colored_diff : str -> None
47 
48 def print_colored_diff(diff_text):
49     for line in diff_text.split("\n"):
50         if color_enabled and (line.startswith("+++") or line.startswith("---")
51                               or line.startswith("@@")):
52             print(ANSI_CYAN + line + ANSI_RESET)
53         elif color_enabled and line.startswith("+"):
54             print(ANSI_GREEN + line + ANSI_RESET)
55         elif color_enabled and line.startswith("-"):
56             print(ANSI_RED + line + ANSI_RESET)
57         else:
58             print(line)
59 
60 
61 # ---------------------------------------------------------------------------
62 # prompt_yes_no_skip : -> 'yes' | 'no' | 'skip'
63 
64 def prompt_yes_no_skip():
65     """-> 'yes' | 'no' | 'skip'. EOF (no human present) counts as 'no'."""
66     while True:
67         sys.stdout.write("\nApply this change? [y]es / [n]o / [s]kip and tell the model why: ")
68         sys.stdout.flush()
69         try:
70             line = input("")
71         except EOFError:
72             print("\n(no answer available: refusing the change)")
73             return "no"
74         norm = (line or "").strip().lower()
75         if norm in ("y", "yes"):
76             return "yes"
77         if norm in ("n", "no"):
78             return "no"
79         if norm in ("s", "skip"):
80             return "skip"
81         print("Please answer y, n, or s.")
82 
83 
84 # ---------------------------------------------------------------------------
85 # prompt_reason : -> str
86 
87 def prompt_reason():
88     sys.stdout.write("Reason (one line): ")
89     sys.stdout.flush()
90     try:
91         return input("")
92     except EOFError:
93         return ""

  1 # chat_loop.py -- provider-agnostic agentic tool-calling loop, shared by
  2 # fireworks_ai.py and mlx_serve.py. Python port of chat-loop.rkt.
  3 #
  4 # Copyright (C) 2026 Mark Watson <markw@markwatson.com>
  5 # Licensed under the GNU Affero General Public License v3.0 (AGPL-3.0)
  6 # See LICENSE file for details
  7 #
  8 # The `post_fn` argument adapts an OpenAI-style chat-completions payload to a
  9 # specific backend and returns a normalized response dict:
 10 #   { 'choices': [ { 'message': <assistant msg>, 'finish_reason': ... } ],
 11 #     'usage':   { 'prompt_tokens': n, 'completion_tokens': n, ... } }
 12 # where <assistant msg> is { 'role': "assistant", 'content': <string> }
 13 # plus, when the model called tools, 'tool_calls' -- a list of
 14 #   { 'id': <string>, 'type': "function",
 15 #     'function': { 'name': <string>, 'arguments': <json string> } }
 16 
 17 from tools import render_tools, execute_tool_calls
 18 
 19 # Repetition detection: track the last few tool-call signatures so a model
 20 # stuck issuing the identical failing call is stopped early rather than
 21 # burning all max-iterations. A signature is (name + args-json) per call,
 22 # sorted, so multi-call batches compare as a set.
 23 REPEAT_WINDOW = 5   # remember the last N batches
 24 REPEAT_LIMIT = 2    # the same batch seen this many times => stuck
 25 
 26 MAX_ITERATIONS_DEFAULT = 20
 27 
 28 
 29 # ---------------------------------------------------------------------------
 30 # Helpers
 31 
 32 def response_message(data):
 33     """Extract the assistant message dict from a normalized response, or None."""
 34     choices = data.get("choices")
 35     if isinstance(choices, list) and choices:
 36         choice = choices[0]
 37         if isinstance(choice, dict):
 38             return choice.get("message")
 39     return None
 40 
 41 
 42 def msg_content(msg):
 43     """The assistant message's 'content', coerced to a string.
 44 
 45     Some OpenAI-compatible servers (e.g. mlx_lm.server) emit "content": null
 46     on tool-only responses; coerce any non-string value to "" here.
 47     """
 48     c = msg.get("content", "") if msg else ""
 49     return c if isinstance(c, str) else ""
 50 
 51 
 52 def clean_assistant_message(msg):
 53     """A copy of an assistant message safe to send back to a strict server.
 54 
 55     Two normalizations happen here. A null or non-string 'content' becomes ""
 56     (mlx_lm.server sends "" on tool-only responses, the OpenAI spec sends
 57     null, and replaying the null back can be rejected). A missing tool-call id
 58     gets a generated one at index 0, 1, ... -- the matching role:"tool" reply
 59     is addressed by that id, so an empty id breaks the pairing.
 60     """
 61     tool_calls = msg.get("tool_calls")
 62     has_calls = isinstance(tool_calls, list) and bool(tool_calls)
 63     out = {"role": "assistant", "content": msg_content(msg)}
 64     if not has_calls:
 65         return out
 66     cleaned = []
 67     for i, tc in enumerate(tool_calls):
 68         func = tc.get("function") or {}
 69         call_id = tc.get("id")
 70         cleaned.append({
 71             "id": call_id if isinstance(call_id, str) and call_id != ""
 72                   else "call_{}".format(i),
 73             "type": tc.get("type") or "function",
 74             "function": {"name": func.get("name", ""),
 75                          "arguments": func.get("arguments", "{}")},
 76         })
 77     out["tool_calls"] = cleaned
 78     reasoning = msg.get("reasoning_content")
 79     if isinstance(reasoning, str) and reasoning != "":
 80         out["reasoning_content"] = reasoning
 81     return out
 82 
 83 
 84 # ---------------------------------------------------------------------------
 85 # Request payload
 86 
 87 def request_payload(model_id, max_tokens, temperature, messages):
 88     """Build a request body, omitting generation parameters the active
 89     provider profile did not declare. No max_tokens/temperature default is
 90     compiled in here; the provider config is the only source for them."""
 91     payload = {"model": model_id, "messages": messages}
 92     if max_tokens is not None:
 93         payload["max_tokens"] = max_tokens
 94     if temperature is not None:
 95         payload["temperature"] = temperature
 96     return payload
 97 
 98 
 99 # ---------------------------------------------------------------------------
100 # chat* : post_fn messages ... -> str
101 
102 def chat(post_fn, messages, model_id, max_tokens=None, temperature=None):
103     payload = request_payload(model_id, max_tokens, temperature, messages)
104     data = post_fn(payload)
105     msg = response_message(data)
106     content = msg_content(msg)
107     return content if content else "No response content"
108 
109 
110 # ---------------------------------------------------------------------------
111 # chat_with_tools : -> (final_text, final_messages)
112 # Multi-turn agentic loop.
113 
114 def chat_with_tools(post_fn, messages, tools, model_id,
115                     max_tokens=None, temperature=None,
116                     max_iterations=MAX_ITERATIONS_DEFAULT):
117     tools_rendered = render_tools(tools)
118     current_messages = list(messages)
119     recent_signatures = []
120 
121     def call_signature(tool_calls):
122         sig = []
123         for tc in tool_calls:
124             f = tc.get("function") or {}
125             sig.append("{}|{}".format(f.get("name", ""), f.get("arguments", "")))
126         return sorted(sig)
127 
128     def remember_and_count(sig):
129         """Record this batch and return how often it has now been seen.
130 
131         The batch is recorded BEFORE the count, so the second identical batch
132         reports 2 and trips REPEAT_LIMIT.
133         """
134         recent_signatures.append(sig)
135         del recent_signatures[:-REPEAT_WINDOW]
136         return sum(1 for s in recent_signatures if s == sig)
137 
138     def append_tool_results(results):
139         for (call_id, name, result_str) in results:
140             current_messages.append({
141                 "role": "tool",
142                 "tool_call_id": call_id,
143                 "name": name,
144                 "content": result_str,
145             })
146 
147     iter_ = 0
148     while True:
149         if iter_ >= max_iterations:
150             # Max iterations -- one final no-tools call for a summary.
151             payload = request_payload(model_id, max_tokens, temperature, current_messages)
152             try:
153                 data = post_fn(payload)
154                 msg = response_message(data)
155                 content = msg_content(msg)
156                 if content == "":
157                     content = "(no summary from model)"
158                 if msg:
159                     current_messages.append(clean_assistant_message(msg))
160                 return content, current_messages
161             except Exception:
162                 return "(max tool iterations reached)", current_messages
163         iter_ += 1
164 
165         payload = request_payload(model_id, max_tokens, temperature, current_messages)
166         if tools_rendered:
167             payload["tools"] = tools_rendered
168             payload["tool_choice"] = "auto"
169         data = post_fn(payload)
170         msg = response_message(data)
171         if msg is None:
172             raise RuntimeError("response has no 'message'. Raw: {}".format(data))
173         tool_calls = msg.get("tool_calls")
174         content = msg_content(msg)
175         assistant_msg = clean_assistant_message(msg)
176         current_messages.append(assistant_msg)
177 
178         if not tool_calls:
179             return (content if content else "(empty response from model)"), current_messages
180 
181         # Echo any reasoning the model emitted alongside its tool calls, then
182         # run the calls. Results are appended before the stuck-model check so
183         # the transcript keeps a role:"tool" reply for every call.
184         if content.strip():
185             print("")
186             print(content.strip())
187         batch = call_signature(tool_calls)
188         results = execute_tool_calls(tool_calls)
189         append_tool_results(results)
190         seen = remember_and_count(batch)
191         if seen >= REPEAT_LIMIT:
192             # The model is stuck re-issuing the identical call(s) -- bail out
193             # with an explanation instead of looping to max-iterations.
194             return ("(stopped: the model repeated the identical tool call(s) "
195                     "{} times without making progress; it may be too weak for "
196                     "this task or its arguments are malformed)".format(seen),
197                     current_messages)

1 data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_a","function":{"name":"grep"}}]}}]}
2 data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"pat"}}]}}]}
3 data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"tern\": \"foo\"}"}}]}}]}
4 data: [DONE]

  1 # fireworks_ai.py -- OpenAI-compatible (Fireworks AI) streaming API client,
  2 # session stats, chat helpers. Python port of fireworks-ai.rkt
  3 # (originally py-coding-agent/fireworks_ai.py).
  4 #
  5 # Copyright (C) 2026 Mark Watson <markw@markwatson.com>
  6 # Licensed under the GNU Affero General Public License v3.0 (AGPL-3.0)
  7 # See LICENSE file for details
  8 #
  9 # Endpoint, model, api_key_env, generation parameters, and pricing all come
 10 # from the active provider profile in the harness config (~/.coding_harness.json
 11 # and/or .local_coding_harness.json); no provider-specific value is compiled
 12 # in here.
 13 
 14 import json
 15 import os
 16 import threading
 17 
 18 import requests
 19 
 20 from harness_config import (config_active_provider, provider_api_key_env,
 21                             provider_endpoint, provider_generation,
 22                             provider_model, provider_pricing, pricing_ref,
 23                             generation_ref)
 24 from chat_loop import chat as chat_star
 25 from chat_loop import chat_with_tools as chat_with_tools_star
 26 from chat_loop import MAX_ITERATIONS_DEFAULT
 27 
 28 # ---------------------------------------------------------------------------
 29 # Modes and timeouts
 30 
 31 debug_log = False  # shared /debug toggle (agent.py flips this)
 32 
 33 # Requests use SSE streaming ("stream": true), so there is NO total
 34 # wall-clock cap on generation: a long response that keeps producing
 35 # tokens simply keeps streaming. The only remaining timeouts are:
 36 #   CONNECT_MAX_TIME    -- seconds to establish the TCP connection.
 37 #   STREAM_IDLE_TIMEOUT -- seconds the server may stay silent, both while
 38 #                          waiting for response headers (time to first token)
 39 #                          and between chunks. Tokens arriving periodically
 40 #                          never trip this; only a stalled connection does.
 41 CONNECT_MAX_TIME = 10
 42 STREAM_IDLE_TIMEOUT = 300
 43 
 44 # ---------------------------------------------------------------------------
 45 # Pricing -- read from the active provider profile's "pricing" block (USD per
 46 # 1M tokens). A profile that declares no pricing yields None rates, and
 47 # callers report the cost as unknown instead of inventing a number.
 48 
 49 def active_pricing():
 50     return provider_pricing(config_active_provider())
 51 
 52 
 53 # ---------------------------------------------------------------------------
 54 # Session stats (thread-safe)
 55 
 56 _stats_lock = threading.Lock()
 57 _session_prompt_tokens = 0
 58 _session_completion_tokens = 0
 59 _session_total_tokens = 0
 60 _session_cached_tokens = 0
 61 
 62 
 63 def reset_session_stats():
 64     global _session_prompt_tokens, _session_completion_tokens
 65     global _session_total_tokens, _session_cached_tokens
 66     with _stats_lock:
 67         _session_prompt_tokens = 0
 68         _session_completion_tokens = 0
 69         _session_total_tokens = 0
 70         _session_cached_tokens = 0
 71 
 72 
 73 # Costs use the active provider's configured rates. Each returns None when
 74 # the profile declares no such rate, so callers can report "unknown" instead
 75 # of a misleading $0.00.
 76 
 77 def rate_cost(tokens, rate):
 78     return (tokens * rate / 1_000_000) if rate is not None else None
 79 
 80 
 81 def prompt_cost(tokens):
 82     return rate_cost(tokens, pricing_ref(active_pricing(), "input"))
 83 
 84 
 85 def cached_cost(tokens):
 86     return rate_cost(tokens, pricing_ref(active_pricing(), "cached_input"))
 87 
 88 
 89 def completion_cost(tokens):
 90     return rate_cost(tokens, pricing_ref(active_pricing(), "output"))
 91 
 92 
 93 def session_snapshot():
 94     """-> (prompt, cached, completion, total) token totals, read under one lock."""
 95     with _stats_lock:
 96         return (_session_prompt_tokens, _session_cached_tokens,
 97                 _session_completion_tokens, _session_total_tokens)
 98 
 99 
100 # Cached input tokens are reported by the server in
101 # usage.prompt_tokens_details.cached_tokens and are part of prompt_tokens;
102 # bill them at the discounted rate and subtract them from the uncached pool.
103 def session_cost(snapshot=None):
104     """-> number, or None when the active profile declares no pricing at all."""
105     rates = active_pricing()
106     inp = pricing_ref(rates, "input")
107     cached = pricing_ref(rates, "cached_input")
108     out = pricing_ref(rates, "output")
109     if inp is None and cached is None and out is None:
110         return None
111     pt, ca, ct = snapshot or session_snapshot()[:3]
112     return ((rate_cost(max(0, pt - ca), inp) or 0)
113             + (rate_cost(ca, cached) or 0)
114             + (rate_cost(ct, out) or 0))
115 
116 
117 def print_session_stats():
118     pt, ca, ct, tt = session_snapshot()
119     cost = session_cost((pt, ca, ct))
120     rates = active_pricing()
121     print("")
122     print("Session token usage:")
123     print("  Prompt tokens:     {}".format(pt))
124     print("  Completion tokens: {}".format(ct))
125     print("  Total tokens:      {}".format(tt))
126     if ca > 0:
127         pct = 100.0 * ca / max(1, pt)
128         print("  Cached tokens:     {} ({:.1f}% of prompt)".format(ca, pct))
129     if cost is not None:
130         print("  Estimated cost:    ${:.6f}  (${:.4f}/M input, ${:.4f}/M cached input, ${:.4f}/M output)".format(
131             cost,
132             pricing_ref(rates, "input") or 0,
133             pricing_ref(rates, "cached_input") or 0,
134             pricing_ref(rates, "output") or 0))
135     else:
136         print('  Estimated cost:    n/a (no "pricing" block for this provider)')
137 
138 
139 def accumulate_usage(data):
140     global _session_prompt_tokens, _session_completion_tokens
141     global _session_total_tokens, _session_cached_tokens
142     usage = data.get("usage") or {}
143     if isinstance(usage, dict) and usage:
144         with _stats_lock:
145             _session_prompt_tokens += usage.get("prompt_tokens", 0)
146             _session_completion_tokens += usage.get("completion_tokens", 0)
147             _session_total_tokens += usage.get("total_tokens", 0)
148             details = usage.get("prompt_tokens_details") or {}
149             if isinstance(details, dict):
150                 _session_cached_tokens += details.get("cached_tokens", 0)
151 
152 
153 # ---------------------------------------------------------------------------
154 # API key
155 #
156 # The env var name comes from the active provider profile's api_key_env when
157 # a harness config is loaded; falls back to FIREWORKS_API_KEY.
158 
159 
160 def get_api_key():
161     p = config_active_provider()
162     env_name = (provider_api_key_env(p) if p else None) or "FIREWORKS_API_KEY"
163     key = os.environ.get(env_name, "")
164     if not key:
165         raise RuntimeError("fireworks-ai: {} environment variable not set".format(env_name))
166     return key
167 
168 
169 # ---------------------------------------------------------------------------
170 # SSE streaming helpers
171 
172 def parse_sse_chunk(body):
173     """Parse one SSE "data: {...}" body into a dict (or None on bad JSON)."""
174     try:
175         return json.loads(body)
176     except Exception:
177         return None
178 
179 
180 def parse_sse_response(resp):
181     """Reconstruct the equivalent non-streaming chat-completions response from
182     an SSE stream:
183 
184         { 'id': ..., 'model': ...,
185           'choices': [ { 'message': <assistant msg>, 'finish_reason': ... } ],
186           'usage': {...} }
187 
188     `message` carries accumulated 'content', (deepseek) 'reasoning_content',
189     and (when present) a list of 'tool_calls' dicts exactly like a
190     non-streaming response.
191     """
192     content_parts = []
193     reasoning_parts = []
194     tool_calls_by_index = {}  # index -> {'id','type','name','arguments'}
195     usage = None
196     finish_reason = None
197     message_id = ""
198     message_model = ""
199 
200     for raw in resp.iter_lines():
201         if raw is None:
202             continue
203         line = raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw
204         trimmed = line.strip()
205         if trimmed == "" or trimmed.startswith(":"):  # comment / keep-alive
206             continue
207         if not trimmed.startswith("data:"):
208             continue
209         body = trimmed[5:].strip()
210         if body == "[DONE]":
211             continue
212         chunk = parse_sse_chunk(body)
213         if not isinstance(chunk, dict):
214             continue
215         # API-level error inside the stream
216         if "error" in chunk:
217             err = chunk["error"]
218             msg = err.get("message", str(err)) if isinstance(err, dict) else str(err)
219             raise RuntimeError("API error: {}".format(msg))
220         if "id" in chunk:
221             message_id = chunk.get("id", "")
222         if "model" in chunk:
223             message_model = chunk.get("model", "")
224         chunk_usage = chunk.get("usage")
225         if isinstance(chunk_usage, dict):
226             usage = chunk_usage
227         for c in chunk.get("choices") or []:
228             delta = c.get("delta") or {}
229             fr = c.get("finish_reason")
230             if fr is not None and fr != finish_reason:
231                 finish_reason = fr
232             c_delta = delta.get("content")
233             if isinstance(c_delta, str):
234                 content_parts.append(c_delta)
235             r_delta = delta.get("reasoning_content")
236             if isinstance(r_delta, str):
237                 reasoning_parts.append(r_delta)
238             tc = delta.get("tool_calls")
239             if isinstance(tc, list):
240                 for t in tc:
241                     idx = t.get("index", 0)
242                     entry = tool_calls_by_index.get(idx)
243                     if entry is None:
244                         entry = {"id": "", "type": "function", "name": "", "arguments": ""}
245                         tool_calls_by_index[idx] = entry
246                     t_id = t.get("id")
247                     if isinstance(t_id, str) and t_id != "":
248                         entry["id"] = t_id
249                     t_type = t.get("type")
250                     if isinstance(t_type, str):
251                         entry["type"] = t_type
252                     f = t.get("function")
253                     if isinstance(f, dict):
254                         f_name = f.get("name")
255                         if isinstance(f_name, str) and f_name != "":
256                             entry["name"] = f_name
257                         f_args = f.get("arguments")
258                         if isinstance(f_args, str) and f_args != "":
259                             entry["arguments"] += f_args
260 
261     content = "".join(content_parts)
262     reasoning = "".join(reasoning_parts)
263     idxs = sorted(tool_calls_by_index.keys())
264     tool_calls = None
265     if idxs:
266         tool_calls = []
267         for idx in idxs:
268             e = tool_calls_by_index[idx]
269             tool_calls.append({
270                 "id": e["id"],
271                 "type": e["type"],
272                 "function": {"name": e["name"], "arguments": e["arguments"]},
273             })
274     if tool_calls:
275         message = {"role": "assistant", "content": content,
276                    "reasoning_content": reasoning, "tool_calls": tool_calls}
277     else:
278         message = {"role": "assistant", "content": content,
279                    "reasoning_content": reasoning}
280     return {
281         "id": message_id,
282         "model": message_model,
283         "choices": [{"message": message, "finish_reason": finish_reason}],
284         "usage": usage or {},
285     }
286 
287 
288 # ---------------------------------------------------------------------------
289 # Low-level POST (streaming)
290 
291 def post_fireworks(payload):
292     api_key = get_api_key()
293     p = config_active_provider()
294     endpoint = (provider_endpoint(p) if p else None)
295     if not endpoint:
296         raise RuntimeError('fireworks-ai: active provider profile has no "endpoint"; '
297                            "set it in the harness config")
298     headers = {
299         "content-type": "application/json",
300         "accept": "application/json",
301         "authorization": "Bearer " + api_key,
302     }
303     stream_payload = dict(payload)
304     stream_payload["stream"] = True
305     stream_payload["stream_options"] = {"include_usage": True}
306     if debug_log:
307         dbg = {k: v for k, v in stream_payload.items() if k != "messages"}
308         print("[DEBUG] request: {}".format(json.dumps(dbg)))
309     try:
310         resp = requests.post(endpoint, headers=headers, json=stream_payload,
311                              stream=True,
312                              timeout=(CONNECT_MAX_TIME, STREAM_IDLE_TIMEOUT))
313         data = parse_sse_response(resp)
314         resp.close()
315     except (KeyboardInterrupt, SystemExit):
316         raise  # let Ctrl-C reach the REPL as a cancelled turn
317     except Exception as e:  # noqa: BLE001
318         raise RuntimeError("fireworks-ai: HTTP error: {}".format(e))
319     if debug_log:
320         print("[DEBUG] response: {}".format(json.dumps(data)))
321     if isinstance(data, dict) and "error" in data:
322         err = data["error"]
323         msg = err.get("message", str(err)) if isinstance(err, dict) else str(err)
324         raise RuntimeError("API error: {}".format(msg))
325     accumulate_usage(data)
326     if "choices" not in data:
327         raise RuntimeError("fireworks-ai: response has no 'choices'. Raw: {}".format(json.dumps(data)))
328     return data
329 
330 
331 # ---------------------------------------------------------------------------
332 # chat / chat_with_tools -- thin wrappers over the shared provider-agnostic
333 # loop in chat_loop.py (also used by mlx_serve.py).
334 #
335 # Generation defaults come from the active provider profile's "generation"
336 # section when a harness config is loaded; explicit keyword args win.
337 # A missing model is an error; missing generation parameters are left out of
338 # the request so the server's own default applies.
339 
340 def active_model_id():
341     p = config_active_provider()
342     m = provider_model(p) if p else None
343     if not m:
344         raise RuntimeError('fireworks-ai: active provider profile has no "model"; '
345                            "set it in the harness config")
346     return m
347 
348 
349 def gen_param(key):
350     return generation_ref(provider_generation(config_active_provider()), key, None)
351 
352 
353 def gen_max_iterations():
354     """Loop budget: profile "generation": {"max_iterations": N} or the default."""
355     n = gen_param("max_iterations")
356     return n if isinstance(n, int) and n > 0 else MAX_ITERATIONS_DEFAULT
357 
358 
359 def chat(messages, model_id=None, max_tokens=None, temperature=None):
360     return chat_star(post_fireworks, messages,
361                      model_id=model_id or active_model_id(),
362                      max_tokens=max_tokens if max_tokens is not None else gen_param("max_tokens"),
363                      temperature=temperature if temperature is not None else gen_param("temperature"))
364 
365 
366 def chat_with_tools(messages, tools, model_id=None, max_tokens=None,
367                     temperature=None, max_iterations=None):
368     return chat_with_tools_star(post_fireworks, messages, tools,
369                                 model_id=model_id or active_model_id(),
370                                 max_tokens=max_tokens if max_tokens is not None else gen_param("max_tokens"),
371                                 temperature=temperature if temperature is not None else gen_param("temperature"),
372                                 max_iterations=max_iterations or gen_max_iterations())