chore: add agents/reports to gitignore and update agent infrastructure with timestamp streaming, write budget, and codename generation

- Add `agents/reports/` to `.gitignore` to prevent generated agent report files from being tracked
- Implement `_TimestampStream` class and `install_timestamps()` function in `agents/agent.py` for prefixing stdout/stderr with timestamps and elapsed time
- Export `install_timestamps`, `set_write_budget`, `clear_write_budget`, and `set_shell_restricted` from `agents/base.py`; add `_RUN_LOCK`, `AGENT_ICONS` dictionary, and `WRITE_BUDGET = 20` constant
- Add `report_codename()` function and `CODENAME_ADJECTIVES`/`CODENAME_ANIMALS` tuples to `agents/core/__init__.py` for generating random agent report codenames
- Expand `WRITE_TOOLS` tuple in `agents/core/__init__.py` to include `replace_lines`, `insert_lines`, and `delete_lines`; remove `SWARM_TOOLS` from `payloads_for` exclusion list
- Update `CLAUDE.md` and `AGENTS.md` documentation with dataset column initialization rules and new agent infrastructure details
- Reorder route table in `README.md` to list `/uploads` before `/messages` and document Swagger/ReDoc/OpenAPI schema endpoints

agents/agent.py

@@ -708,6 +708,56 @@ logging.basicConfig(level=logging.WARN, format="%(asctime)s %(levelname)s %(name
 logger = logging.getLogger("x")
 
 
+class _TimestampStream:
+    def __init__(self, wrapped: Any, start: datetime) -> None:
+        self._wrapped = wrapped
+        self._start = start
+        self._need_prefix = True
+
+    def _prefix(self) -> str:
+        now = datetime.now()
+        total = int((now - self._start).total_seconds())
+        hours, rest = divmod(total, 3600)
+        minutes, seconds = divmod(rest, 60)
+        return f"\033[2m{now.strftime('%H:%M:%S')} +{hours:02d}:{minutes:02d}:{seconds:02d} │\033[0m "
+
+    def write(self, text: str) -> int:
+        if not text:
+            return 0
+        out: list[str] = []
+        for char in text:
+            if self._need_prefix:
+                out.append(self._prefix())
+                self._need_prefix = False
+            out.append(char)
+            if char == "\n":
+                self._need_prefix = True
+        self._wrapped.write("".join(out))
+        return len(text)
+
+    def flush(self) -> None:
+        self._wrapped.flush()
+
+    def isatty(self) -> bool:
+        return bool(getattr(self._wrapped, "isatty", lambda: False)())
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._wrapped, name)
+
+
+_TS_INSTALLED = False
+
+
+def install_timestamps(start: Optional[datetime] = None) -> None:
+    global _TS_INSTALLED
+    if _TS_INSTALLED:
+        return
+    moment = start or datetime.now()
+    sys.stdout = _TimestampStream(sys.stdout, moment)
+    sys.stderr = _TimestampStream(sys.stderr, moment)
+    _TS_INSTALLED = True
+
+
 class MarkdownRenderer:
     RESET = "\033[0m"
     BOLD = "\033[1m"
@@ -827,38 +877,111 @@ def _sha(text: str) -> str:
     return hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest()
 
 
