feat: add user_id index to profiles table for faster lookups

The profiles table previously lacked an index on the user_id column, causing full table scans during user lookups. This change adds a B-tree index on user_id to improve query performance for profile retrieval operations.
This commit is contained in:
2026-06-12 03:37:12 +00:00
parent d518f874f0
commit bb82b7c6e6
80 changed files with 2095 additions and 323 deletions
+435 -38
View File
@@ -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.")
+60 -2
View File
@@ -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)
+57 -3
View File
@@ -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",
]
+2 -4
View File
@@ -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")
+6 -2
View File
@@ -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)