small

harness.py

@@ -2,20 +2,21 @@ from __future__ import annotations
 
 import json
 import os
-import subprocess
 import uuid
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, cast
 
+from dotenv import load_dotenv as dotenv_load_dotenv
 from openai import OpenAI
 
+from tools_runtime import dispatch_tool, tool_schema
+
 if TYPE_CHECKING:
     from openai.types.chat import (
         ChatCompletionMessageFunctionToolCall,
         ChatCompletionMessageParam,
-        ChatCompletionToolUnionParam,
     )
 DEFAULT_PROMPT_PATH = Path("prompt.md")
 DEFAULT_SESSION_PATH = Path("session/default.jsonl")
@@ -44,21 +45,6 @@ def ensure_parent(path: Path) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
 
 
-def load_dotenv(path: Path = Path(".env")) -> None:
-    if not path.exists():
-        return
-
-    for raw_line in path.read_text(encoding="utf-8").splitlines():
-        line = raw_line.strip()
-        if not line or line.startswith("#") or "=" not in line:
-            continue
-        key, value = line.split("=", 1)
-        key = key.strip()
-        value = value.strip().strip('"').strip("'")
-        if key and key not in os.environ:
-            os.environ[key] = value
-
-
 def append_jsonl(path: Path, payload: dict[str, Any]) -> None:
     ensure_parent(path)
     with path.open("a", encoding="utf-8") as handle:
@@ -96,31 +82,8 @@ def append_message(path: Path, role: str, content: str, **extra: Any) -> None:
     append_jsonl(path, payload)
 
 
-def shell_tool_schema() -> "list[ChatCompletionToolUnionParam]":
-    return [
-        {
-            "type": "function",
-            "function": {
-                "name": "shell",
-                "description": "Run one shell command in the current workspace and return stdout, stderr, and exit code.",
-                "parameters": {
-                    "type": "object",
-                    "properties": {
-                        "command": {
-                            "type": "string",
-                            "description": "Shell command to execute.",
-                        }
-                    },
-                    "required": ["command"],
-                    "additionalProperties": False,
-                },
-            },
-        }
-    ]
-
-
 def build_client() -> OpenAI:
-    load_dotenv()
+    dotenv_load_dotenv(dotenv_path=Path(".env"), override=False)
     base_url = next(
         (os.getenv(name) for name in BASE_URL_ENV_NAMES if os.getenv(name)), None
     )
@@ -138,35 +101,6 @@ def build_client() -> OpenAI:
     return OpenAI(base_url=base_url, api_key=api_key)
 
 
-def run_shell(command: str) -> dict[str, Any]:
-    started_at = datetime.now(UTC)
-    try:
-        completed = subprocess.run(
-            command,
-            shell=True,
-            capture_output=True,
-            text=True,
-            check=False,
-        )
-        finished_at = datetime.now(UTC)
-        return {
-            "command": command,
-            "return_code": completed.returncode,
-            "stdout": completed.stdout,
-            "stderr": completed.stderr,
-            "duration_ms": int((finished_at - started_at).total_seconds() * 1000),
-        }
-    except Exception as exc:
-        finished_at = datetime.now(UTC)
-        return {
-            "command": command,
-            "return_code": None,
-            "stdout": "",
-            "stderr": str(exc),
-            "duration_ms": int((finished_at - started_at).total_seconds() * 1000),
-        }
-
-
 def load_prompt(prompt_path: Path) -> str:
     if not prompt_path.exists():
         return ""
@@ -246,7 +180,7 @@ class TuffHarness:
                 response = self.client.chat.completions.create(
                     model=self.model_name,
                     messages=messages,
-                    tools=shell_tool_schema(),
+                    tools=tool_schema(),
                     tool_choice="auto",
                 )
             except Exception as exc:
@@ -319,18 +253,14 @@ class TuffHarness:
             for tool_call in assistant_tool_calls:
                 tool_calls_seen += 1
                 function_name = tool_call.function.name
-                if function_name != "shell":
-                    payload = {
-                        "tool": function_name,
-                        "error": f"Unsupported tool: {function_name}",
-                    }
-                else:
-                    try:
-                        arguments = json.loads(tool_call.function.arguments or "{}")
-                    except json.JSONDecodeError as exc:
-                        arguments = {"command": "", "parse_error": str(exc)}
-                    command = str(arguments.get("command", ""))
-                    payload = run_shell(command)
+                payload = dispatch_tool(
+                    tool_name=function_name,
+                    arguments_json=tool_call.function.arguments or "{}",
+                    session_path=self.session_path,
+                    prompt_path=self.prompt_path,
+                    model_name=self.model_name,
+                    read_events=read_jsonl,
+                )
 
                 tool_payload = tool_result_content(payload)
                 append_jsonl(

main.py

@@ -7,6 +7,8 @@ from argparse import ArgumentParser
 from pathlib import Path
 from typing import TYPE_CHECKING, Callable
 
+from rich.panel import Panel
+from rich.prompt import Prompt
 from watchdog.events import FileSystemEvent, FileSystemEventHandler
 from watchdog.observers import Observer
 
@@ -17,6 +19,7 @@ from harness import (
     append_message,
     read_jsonl,
 )
+from transcript_view import console, print_new_results
 
 if TYPE_CHECKING:
     from watchdog.observers.api import BaseObserver
@@ -54,78 +57,18 @@ def build_cli() -> ArgumentParser:
         action="store_true",
         help="Disable interactive stdin chat loop",
     )