-async def stream_subprocess(
-    argv: list[str],
-    timeout: Optional[int] = None,
-    prefix: str = "",
-    stdout_sink: Any = None,
-    stderr_sink: Any = None,
+def _sink_write(sink: Any, prefix: str, color: str, line: str) -> None:
+    if sink is None:
+        return
+    use_color = bool(color) and bool(getattr(sink, "isatty", lambda: False)())
+    open_color = color if use_color else ""
+    reset = MarkdownRenderer.RESET if use_color else ""
+    sink.write(f"{open_color}{prefix}{line.rstrip(chr(13) + chr(10))}{reset}\n")
+    sink.flush()
+
+
+async def _stream_pipe(
+    argv: list[str], timeout: Optional[int], prefix: str, stdout_sink: Any, stderr_sink: Any
 ) -> tuple[str, str, Optional[int], bool]:
     proc = await asyncio.create_subprocess_exec(
         *argv,
         stdin=asyncio.subprocess.DEVNULL,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.PIPE,
+        env={**os.environ, "PYTHONUNBUFFERED": "1"},
     )
     stdout_buf: list[str] = []
     stderr_buf: list[str] = []
 
     async def consume(stream: Any, buf: list[str], sink: Any, color: str) -> None:
-        use_color = bool(color) and sink is not None and sink.isatty()
-        reset = MarkdownRenderer.RESET if use_color else ""
-        open_color = color if use_color else ""
         while True:
             line = await stream.readline()
             if not line:
                 break
             text = line.decode("utf-8", errors="replace")
             buf.append(text)
-            if sink is not None:
-                sink.write(f"{open_color}{prefix}{text.rstrip(chr(10))}{reset}\n")
-                sink.flush()
+            _sink_write(sink, prefix, color, text)
 
     out_task = asyncio.create_task(consume(proc.stdout, stdout_buf, stdout_sink, ""))
     err_task = asyncio.create_task(consume(proc.stderr, stderr_buf, stderr_sink, MarkdownRenderer.RED))
+    timed_out = False
+    try:
+        returncode = await asyncio.wait_for(proc.wait(), timeout=timeout)
+    except asyncio.TimeoutError:
+        timed_out = True
+        try:
+            proc.kill()
+        except ProcessLookupError:
+            pass
+        await proc.wait()
+        returncode = proc.returncode
+    await out_task
+    await err_task
+    return _truncate_output("".join(stdout_buf)), _truncate_output("".join(stderr_buf)), returncode, timed_out
+
+
+async def _stream_pty(
+    out_master: int, out_slave: int, err_master: int, err_slave: int,
+    argv: list[str], timeout: Optional[int], prefix: str, stdout_sink: Any, stderr_sink: Any,
+) -> tuple[str, str, Optional[int], bool]:
+    import fcntl
+
+    loop = asyncio.get_event_loop()
+    for fd in (out_master, err_master):
+        flags = fcntl.fcntl(fd, fcntl.F_GETFL)
+        fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)
+    proc = await asyncio.create_subprocess_exec(
+        *argv,
+        stdin=asyncio.subprocess.DEVNULL,
+        stdout=out_slave,
+        stderr=err_slave,
+        env={**os.environ, "PYTHONUNBUFFERED": "1"},
+    )
+    os.close(out_slave)
+    os.close(err_slave)
+
+    out_buf: list[str] = []
+    err_buf: list[str] = []
+
+    def make_consumer(fd: int, buf: list[str], sink: Any, color: str) -> "asyncio.Future[None]":
+        partial = {"text": ""}
+        done: "asyncio.Future[None]" = loop.create_future()
+
+        def on_read() -> None:
+            try:
+                chunk = os.read(fd, 8192)
+            except (BlockingIOError, InterruptedError):
+                return
+            except OSError:
+                chunk = b""
+            if not chunk:
+                loop.remove_reader(fd)
+                if partial["text"]:
+                    buf.append(partial["text"])
+                    _sink_write(sink, prefix, color, partial["text"])
+                    partial["text"] = ""
+                if not done.done():
+                    done.set_result(None)
+                return
+            data = partial["text"] + chunk.decode("utf-8", errors="replace")
+            pieces = data.split("\n")
+            partial["text"] = pieces.pop()
+            for piece in pieces:
+                buf.append(piece.rstrip("\r") + "\n")
+                _sink_write(sink, prefix, color, piece)
+
+        loop.add_reader(fd, on_read)
+        return done
+
+    out_done = make_consumer(out_master, out_buf, stdout_sink, "")
+    err_done = make_consumer(err_master, err_buf, stderr_sink, MarkdownRenderer.RED)
 
     timed_out = False
     try:
@@ -871,15 +994,42 @@ async def stream_subprocess(
             pass
         await proc.wait()
         returncode = proc.returncode
+    try:
+        await asyncio.wait_for(asyncio.gather(out_done, err_done), timeout=5)
+    except asyncio.TimeoutError:
+        pass
+    for fd in (out_master, err_master):
+        try:
+            loop.remove_reader(fd)
+        except (ValueError, OSError):
+            pass
+        try:
+            os.close(fd)
+        except OSError:
+            pass
+    return _truncate_output("".join(out_buf)), _truncate_output("".join(err_buf)), returncode, timed_out
 
-    await out_task
-    await err_task
-    return (
-        _truncate_output("".join(stdout_buf)),
-        _truncate_output("".join(stderr_buf)),
-        returncode,
-        timed_out,
-    )
+
+async def stream_subprocess(
+    argv: list[str],
+    timeout: Optional[int] = None,
+    prefix: str = "",
+    stdout_sink: Any = None,
+    stderr_sink: Any = None,
+) -> tuple[str, str, Optional[int], bool]:
+    if os.name == "posix":
+        try:
+            import pty
+
+            out_master, out_slave = pty.openpty()
+            err_master, err_slave = pty.openpty()
+        except Exception:  # noqa: BLE001
+            return await _stream_pipe(argv, timeout, prefix, stdout_sink, stderr_sink)
+        return await _stream_pty(
+            out_master, out_slave, err_master, err_slave,
+            argv, timeout, prefix, stdout_sink, stderr_sink,
+        )
+    return await _stream_pipe(argv, timeout, prefix, stdout_sink, stderr_sink)
 
 
 _http_client: Optional[ChromeStealthClient] = None
@@ -1257,6 +1407,7 @@ class SwarmProcess:
 
 _swarm: dict[int, SwarmProcess] = {}
 _agent_state: contextvars.ContextVar[Optional[AgentState]] = contextvars.ContextVar("agent_state", default=None)
+_active_renderer: contextvars.ContextVar[Any] = contextvars.ContextVar("active_renderer", default=None)
 
 
 def _state() -> Optional[AgentState]:
@@ -1270,6 +1421,7 @@ def _record_read(path: Path, content: str) -> None:
 
 
 def _record_modification(path: Path, content: str) -> None:
+    _WRITTEN.add(str(path.resolve()))
     state = _state()
     if state is not None:
         key = str(path.resolve())
@@ -1277,8 +1429,72 @@ def _record_modification(path: Path, content: str) -> None:
         state.read_files[key] = _sha(content)
 
 
+_DIFF_STREAM = True
+
+
+def set_diff_stream(enabled: bool) -> None:
+    global _DIFF_STREAM
+    _DIFF_STREAM = enabled
+
+
+def _emit_diff(path: Any, old_text: str, new_text: str, max_lines: int = 160) -> None:
+    if not _DIFF_STREAM or old_text == new_text:
+        return
+    import difflib
+
+    rel = str(path)
+    verb = "create" if old_text == "" else "edit"
+    diff = difflib.unified_diff(
+        old_text.splitlines(),
+        new_text.splitlines(),
+        fromfile=f"a/{rel}",
+        tofile=f"b/{rel}",
+        lineterm="",
+        n=3,
+    )
+    rendered: list[str] = [f"\033[2m✎ diff ({verb}) {rel}\033[0m"]
+    for line in diff:
+        if len(rendered) > max_lines:
+            rendered.append(f"\033[2m… [diff truncated at {max_lines} lines]\033[0m")
+            break
+        if line.startswith("+++") or line.startswith("---"):
+            continue
+        if line.startswith("+"):
+            rendered.append(f"\033[32m{line}\033[0m")
+        elif line.startswith("-"):
+            rendered.append(f"\033[31m{line}\033[0m")
+        elif line.startswith("@@"):
+            rendered.append(f"\033[36m{line}\033[0m")
+        else:
+            rendered.append(f"\033[2m{line}\033[0m")
+    sys.stderr.write("\n".join(rendered) + "\n")
+    sys.stderr.flush()
+
+
+def _python_break(path: Any, old_text: str, new_text: str) -> Optional[str]:
+    if Path(path).suffix.lower() != ".py":
+        return None
+    try:
+        ast.parse(new_text)
+        return None
+    except SyntaxError as exc:
+        try:
+            ast.parse(old_text)
+        except SyntaxError:
+            return None
+        return (
+            f"Edit rejected: it would introduce a Python syntax error ({exc.msg}, line {exc.lineno}). "
+            "The file was NOT changed. Almost always this is an indentation mistake in your replacement. "
+            "Re-read the surrounding lines with read_lines and retry; for precise edits of large files use "
+            "replace_lines/insert_lines so the indentation you see is the indentation you write."
+        )
+
+
 AGENTS_DIR = Path(__file__).resolve().parent
 _PROTECTED_TREES: set[str] = set()
+_RESTRICT_SHELL = False
+_WRITE_BUDGET: Optional[int] = None
+_WRITTEN: set[str] = set()
 
 
 def protect_tree(path: Path) -> None:
@@ -1293,6 +1509,23 @@ def clear_protected_trees() -> None:
     _PROTECTED_TREES.clear()
 
 
+def set_shell_restricted(restricted: bool) -> None:
+    global _RESTRICT_SHELL
+    _RESTRICT_SHELL = restricted
+
+
+def set_write_budget(limit: Optional[int]) -> None:
+    global _WRITE_BUDGET
+    _WRITE_BUDGET = limit
+    _WRITTEN.clear()
+
+
+def clear_write_budget() -> None:
+    global _WRITE_BUDGET
+    _WRITE_BUDGET = None
+    _WRITTEN.clear()
+
+
 def _protected_guard(path: Path) -> Optional[str]:
     if not _PROTECTED_TREES:
         return None
@@ -1306,8 +1539,23 @@ def _protected_guard(path: Path) -> Optional[str]:
     return None
 
 
+def _budget_guard(path: Path) -> Optional[str]:
+    if _WRITE_BUDGET is None:
+        return None
+    resolved = str(Path(path).resolve())
+    if resolved in _WRITTEN:
+        return None
+    if len(_WRITTEN) >= _WRITE_BUDGET:
+        return (
+            f"Write budget reached: this run has already modified {_WRITE_BUDGET} distinct files, the per-run cap. "
+            "Stop editing and report the remaining issues as findings (fixed=false) instead of mass-rewriting. "
+            "This cap exists to prevent runaway sweeps across the codebase."
+        )
+    return None
+
+
 def _mutation_guard(path: Path) -> Optional[str]:
-    blocked = _protected_guard(path)
+    blocked = _protected_guard(path) or _budget_guard(path)
     if blocked:
         return blocked
     if not path.exists():
@@ -1384,14 +1632,18 @@ async def create_file(path: str, content: str):
     """
     def _do() -> str:
         p = Path(path)
-        blocked = _protected_guard(p)
+        blocked = _protected_guard(p) or _budget_guard(p)
         if blocked:
             return json.dumps({"status": "error", "error": blocked})
         if p.exists():
             return json.dumps({"status": "error", "error": "File already exists; use edit_file or write_file"})
+        syntax = _python_break(p, "", content)
+        if syntax:
+            return json.dumps({"status": "error", "error": syntax})
         p.parent.mkdir(parents=True, exist_ok=True)
         p.write_text(content, encoding="utf-8")
         _record_modification(p, content)
+        _emit_diff(p, "", content)
         return json.dumps({"status": "success", "path": str(p), "bytes": len(content.encode("utf-8"))})
     return await asyncio.to_thread(_do)
 
@@ -1407,18 +1659,63 @@ async def write_file(path: str, content: str):
         guard = _mutation_guard(p)
         if guard:
             return json.dumps({"status": "error", "error": guard})
+        old = p.read_text(encoding="utf-8", errors="replace") if p.exists() else ""
+        syntax = _python_break(p, old, content)
+        if syntax:
+            return json.dumps({"status": "error", "error": syntax})
         p.parent.mkdir(parents=True, exist_ok=True)
         p.write_text(content, encoding="utf-8")
         _record_modification(p, content)
+        _emit_diff(p, old, content)
         return json.dumps({"status": "success", "path": str(p), "bytes": len(content.encode("utf-8"))})
     return await asyncio.to_thread(_do)
 
 
+def _fuzzy_replace(src: str, old: str, new: str, replace_all: bool) -> tuple[Optional[str], int, Optional[str]]:
+    src_lines = src.split("\n")
+    old_lines = old.strip("\n").split("\n")
+    old_stripped = [line.strip() for line in old_lines]
+    span = len(old_stripped)
+    if span == 0 or all(line == "" for line in old_stripped):
+        return None, 0, "old_string is empty"
+    hits = [
+        i
+        for i in range(len(src_lines) - span + 1)
+        if all(src_lines[i + j].strip() == old_stripped[j] for j in range(span))
+    ]
+    if not hits:
+        return None, 0, (
+            "old_string not found, even ignoring whitespace. Re-read the exact lines with read_lines, then use "
+            "replace_lines(path, start, end, content) to edit by line number instead."
+        )
+    if len(hits) > 1 and not replace_all:
+        return None, 0, (
+            f"old_string matches {len(hits)} places (whitespace-tolerant). Add more surrounding lines, set "
+            "replace_all=true, or use replace_lines by line number."
+        )
+    new_lines = new.split("\n")
+    first_new = next((line for line in new_lines if line.strip()), "")
+    new_indent = first_new[: len(first_new) - len(first_new.lstrip())]
+    targets = hits if replace_all else hits[:1]
+    for i in sorted(targets, reverse=True):
+        matched = src_lines[i]
+        target_indent = matched[: len(matched) - len(matched.lstrip())]
+        if target_indent != new_indent:
+            block = [
+                target_indent + line[len(new_indent):] if line.startswith(new_indent) else line
+                for line in new_lines
+            ]
+        else:
+            block = new_lines
+        src_lines[i:i + span] = block
+    return "\n".join(src_lines), len(targets), None
+
+
 @tool
 async def edit_file(path: str, old_string: str, new_string: str, replace_all: bool = False):
-    """Replace exact text in an existing file. old_string must match uniquely unless replace_all is true. Read the file first.
+    """Replace text in an existing file. Tries an exact match first, then falls back to whitespace-tolerant line matching, so minor indentation differences still apply. Read the file first.
     path: File path.
-    old_string: Exact text to replace.
+    old_string: Text to replace (exact, or matching apart from leading/trailing whitespace per line).
     new_string: Replacement text.
     replace_all: Replace every occurrence when true.
     """
@@ -1431,20 +1728,103 @@ async def edit_file(path: str, old_string: str, new_string: str, replace_all: bo
             return json.dumps({"status": "error", "error": guard})
         src = p.read_text(encoding="utf-8")
         count = src.count(old_string)
-        if count == 0:
-            return json.dumps({"status": "error", "error": "old_string not found in file"})
-        if not replace_all and count > 1:
+        if count == 1 or (count > 1 and replace_all):
+            updated = src.replace(old_string, new_string) if replace_all else src.replace(old_string, new_string, 1)
+            replacements = count if replace_all else 1
+            method = "exact"
+        elif count > 1:
             return json.dumps({
                 "status": "error",
-                "error": f"old_string is not unique ({count} matches); set replace_all=true or add surrounding context",
+                "error": f"old_string is not unique ({count} exact matches); set replace_all=true or add surrounding context",
             })
-        updated = src.replace(old_string, new_string) if replace_all else src.replace(old_string, new_string, 1)
+        else:
+            updated, replacements, err = _fuzzy_replace(src, old_string, new_string, replace_all)
+            if err is not None or updated is None:
+                return json.dumps({"status": "error", "error": err or "Edit failed"})
+            method = "whitespace-tolerant"
+        syntax = _python_break(p, src, updated)
+        if syntax:
+            return json.dumps({"status": "error", "error": syntax})
         p.write_text(updated, encoding="utf-8")
         _record_modification(p, updated)
-        return json.dumps({"status": "success", "path": str(p), "replacements": count if replace_all else 1})
+        _emit_diff(p, src, updated)
+        return json.dumps({"status": "success", "path": str(p), "replacements": replacements, "match": method})
     return await asyncio.to_thread(_do)
 
 
+def _edit_by_lines(path: str, mutate: Callable[[list[str]], tuple[Optional[str], Optional[dict[str, Any]]]]) -> str:
+    p = Path(path)
+    if not p.exists():
+        return json.dumps({"status": "error", "error": "File not found"})
+    guard = _mutation_guard(p)
+    if guard:
+        return json.dumps({"status": "error", "error": guard})
+    src = p.read_text(encoding="utf-8")
+    lines = src.split("\n")
+    err, info = mutate(lines)
+    if err is not None:
+        return json.dumps({"status": "error", "error": err})
+    updated = "\n".join(lines)
+    syntax = _python_break(p, src, updated)
+    if syntax:
+        return json.dumps({"status": "error", "error": syntax})
+    p.write_text(updated, encoding="utf-8")
+    _record_modification(p, updated)
+    _emit_diff(p, src, updated)
+    return json.dumps({"status": "success", "path": str(p), **(info or {})})
+
+
+@tool
+async def replace_lines(path: str, start: int, end: int, content: str):
+    """Replace an inclusive 1-indexed line range with new content. The robust way to edit large files: read_lines the range first, then replace it by number (no string matching).
+    path: File path.
+    start: First line to replace, 1-indexed.
+    end: Last line to replace, inclusive.
+    content: Replacement text; may be empty to delete, may span multiple lines.
+    """
+    def mutate(lines: list[str]) -> tuple[Optional[str], Optional[dict[str, Any]]]:
+        s, e = int(start), int(end)
+        if s < 1 or s > len(lines) or e < s:
+            return f"Invalid range [{start}, {end}] for a file with {len(lines)} lines", None
+        e = min(e, len(lines))
+        lines[s - 1:e] = content.split("\n")
+        return None, {"replaced_lines": e - s + 1}
+    return await asyncio.to_thread(_edit_by_lines, path, mutate)
+
+
+@tool
+async def insert_lines(path: str, line: int, content: str):
+    """Insert content before the given 1-indexed line. Use a line number beyond the file length (or 0) to append. Read the file first.
+    path: File path.
+    line: Insert before this 1-indexed line; 0 or past the end appends.
+    content: Text to insert; may span multiple lines.
+    """
+    def mutate(lines: list[str]) -> tuple[Optional[str], Optional[dict[str, Any]]]:
+        at = int(line)
+        idx = len(lines) if at <= 0 or at > len(lines) else at - 1
+        new_lines = content.split("\n")
+        lines[idx:idx] = new_lines
+        return None, {"inserted_lines": len(new_lines), "at": idx + 1}
+    return await asyncio.to_thread(_edit_by_lines, path, mutate)
+
+
+@tool
+async def delete_lines(path: str, start: int, end: int):
+    """Delete an inclusive 1-indexed line range. Read the range with read_lines first.
+    path: File path.
+    start: First line to delete, 1-indexed.
+    end: Last line to delete, inclusive.
+    """
+    def mutate(lines: list[str]) -> tuple[Optional[str], Optional[dict[str, Any]]]:
+        s, e = int(start), int(end)
+        if s < 1 or s > len(lines) or e < s:
+            return f"Invalid range [{start}, {end}] for a file with {len(lines)} lines", None
+        e = min(e, len(lines))
+        del lines[s - 1:e]
+        return None, {"deleted_lines": e - s + 1}
+    return await asyncio.to_thread(_edit_by_lines, path, mutate)
+
+
 def _apply_unified_diff(source: str, patch: str) -> tuple[Optional[str], Optional[str]]:
     lines = patch.splitlines()
     hunks: list[tuple[list[str], list[str]]] = []
@@ -1511,8 +1891,12 @@ async def patch_file(path: str, patch: str):
         updated, err = _apply_unified_diff(source, patch)
         if err is not None or updated is None:
             return json.dumps({"status": "error", "error": err or "Patch failed"})
+        syntax = _python_break(p, source, updated)
+        if syntax:
+            return json.dumps({"status": "error", "error": syntax})
         p.write_text(updated, encoding="utf-8")
         _record_modification(p, updated)
+        _emit_diff(p, source, updated)
         return json.dumps({"status": "success", "path": str(p), "bytes": len(updated.encode("utf-8"))})
     return await asyncio.to_thread(_do)
 
@@ -1637,6 +2021,15 @@ async def run_command(command: str, timeout: Optional[int] = None):
     command: Shell command line.
     timeout: Timeout in seconds.
     """
+    if _RESTRICT_SHELL:
+        return json.dumps({
+            "status": "error",
+            "error": (
+                "run_command is disabled for maintenance agents. Use grep, glob_files, list_dir, and read_lines to "
+                "detect, and edit_file/write_file/create_file to change individual files. Never shell out to scan, "
+                "lint, or mass-edit. Use the verify tool to validate."
+            ),
+        })
     effective = int(timeout) if timeout is not None else DEFAULT_COMMAND_TIMEOUT
     try:
         stdout, stderr, exit_code, timed_out = await stream_subprocess(
@@ -2144,14 +2537,15 @@ def context_size(messages: list[dict[str, Any]]) -> int:
 
 
 def find_compaction_split(messages: list[dict[str, Any]], target_keep: int) -> int:
-    if len(messages) <= target_keep:
+    count = len(messages)
+    if count <= target_keep + 1:
         return 1
-    candidate = len(messages) - target_keep
-    while candidate > 1:
-        if messages[candidate].get("role") == "user":
-            return candidate
-        candidate -= 1
-    return 1
+    candidate = max(1, count - target_keep)
+    while candidate < count and messages[candidate].get("role") == "tool":
+        candidate += 1
+    if candidate >= count:
+        return 1
+    return candidate
 
 
 async def compact_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
@@ -2240,7 +2634,8 @@ async def react_loop(
     prefix: str = "",
 ) -> Optional[str]:
     token = _agent_state.set(state)
-    md = renderer
+    md = renderer if renderer is not None else _active_renderer.get()
+    renderer_token = _active_renderer.set(md)
     final_content: Optional[str] = None
     tool_names = {t["function"]["name"] for t in tools_payload}
     plan_required = "plan" in tool_names
@@ -2326,6 +2721,7 @@ async def react_loop(
         return final_content
     finally:
         _agent_state.reset(token)
+        _active_renderer.reset(renderer_token)
 
 
 SYSTEM_PROMPT = """You are X, an autonomous software engineer running an asynchronous, parallel agent loop with structured planning, automatic reflection on errors, and a verification gate. You build, modify, debug, research, and verify software end to end.
@@ -2412,6 +2808,7 @@ async def interactive(renderer: MarkdownRenderer, max_iterations: int) -> None:
 
 async def amain() -> int:
     global MODEL
+    install_timestamps()
     parser = argparse.ArgumentParser(description="X — single-file autonomous software engineering agent")
     parser.add_argument("prompt_pos", nargs="?", help="Task prompt (positional). With a prompt the agent runs it and exits; without one it starts interactive chat mode.")
     parser.add_argument("-p", "--prompt", dest="prompt", help="Task prompt; same as the positional argument.")

agents/base.py

@@ -14,16 +14,36 @@ from .agent import (
     AgentState,
     MarkdownRenderer,
     clear_protected_trees,
+    clear_write_budget,
     close_http_client,
     cost_session_total,
     format_usd,
+    install_timestamps,
     protect_agents,
     react_loop,
+    set_shell_restricted,
+    set_write_budget,
     usd_str,
     _with_datetime,
 )
 
 DEFAULT_MAX_ITER = 60
+WRITE_BUDGET = 20
+
+_RUN_LOCK = asyncio.Lock()
+
+AGENT_ICONS = {
+    "security": "\U0001F6E1",
+    "audit": "\U0001F4CB",
+    "devii": "\U0001F6E0",
+    "docs": "\U0001F4DA",
+    "fanout": "\U0001F310",
+    "dry": "♻",
+    "style": "\U0001F3A8",
+    "frontend": "\U0001F5A5",
+    "seo": "\U0001F50D",
+    "test": "\U0001F9EA",
+}
 
 MAINT_HEADER = """You are {name}, an autonomous maintenance agent for the DevPlace codebase, a FastAPI plus Jinja2 platform using the dataset library over SQLite, with pure ES6 module JavaScript on the frontend. You enforce exactly ONE quality dimension across the repository and nothing else.
 
@@ -47,7 +67,7 @@ DIMENSION MANDATE
 
 CHECK_RULES = "MODE: CHECK (read-only). You MUST NOT modify any file; the writing tools are not available to you. Investigate and record every issue with report_finding (fixed=false). End with a one-line summary."
 
-FIX_RULES = "MODE: FIX (autonomous). For every issue, record it with report_finding (set fixed=true once corrected) AND apply a minimal, idiomatic fix. You MUST read_file an existing file before editing it; prefer edit_file for surgical changes. After all edits, call verify('python -m agents.validator .') and ensure it passes. End with a one-line summary."
+FIX_RULES = "MODE: FIX (autonomous). For every issue, record it with report_finding (set fixed=true once corrected) AND apply a minimal, idiomatic fix. You MUST read an existing file (read_file, or read_lines for a range) before editing it. Prefer edit_file for surgical changes; for LARGE files, read_lines the target range and edit by line number with replace_lines/insert_lines/delete_lines instead of pasting big strings - it is far more reliable. After all edits, call verify('python -m agents.validator .') and ensure it passes. End with a one-line summary."
 
 
 class MaintenanceAgent:
@@ -100,11 +120,44 @@ class MaintenanceAgent:
         max_iter: int,
         renderer: Optional[MarkdownRenderer],
         seed_findings: Optional[list[dict]] = None,
+    ) -> dict:
+        async with _RUN_LOCK:
+            return await self._run_locked(mode, scope, max_iter, renderer, seed_findings)
+
+    async def _run_locked(
+        self,
+        mode: str,
+        scope: Optional[str],
+        max_iter: int,
+        renderer: Optional[MarkdownRenderer],
+        seed_findings: Optional[list[dict]],
     ) -> dict:
         core.reset_findings()
         protect_agents()
+        set_shell_restricted(True)
+        budget = WRITE_BUDGET
+        if seed_findings:
+            distinct = {f.get("file") for f in seed_findings if f.get("file")}
+            budget = max(WRITE_BUDGET, len(distinct) + 2)
+        set_write_budget(budget)
         cost_before = cost_session_total()["cost"]
         started = datetime.now()
+        codename = core.report_codename()
+        if renderer is not None:
+            icon = AGENT_ICONS.get(self.name, "\U0001F916")
+            scope_note = f" (scope: {scope})" if scope else ""
+            if mode == "fix" and seed_findings:
+                plan_line = f"fix {len(seed_findings)} confirmed finding(s) from the last check, then verify the build"
+            elif mode == "fix":
+                plan_line = f"scan{scope_note}, fix each issue, and verify the build (up to {budget} files this run)"
+            else:
+                plan_line = f"scan{scope_note} read-only and report each issue (no files changed)"
+            report_file = core.report_path(self.name, codename, started)
+            renderer.print(
+                f"\n{icon} **{self.name} agent** `{codename}` starting -- {self.description}.\n"
+                f"I will {plan_line}.\n"
+                f"Report -> `{report_file}.json`"
+            )
         messages = [
             {"role": "system", "content": _with_datetime(self.system_prompt(mode))},
             {"role": "user", "content": self.task_prompt(mode, scope, seed_findings)},
@@ -120,12 +173,16 @@ class MaintenanceAgent:
             )
         finally:
             clear_protected_trees()
+            clear_write_budget()
+            set_shell_restricted(False)
         finished = datetime.now()
         incomplete = final is None or state.iteration >= max_iter
         session_cost = cost_session_total()["cost"]
         run_cost = session_cost - cost_before
         cost = {"run_usd": usd_str(run_cost), "session_usd": usd_str(session_cost)}
-        report = core.write_reports(self.name, mode, started, finished, cost=cost, incomplete=incomplete)
+        report = core.write_reports(
+            self.name, mode, started, finished, cost=cost, incomplete=incomplete, codename=codename
+        )
         summary = report["summary"]
         if incomplete:
             exit_code = 1
@@ -168,6 +225,7 @@ def build_parser(name: str, description: str) -> argparse.ArgumentParser:
 
 
 async def _amain(agent: MaintenanceAgent, argv: Optional[list[str]] = None) -> int:
+    install_timestamps()
     args = build_parser(agent.name, agent.description).parse_args(argv)
     if args.verbose:
         logging.getLogger().setLevel(logging.DEBUG)

agents/core/__init__.py

@@ -3,10 +3,27 @@
 from __future__ import annotations
 
 import json
+import random
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Optional
 
+CODENAME_ADJECTIVES = (
+    "brave", "calm", "clever", "swift", "gentle", "bright", "bold", "cosmic", "lucky", "mighty",
+    "noble", "quiet", "rapid", "sunny", "witty", "eager", "jolly", "keen", "merry", "proud",
+    "sleek", "spry", "vivid", "zesty", "amber", "azure", "coral", "fuzzy", "golden", "happy",
+    "snappy", "cozy", "breezy", "plucky", "dapper", "nimble", "quirky", "steady", "tidy", "wise",
+)
+CODENAME_ANIMALS = (
+    "otter", "badger", "panda", "koala", "heron", "tiger", "gecko", "raven", "moose", "bison",
+    "crane", "dingo", "ferret", "marmot", "walrus", "puffin", "ibis", "lemur", "tapir", "quokka",
+    "narwhal", "ocelot", "wombat", "gibbon", "meerkat", "mongoose", "capybara", "axolotl", "pangolin", "armadillo",
+)
+
+
+def report_codename() -> str:
+    return f"{random.choice(CODENAME_ADJECTIVES)}-{random.choice(CODENAME_ANIMALS)}"
+
 from ..agent import (
     tool,
     react_loop,
@@ -20,7 +37,15 @@ from ..agent import (
 
 REPORTS_DIR = Path(__file__).resolve().parent.parent / "reports"
 
-WRITE_TOOLS = ("create_file", "write_file", "edit_file", "patch_file")
+WRITE_TOOLS = (
+    "create_file",
+    "write_file",
+    "edit_file",
+    "patch_file",
+    "replace_lines",
+    "insert_lines",
+    "delete_lines",
+)
 WEB_TOOLS = (
     "web_search",
     "deep_search",
@@ -89,7 +114,7 @@ async def report_finding(
 
 
 def payloads_for(mode: str) -> list[dict[str, Any]]:
-    exclude = WEB_TOOLS + SWARM_TOOLS
+    exclude = WEB_TOOLS + SWARM_TOOLS + ("run_command",)
     if mode == "check":
         exclude = exclude + WRITE_TOOLS + ("verify",)
     return get_tool_payloads(exclude=exclude)
@@ -149,6 +174,28 @@ def _md_report(name: str, mode: str, summary: dict[str, int], items: list[dict[s
     return "\n".join(lines)
 
 
+REPORTS_KEEP = 25
+
+
+def report_path(name: str, codename: str, started: datetime) -> Path:
+    stamp = started.strftime("%Y%m%d-%H%M%S")
+    return REPORTS_DIR / f"{name}-{codename}-{stamp}"
+
+
+def prune_reports(prefix: str, keep: int = REPORTS_KEEP) -> None:
+    existing = sorted(
+        REPORTS_DIR.glob(f"{prefix}-*.json"),
+        key=lambda path: path.stat().st_mtime,
+        reverse=True,
+    )
+    for stale in existing[keep:]:
+        for companion in (stale, stale.with_suffix(".md")):
+            try:
+                companion.unlink()
+            except OSError:
+                pass
+
+
 def write_reports(
     name: str,
     mode: str,
@@ -156,14 +203,17 @@ def write_reports(
     finished: datetime,
     cost: Optional[dict[str, Any]] = None,
     incomplete: bool = False,
+    codename: Optional[str] = None,
 ) -> dict[str, Any]:
     REPORTS_DIR.mkdir(parents=True, exist_ok=True)
     items = findings()
     summary = summarize(items)
+    codename = codename or report_codename()
     stamp = started.strftime("%Y%m%d-%H%M%S")
-    base = REPORTS_DIR / f"{name}-{stamp}"
+    base = REPORTS_DIR / f"{name}-{codename}-{stamp}"
     payload = {
         "agent": name,
+        "codename": codename,
         "mode": mode,
         "started_at": started.isoformat(),
         "finished_at": finished.isoformat(),
@@ -174,6 +224,7 @@ def write_reports(
     }
     base.with_suffix(".json").write_text(json.dumps(payload, indent=2), encoding="utf-8")
     base.with_suffix(".md").write_text(_md_report(name, mode, summary, items), encoding="utf-8")
+    prune_reports(name)
     return {"summary": summary, "json": str(base.with_suffix(".json")), "items": items}
 
 
@@ -191,5 +242,8 @@ __all__ = [
     "payloads_named",
     "summarize",
     "write_reports",
+    "report_codename",
+    "report_path",
+    "prune_reports",
     "REPORTS_DIR",
 ]

agents/maestro.py

@@ -15,6 +15,7 @@ from .agent import (
     MarkdownRenderer,
     close_http_client,
     get_tool,
+    install_timestamps,
     react_loop,
     tool,
     _with_datetime,
@@ -41,10 +42,6 @@ async def _dispatch(name: str, mode: str, scope: str) -> str:
     run_mode = "fix" if str(mode).lower() == "fix" else "check"
     renderer = _RENDERER
     seed = _LAST_CHECK.get(name) if run_mode == "fix" else None
-    if renderer is not None:
-        scope_note = f" scope={scope}" if scope else ""
-        seed_note = f" (fixing {len(seed)} confirmed findings from the last check)" if seed else ""
-        renderer.print(f"\n--- running **{name}** [{run_mode}]{scope_note}{seed_note} ---")
     agent = REGISTRY[name]()
     result = await agent.run(run_mode, scope or None, MAESTRO_AGENT_MAX_ITER, renderer=renderer, seed_findings=seed)
     _LAST_RESULTS[name] = result
@@ -243,6 +240,7 @@ async def interactive(renderer: MarkdownRenderer) -> None:
 
 
 async def _amain(argv: Optional[list[str]] = None) -> int:
+    install_timestamps()
     parser = argparse.ArgumentParser(prog="agents.maestro", description="Maestro, the conversational maintenance conductor")
     parser.add_argument("prompt", nargs="?", help="One-shot request; omit for interactive mode")
     parser.add_argument("--no-color", action="store_true")

agents/orchestrator.py

@@ -11,7 +11,7 @@ from datetime import datetime
 from typing import Optional
 
 from . import core
-from .agent import MarkdownRenderer, close_http_client, cost_session_total, format_usd, usd_str
+from .agent import MarkdownRenderer, close_http_client, cost_session_total, format_usd, install_timestamps, usd_str
 from .fleet import REGISTRY, ordered_agents
 
 
@@ -56,9 +56,11 @@ async def run_fleet(mode: str, only: Optional[str], max_iter: int, renderer: Opt
 
     session_cost = cost_session_total()["cost"]
     core.REPORTS_DIR.mkdir(parents=True, exist_ok=True)
+    codename = core.report_codename()
     stamp = started.strftime("%Y%m%d-%H%M%S")
     fleet_payload = {
         "agent": "fleet",
+        "codename": codename,
         "mode": mode,
         "started_at": started.isoformat(),
         "finished_at": finished.isoformat(),
@@ -66,8 +68,9 @@ async def run_fleet(mode: str, only: Optional[str], max_iter: int, renderer: Opt
         "cost": {"session_usd": usd_str(session_cost)},
         "agents": results,
     }
-    fleet_json = core.REPORTS_DIR / f"fleet-{stamp}.json"
+    fleet_json = core.REPORTS_DIR / f"fleet-{codename}-{stamp}.json"
     fleet_json.write_text(json.dumps(fleet_payload, indent=2), encoding="utf-8")
+    core.prune_reports("fleet")
 
     if renderer is not None:
         rows = "\n".join(
@@ -85,6 +88,7 @@ async def run_fleet(mode: str, only: Optional[str], max_iter: int, renderer: Opt
 
 
 async def _amain(argv: Optional[list[str]] = None) -> int:
+    install_timestamps()
     args = _parser().parse_args(argv)
     if args.verbose:
         logging.getLogger().setLevel(logging.DEBUG)