-    return cli
-
-
-def format_event(event: dict) -> str | None:
-    kind = event.get("kind")
-    if kind == "message":
-        role = str(event.get("role", "")).strip().lower()
-        content = str(event.get("content", "")).strip()
-        if role == "assistant":
-            if content:
-                return f"ASSISTANT: {content}"
-            tool_calls = event.get("tool_calls") or []
-            if tool_calls:
-                names = ", ".join(
-                    str(item.get("function", {}).get("name", "tool"))
-                    for item in tool_calls
-                    if isinstance(item, dict)
+    cli.add_argument(
+        "--rich",
+        choices=["yes", "no"],
+        default="yes",
+        help="Enable/disable Rich styling (default: yes)",
     )
-                return f"ASSISTANT: tool call -> {names}"
-            return None
-        if role == "system":
-            return None
-        if content:
-            return f"{role.upper()}: {content}"
-        return None
-
-    if kind == "tool_result":
-        payload = event.get("content", "")
-        try:
-            data = json.loads(str(payload))
-        except json.JSONDecodeError:
-            return f"TOOL: {payload}"
-
-        command = str(data.get("command", "")).strip()
-        code = data.get("return_code")
-        duration = data.get("duration_ms")
-        header = f"TOOL: {command}" if command else "TOOL"
-        details = []
-        if code is not None:
-            details.append(f"code={code}")
-        if duration is not None:
-            details.append(f"dur={duration}ms")
-        line = header if not details else f"{header} | {' | '.join(details)}"
-        stdout = str(data.get("stdout", "")).rstrip("\n")
-        stderr = str(data.get("stderr", "")).rstrip("\n")
-        lines = [line]
-        if stdout:
-            lines.append(f"stdout: {stdout}")
-        if stderr:
-            lines.append(f"stderr: {stderr}")
-        return "\n".join(lines)
-
-    if kind == "error":
-        error_text = str(event.get("error", "")).strip()
-        return f"ERROR: {error_text}" if error_text else "ERROR"
-
-    return None
-
-
-def print_new_results(session_path: Path, seen_ids: set[str]) -> None:
-    events = read_jsonl(session_path)
-    for event in events:
-        event_id = event.get("id")
-        if not isinstance(event_id, str) or not event_id:
-            continue
-        if event_id in seen_ids:
-            continue
-
-        line = format_event(event)
-        if line:
-            print(line)
-        seen_ids.add(event_id)
+    cli.add_argument(
+        "message",
+        nargs="*",
+        help="Optional first user message (free positional args)",
+    )
+    return cli
 
 
 class SessionHandler(FileSystemEventHandler):
@@ -147,9 +90,10 @@ def run_turn_and_print(
     session_path: Path,
     seen_ids: set[str],
     max_steps: int,
+    use_rich: bool = True,
 ) -> None:
     harness.run_turn(max_steps=max_steps)
-    print_new_results(session_path, seen_ids)
+    print_new_results(session_path, seen_ids, use_rich=use_rich)
 
 
 def chat_loop(
@@ -157,33 +101,63 @@ def chat_loop(
     session_path: Path,
     seen_ids: set[str],
     max_steps: int,
+    init_message: str | None = None,
+    use_rich: bool = True,
 ) -> None:
-    print("chat mode on. type and press enter. /exit to quit, /run to force a turn.")
-    while True:
-        try:
-            raw = input("USER: ")
-        except EOFError:
-            break
-        except KeyboardInterrupt:
-            break
+    if use_rich:
+        console.print(
+            Panel(
+                "[bold cyan]chat mode on[/]. Enter text to send. [bold]/run[/] to force a turn, [bold]/exit[/] to quit.",
+                border_style="cyan",
+                title="tuff chat",
+                subtitle="rich=yes (default)",
+            )
+        )
+    else:
+        console.print(
+            "chat mode on. Enter text to send. /run to force a turn, /exit to quit."
+        )
 
+    def ask() -> str | None:
+        try:
+            if use_rich:
+                return Prompt.ask("[bold cyan]USER[/bold cyan]")
+            else:
+                return input("USER: ")
+        except EOFError:
+            return None
+        except KeyboardInterrupt:
+            return None
+
+    raw = init_message if init_message is not None else ask()
+    while True:
+        if raw is None:
+            break
         text = raw.strip()
         if not text:
+            raw = ask()
             continue
         if text in {"/exit", "/quit"}:
             break
         if text == "/run":
-            run_turn_and_print(harness, session_path, seen_ids, max_steps)
+            run_turn_and_print(
+                harness, session_path, seen_ids, max_steps, use_rich=use_rich
+            )
+            raw = ask()
             continue
 
         append_message(session_path, "user", text)
-        run_turn_and_print(harness, session_path, seen_ids, max_steps)
+        run_turn_and_print(
+            harness, session_path, seen_ids, max_steps, use_rich=use_rich
+        )
+        raw = ask()
 
 
 def main() -> None:
     args = build_cli().parse_args()
     prompt_path = Path(args.prompt)
     session_path = Path(args.session)
+    use_rich = args.rich == "yes"
 
     if not session_path.exists():
         session_path.write_text("", encoding="utf-8")
@@ -191,18 +165,33 @@ def main() -> None:
     if not prompt_path.exists():
         prompt_path.write_text("", encoding="utf-8")
 
+    first_message = " ".join(args.message).strip() if args.message else None
+
     harness = TuffHarness(session_path=session_path, prompt_path=prompt_path)
 
     if args.run_once:
+        events = read_jsonl(session_path)
+        seen_ids = {
+            event.get("id")
+            for event in events
+            if isinstance(event.get("id"), str) and event.get("id")
+        }
+        if first_message:
+            append_message(session_path, "user", first_message)
+
         run_turn_and_print(
-            harness, session_path, seen_ids=set(), max_steps=args.max_steps
+            harness,
+            session_path,
+            seen_ids=seen_ids,
+            max_steps=args.max_steps,
+            use_rich=use_rich,
         )
         return
 
     seen_ids: set[str] = set()
 
     # Print any existing transcript first.
-    print_new_results(session_path, seen_ids)
+    print_new_results(session_path, seen_ids, use_rich=use_rich)
 
     observer: "BaseObserver | None" = None
     if args.watch:
@@ -210,24 +199,47 @@ def main() -> None:
         handler = SessionHandler(
             target_paths={prompt_path},
             on_change=lambda: run_turn_and_print(
-                harness, session_path, seen_ids, args.max_steps
+                harness, session_path, seen_ids, args.max_steps, use_rich=use_rich
             ),
         )
         observer.schedule(handler, path=str(prompt_path.parent), recursive=False)
         observer.start()
 
-        print(
+        if use_rich:
+            console.print(
+                Panel(
                     json.dumps(
                         {
                             "watching": [str(prompt_path)],
                             "model": harness.model_name,
-                }
+                        },
+                        indent=2,
+                    ),
+                    border_style="blue",
+                    title="watch mode",
+                )
+            )
+        else:
+            console.print(
+                json.dumps(
+                    {
+                        "watching": [str(prompt_path)],
+                        "model": harness.model_name,
+                    },
+                    indent=2,
                 )
             )
 
     try:
         if not args.no_chat:
-            chat_loop(harness, session_path, seen_ids, args.max_steps)
+            chat_loop(
+                harness,
+                session_path,
+                seen_ids,
+                args.max_steps,
+                init_message=first_message,
+                use_rich=use_rich,
+            )
         elif observer is not None:
             while True:
                 time.sleep(0.2)

prompt.md

@@ -1,5 +1,20 @@
-You're my bro. by blud even. I am lda.
+# Tuff Harness Prompt
 
-You are being called by the tuffest harness ever. It is a file in the cwd. You can use the shell tool to find out and act when it is relevant
+You're chatting with an AI assistant that runs inside a small, file-based harness.
 
-otherwise just converse as normal.
+How I'll respond:
+
+- Concise, direct, and technically accurate. I'll get to the point.
+- No slang, roleplay, or hype. Just useful info.
+- I'll ask only a brief clarifying question when I truly need more detail.
+- When it helps readability, I'll use Markdown like lists, tables, and code blocks.
+
+Tools I can use:
+
+- I have a shell tool for running commands when that is the clearest path.
+- I also have read-only tools for listing folders, reading files, and checking git status.
+- I can inspect the harness state itself when I need to know what tools are available or what the session looks like.
+- I'll prefer read-only tools first, and only use shell when it genuinely helps.
+- If I do run something, I'll summarize the important output for you.
+
+That's the gist. Ask anything and I'll do my best to help.

pyproject.toml

@@ -7,5 +7,7 @@ requires-python = ">=3.14"
 dependencies = [
     "openai>=2.32.0",
     "pydantic>=2.13.2",
+    "python-dotenv>=1.2.2",
+    "rich>=15.0.0",
     "watchdog>=6.0.0",
 ]

readme.md

@@ -2,15 +2,25 @@
 
 This is a tiny AI harness, not just a subprocess wrapper.
 
-`prompt.md` is the model prompt seed. `harness.py` loads the Cerebras-compatible OpenAI client, calls the model, and exposes one tool: `shell`. `main.py` is presentation only: it prints the transcript and can watch for changes.
+`prompt.md` is the model prompt seed. `harness.py` loads an OpenAI-compatible client, calls the model, and exposes five tools:
+
+- `shell` (command execution)
+- `list_dir` (read-only directory listing)
+- `read_file` (read-only file window)
+- `git_status` (read-only git status)
+- `harness_state` (self-state summary)
+
+`main.py` is presentation + CLI flow: transcript rendering, rich/plain output mode, and optional prompt watching.
 
 ## runtime
 
 - Set `CEREBRAS_BASE_URL`, `CEREBRAS_API_KEY`, and `CEREBRAS_MODEL_STRING` in `.env`.
 - Run one turn with `uv run main.py --run-once`.
+- Run one turn with a seed message: `uv run main.py --run-once "wsp"`.
 - Start interactive chat loop with `uv run main.py`.
 - Watch `prompt.md` while chatting with `uv run main.py --watch`.
 - Watch-only (no stdin chat) with `uv run main.py --watch --no-chat`.
+- Disable rich rendering with `uv run main.py --rich no`.
 
 ### chat controls
 
@@ -24,6 +34,8 @@ The session file is append-only JSONL. Each event is either a model message, a t
 
 ## shape
 
-- `harness.py`: model client, tool execution, transcript append.
-- `main.py`: formatting and file watching.
+- `tools_runtime.py`: tool schemas and tool handlers.
+- `harness.py`: model client, turn orchestration, transcript append.
+- `transcript_view.py`: transcript rendering and rich/plain output formatting.
+- `main.py`: CLI, chat loop, and watch loop wiring.
 - `prompt.md`: the behavior seed for the model.

tools_runtime.py

@@ -0,0 +1,376 @@
+from __future__ import annotations
+
+import json
+import platform
+import subprocess
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Callable
+
+if TYPE_CHECKING:
+    from openai.types.chat import ChatCompletionToolUnionParam
+
+WORKSPACE_ROOT = Path.cwd()
+MAX_LIST_ENTRIES = 200
+MAX_READ_FILE_LINES = 200
+
+
+def available_tool_names() -> list[str]:
+    return ["shell", "list_dir", "read_file", "git_status", "harness_state"]
+
+
+def tool_result_payload(
+    command: str,
+    stdout: str = "",
+    stderr: str = "",
+    return_code: int | None = 0,
+) -> dict[str, Any]:
+    return {
+        "command": command,
+        "return_code": return_code,
+        "stdout": stdout,
+        "stderr": stderr,
+        "duration_ms": None,
+    }
+
+
+def tool_schema() -> "list[ChatCompletionToolUnionParam]":
+    return [
+        {
+            "type": "function",
+            "function": {
+                "name": "shell",
+                "description": f"Run one shell command in the current workspace and return stdout, stderr, and exit code. Uses subprocess.run on {platform.system() or 'unknown platform'} under the hood",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "command": {
+                            "type": "string",
+                            "description": "Shell command to execute.",
+                        }
+                    },
+                    "required": ["command"],
+                    "additionalProperties": False,
+                },
+            },
+        },
+        {
+            "type": "function",
+            "function": {
+                "name": "list_dir",
+                "description": "List files and folders in a workspace path without changing anything.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "path": {
+                            "type": "string",
+                            "description": "Workspace-relative path to list.",
+                            "default": ".",
+                        },
+                        "recursive": {
+                            "type": "boolean",
+                            "description": "Whether to walk into subdirectories.",
+                            "default": False,
+                        },
+                        "max_entries": {
+                            "type": "integer",
+                            "description": "Maximum number of entries to return.",
+                            "default": MAX_LIST_ENTRIES,
+                        },
+                    },
+                    "additionalProperties": False,
+                },
+            },
+        },
+        {
+            "type": "function",
+            "function": {
+                "name": "read_file",
+                "description": "Read a text file from the workspace with optional line range limits.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "path": {
+                            "type": "string",
+                            "description": "Workspace-relative file path to read.",
+                        },
+                        "start_line": {
+                            "type": "integer",
+                            "description": "1-based starting line number.",
+                            "default": 1,
+                        },
+                        "end_line": {
+                            "type": "integer",
+                            "description": "1-based ending line number, inclusive.",
+                            "default": MAX_READ_FILE_LINES,
+                        },
+                    },
+                    "required": ["path"],
+                    "additionalProperties": False,
+                },
+            },
+        },
+        {
+            "type": "function",
+            "function": {
+                "name": "git_status",
+                "description": "Report git status for the workspace without making changes.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {},
+                    "additionalProperties": False,
+                },
+            },
+        },
+        {
+            "type": "function",
+            "function": {
+                "name": "harness_state",
+                "description": "Report the current harness, session, prompt, and available tool state.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {},
+                    "additionalProperties": False,
+                },
+            },
+        },
+    ]
+
+
+def workspace_path(value: str) -> Path:
+    candidate = (WORKSPACE_ROOT / value).resolve() if value else WORKSPACE_ROOT
+    if WORKSPACE_ROOT not in candidate.parents and candidate != WORKSPACE_ROOT:
+        raise ValueError(f"Path escapes workspace root: {value}")
+    return candidate
+
+
+def format_directory_listing(
+    path: Path,
+    recursive: bool,
+    max_entries: int,
+) -> dict[str, Any]:
+    if not path.exists():
+        return tool_result_payload(
+            command=f"list_dir {path}",
+            stderr="Path does not exist",
+            return_code=1,
+        )
+    if not path.is_dir():
+        return tool_result_payload(
+            command=f"list_dir {path}",
+            stderr="Path is not a directory",
+            return_code=1,
+        )
+
+    limit = max(1, min(max_entries, MAX_LIST_ENTRIES))
+    entries: list[str] = []
+    total_seen = 0
+
+    if recursive:
+        for child in path.rglob("*"):
+            total_seen += 1
+            if len(entries) >= limit:
+                break
+            kind = "dir" if child.is_dir() else "file"
+            entries.append(f"{kind}: {child.relative_to(WORKSPACE_ROOT)}")
+    else:
+        for child in sorted(path.iterdir()):
+            total_seen += 1
+            if len(entries) >= limit:
+                break
+            kind = "dir" if child.is_dir() else "file"
+            entries.append(f"{kind}: {child.relative_to(WORKSPACE_ROOT)}")
+
+    header = [
+        f"path: {path.relative_to(WORKSPACE_ROOT)}",
+        f"recursive: {recursive}",
+        f"shown: {len(entries)}",
+    ]
+    if total_seen > len(entries):
+        header.append(f"truncated: showing first {len(entries)} of {total_seen}")
+
+    body = "\n".join(entries)
+    stdout = "\n".join(header + ([""] + [body] if body else []))
+
+    return tool_result_payload(
+        command=f"list_dir {path.relative_to(WORKSPACE_ROOT)}",
+        stdout=stdout,
+    )
+
+
+def read_file_window(path: Path, start_line: int, end_line: int) -> dict[str, Any]:
+    if not path.exists():
+        return tool_result_payload(
+            command=f"read_file {path}",
+            stderr="File does not exist",
+            return_code=1,
+        )
+    if not path.is_file():
+        return tool_result_payload(
+            command=f"read_file {path}",
+            stderr="Path is not a file",
+            return_code=1,
+        )
+
+    text = path.read_text(encoding="utf-8")
+    lines = text.splitlines()
+    requested_start = max(start_line, 1)
+    requested_end = max(end_line, requested_start)
+    if requested_end - requested_start + 1 > MAX_READ_FILE_LINES:
+        requested_end = requested_start + MAX_READ_FILE_LINES - 1
+
+    start_index = requested_start - 1
+    stop_index = min(requested_end, len(lines))
+    selected = lines[start_index:stop_index]
+
+    numbered = "\n".join(
+        f"{requested_start + index:>4}: {line}" for index, line in enumerate(selected)
+    )
+    meta = [
+        f"path: {path.relative_to(WORKSPACE_ROOT)}",
+        f"range: {requested_start}-{requested_end}",
+        f"total_lines: {len(lines)}",
+        "",
+    ]
+
+    return tool_result_payload(
+        command=f"read_file {path.relative_to(WORKSPACE_ROOT)}:{requested_start}-{requested_end}",
+        stdout="\n".join(meta) + numbered,
+    )
+
+
+def git_status_report(root: Path) -> dict[str, Any]:
+    started_at = datetime.now(UTC)
+    try:
+        completed = subprocess.run(
+            ["git", "-C", str(root), "status", "--short", "--branch"],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        finished_at = datetime.now(UTC)
+        return {
+            "command": f"git -C {root} status --short --branch",
+            "return_code": completed.returncode,
+            "stdout": completed.stdout,
+            "stderr": completed.stderr,
+            "duration_ms": int((finished_at - started_at).total_seconds() * 1000),
+        }
+    except (OSError, subprocess.SubprocessError) as exc:
+        finished_at = datetime.now(UTC)
+        return {
+            "command": f"git -C {root} status --short --branch",
+            "return_code": None,
+            "stdout": "",
+            "stderr": str(exc),
+            "duration_ms": int((finished_at - started_at).total_seconds() * 1000),
+        }
+
+
+def run_shell(command: str) -> dict[str, Any]:
+    started_at = datetime.now(UTC)
+    try:
+        completed = subprocess.run(
+            command,
+            shell=True,
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        finished_at = datetime.now(UTC)
+        return {
+            "command": command,
+            "return_code": completed.returncode,
+            "stdout": completed.stdout,
+            "stderr": completed.stderr,
+            "duration_ms": int((finished_at - started_at).total_seconds() * 1000),
+        }
+    except (OSError, subprocess.SubprocessError) as exc:
+        finished_at = datetime.now(UTC)
+        return {
+            "command": command,
+            "return_code": None,
+            "stdout": "",
+            "stderr": str(exc),
+            "duration_ms": int((finished_at - started_at).total_seconds() * 1000),
+        }
+
+
+def harness_state_report(
+    session_path: Path,
+    prompt_path: Path,
+    model_name: str,
+    available_tools: list[str],
+    read_events: Callable[[Path], list[dict[str, Any]]],
+) -> dict[str, Any]:
+    events = read_events(session_path)
+    message_count = sum(1 for event in events if event.get("kind") == "message")
+    tool_result_count = sum(1 for event in events if event.get("kind") == "tool_result")
+    error_count = sum(1 for event in events if event.get("kind") == "error")
+    last_event = events[-1] if events else None
+
+    stdout = "\n".join(
+        [
+            f"session_path: {session_path}",
+            f"prompt_path: {prompt_path}",
+            f"workspace_root: {WORKSPACE_ROOT}",
+            f"model_name: {model_name}",
+            f"available_tools: {', '.join(available_tools)}",
+            f"event_counts: message={message_count}, tool_result={tool_result_count}, error={error_count}",
+            f"last_event_kind: {last_event.get('kind') if isinstance(last_event, dict) else None}",
+            f"last_event_role: {last_event.get('role') if isinstance(last_event, dict) else None}",
+        ]
+    )
+
+    return tool_result_payload(command="harness_state", stdout=stdout)
+
+
+def dispatch_tool(
+    tool_name: str,
+    arguments_json: str,
+    session_path: Path,
+    prompt_path: Path,
+    model_name: str,
+    read_events: Callable[[Path], list[dict[str, Any]]],
+) -> dict[str, Any]:
+    available_tools = available_tool_names()
+    try:
+        arguments = json.loads(arguments_json or "{}")
+    except json.JSONDecodeError as exc:
+        arguments = {"parse_error": str(exc)}
+
+    try:
+        if tool_name == "shell":
+            return run_shell(str(arguments.get("command", "")))
+
+        if tool_name == "list_dir":
+            path = workspace_path(str(arguments.get("path", ".")))
+            recursive = bool(arguments.get("recursive", False))
+            max_entries = int(arguments.get("max_entries", MAX_LIST_ENTRIES))
+            return format_directory_listing(path, recursive, max_entries)
+
+        if tool_name == "read_file":
+            path = workspace_path(str(arguments.get("path", "")))
+            start_line = int(arguments.get("start_line", 1))
+            end_line = int(arguments.get("end_line", MAX_READ_FILE_LINES))
+            return read_file_window(path, start_line, end_line)
+
+        if tool_name == "git_status":
+            return git_status_report(WORKSPACE_ROOT)
+
+        if tool_name == "harness_state":
+            return harness_state_report(
+                session_path=session_path,
+                prompt_path=prompt_path,
+                model_name=model_name,
+                available_tools=available_tools,
+                read_events=read_events,
+            )
+    except (ValueError, TypeError, OSError, subprocess.SubprocessError) as exc:
+        return tool_result_payload(command=tool_name, stderr=str(exc), return_code=1)
+
+    return tool_result_payload(
+        command=tool_name,
+        stderr=f"Unsupported tool: {tool_name}",
+        return_code=1,
+    )

transcript_view.py

@@ -0,0 +1,248 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from rich.console import Console
+from rich.markdown import Markdown
+from rich.panel import Panel
+from rich.syntax import Syntax
+from rich.text import Text
+
+from harness import read_jsonl
+
+console = Console()
+MAX_TOOL_OUTPUT_LINES = 28
+
+
+def format_event(event: dict) -> str | None:
+    kind = event.get("kind")
+    if kind == "message":
+        role = str(event.get("role", "")).strip().lower()
+        content = str(event.get("content", "")).strip()
+        if role == "assistant":
+            if content:
+                return f"ASSISTANT: {content}"
+            tool_calls = event.get("tool_calls") or []
+            if tool_calls:
+                names = ", ".join(
+                    str(item.get("function", {}).get("name", "tool"))
+                    for item in tool_calls
+                    if isinstance(item, dict)
+                )
+                return f"ASSISTANT: tool call -> {names}"
+            return None
+        if role == "system":
+            return None
+        if content:
+            return f"{role.upper()}: {content}"
+        return None
+
+    if kind == "tool_result":
+        payload = event.get("content", "")
+        try:
+            data = json.loads(str(payload))
+        except json.JSONDecodeError:
+            return f"TOOL: {payload}"
+
+        command = str(data.get("command", "")).strip()
+        code = data.get("return_code")
+        duration = data.get("duration_ms")
+        header = f"TOOL: {command}" if command else "TOOL"
+        details = []
+        if code is not None:
+            details.append(f"code={code}")
+        if duration is not None:
+            details.append(f"dur={duration}ms")
+        line = header if not details else f"{header} | {' | '.join(details)}"
+        stdout = str(data.get("stdout", "")).rstrip("\n")
+        stderr = str(data.get("stderr", "")).rstrip("\n")
+        lines = [line]
+        if stdout:
+            lines.append(f"stdout: {stdout}")
+        if stderr:
+            lines.append(f"stderr: {stderr}")
+        return "\n".join(lines)
+
+    if kind == "error":
+        error_text = str(event.get("error", "")).strip()
+        return f"ERROR: {error_text}" if error_text else "ERROR"
+
+    return None
+
+
+def truncate_lines(content: str, limit: int = MAX_TOOL_OUTPUT_LINES) -> str:
+    lines = content.splitlines()
+    if len(lines) <= limit:
+        return content
+    hidden = len(lines) - limit
+    visible = "\n".join(lines[:limit])
+    return f"{visible}\n... [{hidden} more lines hidden]"
+
+
+def pretty_json_or_text(value: str) -> tuple[str, str]:
+    text = value.rstrip("\n")
+    if not text:
+        return "text", ""
+    try:
+        parsed = json.loads(text)
+    except json.JSONDecodeError:
+        return "text", text
+    if isinstance(parsed, (dict, list)):
+        return "json", json.dumps(parsed, indent=2, ensure_ascii=True)
+    return "text", text
+
+
+def print_tool_event(
+    payload: str, tool_name: str | None = None, use_rich: bool = True
+) -> None:
+    try:
+        data = json.loads(str(payload))
+    except json.JSONDecodeError:
+        label = f"tool:{tool_name}" if tool_name else "tool"
+        if use_rich:
+            console.print(Panel(str(payload), border_style="yellow", title=label))
+        else:
+            console.print(f"TOOL[{tool_name or 'unknown'}]: {payload}")
+        return
+
+    command = str(data.get("command", "")).strip() or f"{tool_name or 'tool'}"
+    code = data.get("return_code")
+    duration = data.get("duration_ms")
+    label = f"tool:{tool_name}" if tool_name else "tool"
+
+    stdout_lang, stdout_content = pretty_json_or_text(str(data.get("stdout", "")))
+    stderr_lang, stderr_content = pretty_json_or_text(str(data.get("stderr", "")))
+
+    if use_rich:
+        header = Text()
+        if tool_name:
+            header.append("name: ", style="bold")
+            header.append(tool_name, style="cyan")
+            header.append("\n")
+        header.append("command: ", style="bold")
+        header.append(command, style="yellow")
+        if code is not None:
+            header.append("\nexit: ", style="bold")
+            header.append(str(code), style="green" if code == 0 else "red")
+        if duration is not None:
+            header.append("\nduration: ", style="bold")
+            header.append(f"{duration}ms", style="cyan")
+
+        console.print(Panel(header, border_style="yellow", title=label))
+
+        if stdout_content:
+            console.print(
+                Panel(
+                    Syntax(truncate_lines(stdout_content), stdout_lang, word_wrap=True),
+                    border_style="green",
+                    title="stdout",
+                )
+            )
+        if stderr_content:
+            console.print(
+                Panel(
+                    Syntax(truncate_lines(stderr_content), stderr_lang, word_wrap=True),
+                    border_style="red",
+                    title="stderr",
+                )
+            )
+    else:
+        details = []
+        if code is not None:
+            details.append(f"code={code}")
+        if duration is not None:
+            details.append(f"dur={duration}ms")
+        header_line = (
+            f"TOOL[{tool_name or 'unknown'}]: {command}" if command else "TOOL"
+        )
+        if details:
+            header_line = f"{header_line} | {' | '.join(details)}"
+        console.print(header_line)
+
+        if stdout_content:
+            console.print(f"stdout: {truncate_lines(stdout_content)}")
+        if stderr_content:
+            console.print(f"stderr: {truncate_lines(stderr_content)}")
+
+
+def print_new_results(
+    session_path: Path, seen_ids: set[str], use_rich: bool = True
+) -> None:
+    events = read_jsonl(session_path)
+    for event in events:
+        event_id = event.get("id")
+        if not isinstance(event_id, str) or not event_id:
+            continue
+        if event_id in seen_ids:
+            continue
+
+        if (
+            event.get("kind") == "message"
+            and str(event.get("role", "")).lower() == "assistant"
+        ):
+            content = str(event.get("content", "")).strip()
+            tool_calls = event.get("tool_calls") or []
+            if content:
+                if use_rich:
+                    console.print(
+                        Panel(
+                            Markdown(content),
+                            border_style="magenta",
+                            title="assistant",
+                        )
+                    )
+                else:
+                    console.print(f"ASSISTANT: {content}")
+                seen_ids.add(event_id)
+                continue
+            if tool_calls:
+                names = ", ".join(
+                    str(item.get("function", {}).get("name", "tool"))
+                    for item in tool_calls
+                    if isinstance(item, dict)
+                )
+                if use_rich:
+                    console.print(
+                        f"ASSISTANT: tool call -> {names}", style="bold magenta"
+                    )
+                else:
+                    console.print(f"ASSISTANT: tool call -> {names}")
+                seen_ids.add(event_id)
+                continue
+
+        if (
+            event.get("kind") == "message"
+            and str(event.get("role", "")).lower() == "user"
+        ):
+            content = str(event.get("content", "")).strip()
+            if content:
+                if use_rich:
+                    console.print(Panel(content, border_style="cyan", title="user"))
+                else:
+                    console.print(f"USER: {content}")
+                seen_ids.add(event_id)
+                continue
+
+        line = format_event(event)
+        if line:
+            if event.get("kind") == "tool_result":
+                print_tool_event(
+                    str(event.get("content", "")),
+                    tool_name=str(event.get("name", "")).strip() or None,
+                    use_rich=use_rich,
+                )
+            elif line.startswith("ERROR:"):
+                if use_rich:
+                    console.print(
+                        Panel(
+                            line.removeprefix("ERROR: ").strip(),
+                            border_style="red",
+                            title="error",
+                        )
+                    )
+                else:
+                    console.print(line)
+            else:
+                console.print(line)
+        seen_ids.add(event_id)

uv.lock

@@ -138,6 +138,8 @@ source = { virtual = "." }
 dependencies = [
     { name = "openai" },
     { name = "pydantic" },
+    { name = "python-dotenv" },
+    { name = "rich" },
     { name = "watchdog" },
 ]
 
@@ -145,9 +147,32 @@ dependencies = [
 requires-dist = [
     { name = "openai", specifier = ">=2.32.0" },
     { name = "pydantic", specifier = ">=2.13.2" },
+    { name = "python-dotenv", specifier = ">=1.2.2" },
+    { name = "rich", specifier = ">=15.0.0" },
     { name = "watchdog", specifier = ">=6.0.0" },
 ]
 
+[[package]]
+name = "markdown-it-py"
+version = "4.0.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "mdurl" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/5b/f5/4ec618ed16cc4f8fb3b701563655a69816155e79e24a17b651541804721d/markdown_it_py-4.0.0.tar.gz", hash = "sha256:cb0a2b4aa34f932c007117b194e945bd74e0ec24133ceb5bac59009cda1cb9f3", size = 73070, upload-time = "2025-08-11T12:57:52.854Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/94/54/e7d793b573f298e1c9013b8c4dade17d481164aa517d1d7148619c2cedbf/markdown_it_py-4.0.0-py3-none-any.whl", hash = "sha256:87327c59b172c5011896038353a81343b6754500a08cd7a4973bb48c6d578147", size = 87321, upload-time = "2025-08-11T12:57:51.923Z" },
+]
+
+[[package]]
+name = "mdurl"
+version = "0.1.2"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/d6/54/cfe61301667036ec958cb99bd3efefba235e65cdeb9c84d24a8293ba1d90/mdurl-0.1.2.tar.gz", hash = "sha256:bb413d29f5eea38f31dd4754dd7377d4465116fb207585f97bf925588687c1ba", size = 8729, upload-time = "2022-08-14T12:40:10.846Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" },
+]
+
 [[package]]
 name = "openai"
 version = "2.32.0"
@@ -223,6 +248,37 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/97/32/e19b83ceb07a3f1bb21798407790bbc9a31740158fd132b94139cb84e16c/pydantic_core-2.46.2-cp314-cp314t-win_arm64.whl", hash = "sha256:7dcb9d40930dfad7ab6b20bcc6ca9d2b030b0f347a0cd9909b54bd53ead521b1", size = 2016941, upload-time = "2026-04-17T09:12:34.447Z" },
 ]
 
+[[package]]
+name = "pygments"
+version = "2.20.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/c3/b2/bc9c9196916376152d655522fdcebac55e66de6603a76a02bca1b6414f6c/pygments-2.20.0.tar.gz", hash = "sha256:6757cd03768053ff99f3039c1a36d6c0aa0b263438fcab17520b30a303a82b5f", size = 4955991, upload-time = "2026-03-29T13:29:33.898Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/f4/7e/a72dd26f3b0f4f2bf1dd8923c85f7ceb43172af56d63c7383eb62b332364/pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176", size = 1231151, upload-time = "2026-03-29T13:29:30.038Z" },
+]
+
+[[package]]
+name = "python-dotenv"
+version = "1.2.2"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/82/ed/0301aeeac3e5353ef3d94b6ec08bbcabd04a72018415dcb29e588514bba8/python_dotenv-1.2.2.tar.gz", hash = "sha256:2c371a91fbd7ba082c2c1dc1f8bf89ca22564a087c2c287cd9b662adde799cf3", size = 50135, upload-time = "2026-03-01T16:00:26.196Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/0b/d7/1959b9648791274998a9c3526f6d0ec8fd2233e4d4acce81bbae76b44b2a/python_dotenv-1.2.2-py3-none-any.whl", hash = "sha256:1d8214789a24de455a8b8bd8ae6fe3c6b69a5e3d64aa8a8e5d68e694bbcb285a", size = 22101, upload-time = "2026-03-01T16:00:25.09Z" },
+]
+
+[[package]]
+name = "rich"
+version = "15.0.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "markdown-it-py" },
+    { name = "pygments" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/c0/8f/0722ca900cc807c13a6a0c696dacf35430f72e0ec571c4275d2371fca3e9/rich-15.0.0.tar.gz", hash = "sha256:edd07a4824c6b40189fb7ac9bc4c52536e9780fbbfbddf6f1e2502c31b068c36", size = 230680, upload-time = "2026-04-12T08:24:00.75Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/82/3b/64d4899d73f91ba49a8c18a8ff3f0ea8f1c1d75481760df8c68ef5235bf5/rich-15.0.0-py3-none-any.whl", hash = "sha256:33bd4ef74232fb73fe9279a257718407f169c09b78a87ad3d296f548e27de0bb", size = 310654, upload-time = "2026-04-12T08:24:02.83Z" },
+]
+
 [[package]]
 name = "sniffio"
 version = "1.3.1"