diff --git a/PERF_TRACKER.md b/PERF_TRACKER.md new file mode 100644 index 0000000..c97b334 --- /dev/null +++ b/PERF_TRACKER.md @@ -0,0 +1,46 @@ +# clawde performance: analysis and fix — live tracker + +Branch: `perf/analysis-and-fix` (worktree `~/repo/clawde-perf`). Single source of live state; updated as last step of every increment. + +## Goal +Profile the clawde agent loop, rank bottlenecks with evidence, implement and measure the highest-impact fixes. The 1M-context bump and outer-compaction are already done; do not redo. + +## Loop map (verified by reading source) +- Supervisor `clawde-service.py`: reconciles every 10s. Per cycle per session: `pgrep -f wrapper` + `ps -ww -p PID` per matched pid + JSON config read per pid + `tmux list-panes` + `tmux rename-window` per mismatch. Runs unconditionally even when nothing changed. +- Heartbeat `driver.py`: wakes every minute boundary; on cron-matched minute runs `pane_is_idle` (1 tmux capture) + gate (`bash -c gate_command`). Steward gate = change-gate → `steward-heartbeat-probe` → `steward-status` (git fetch origin + `gh` CI API). Steward cron `*/15`. +- Wrapper `wrapper.py` + `session_watchdog.py`: restart loop; while agent runs, captures 80 pane lines every 30s. +- Discord Stop hook `enforce-discord-reply-stop-hook.py`: on EVERY turn-end, reads + `json.loads` the ENTIRE transcript jsonl. + +## Live evidence (2026-06-27) +- Running: clawde-service (pid 17349), 4 wrappers, 1 heartbeat driver. +- Transcript sizes: 149 MB max; many 25-52 MB. Stop hook parses all of it per turn. + +## Ranked bottlenecks (candidate, pre-verification) +1. **Discord Stop hook full-transcript parse per turn** — O(transcript) every turn-end; amplified by 1M bump (transcripts 25-50 MB). Fix: tail-scan only the bytes after the last user turn. HIGH. +2. **Supervisor full reconcile every 10s** — ~2+2N subprocesses/cycle (N agents) even when nothing changed; config file read per pid per cycle. Fix: skip expensive ps/config/rename when the pgrep pid-set is unchanged. MED-HIGH. +3. **steward-status git fetch + gh api on every call** — network-bound; runs in probe (15 min, gated) and on every agent steward-status. Fix: lower priority, cadence acceptable. LOW-MED. +4. **Duplicate pane/repl helpers + capture_pane_content** across modules — maintainability, minor. LOW. + +## Adversarial analysis (35-agent workflow, ~1.04M tokens, all findings verified against source) +11 findings confirmed real + on hot path. Two are MED, rest LOW. Ranked: +1. MED discord stop hook full-transcript parse per turn — FIXED (I1). Live on the one discord-wired agent. +2. MED a2a observer forks 2 tmux subprocs/s/agent forever — DORMANT (a2a server not running for any live agent); deferred, fix sketch below. +3-7. LOW heartbeat git+gh+submodule+health storm every 15 min — gated background, verifiers rated the change-aware early-out UNSAFE (skipping gh/health/submodule when the cheap repo signals look clean can miss a same-head CI pending→failing transition). Deferred by design. +8-10. LOW supervisor: N×ps (FIXED I2), config-read-per-pid (verifier: caching unsafe), global pgrep O(S²N) (safe, marginal at S=2; deferred follow-up). +11. LOW a2a unbounded task output_text — dormant; deferred. + +## Increments (done-per-increment = tests green + before/after measured + committed) +- [x] I0: baseline regression net — pytest 229 passed green on branch baseline. +- [x] I1 (91d1e97): discord stop hook tail-scan. Benchmark 47MB/8000-entry transcript, best of 5: + common case 217.8ms→0.8ms (272x); 50-deep 217.2ms→36.0ms (6x). 10 hook tests green (5 new). Semantics identical. +- [x] I2 (b92e5da): supervisor single `ps -axww` scan vs pgrep+N×ps. Discovery forks 1+N→1 per session call + (live 10→2 per cycle, ~48k fewer forks/day). 18 supervisor tests green (2 new). Matching identical to old pgrep -f. + +## Deferred (real but not fixed, with reason) +- a2a observer idle-gating: a2a server is not running live; the safe fix is non-trivial (observe() does double duty — output-diff baseline + target-death watchdog — so a naive idle-skip corrupts output attribution and breaks the death watchdog). Safe sketch: liveness-only cheap poll when no active non-terminal task + reset diff baseline at task submit; the redundant second `list-windows` fork can also be derived from capture-pane success. Not worth the risk on a dormant path. +- heartbeat storm early-out: verifiers rated unsafe (correctness risk on same-head CI/submodule/health transitions). Submodule fetches could be parallelized (safe) but it is a 15-min gated background task, not on the agent's blocking path. +- supervisor one-scan-per-cycle (O(S²N)→O(SN)): safe but marginal at S=2; needs a call-chain refactor through ensure_all_agent_windows. Recommended follow-up. + +## Assumptions +- assumed measurement via synthetic benchmarks (constructed large transcript, subprocess counters) is acceptable proof since live wrapper changes only apply on respawn; all edits are reversible and behavior-preserving. +- assumed I1's live target is the single discord-wired agent (hook wired per-discord-agent in workspace-files.nix); its transcript is daily-rotated so the per-turn parse cost grows through each day toward the benchmarked figures. diff --git a/module/channel-adapters/discord/scripts/enforce-discord-reply-stop-hook.py b/module/channel-adapters/discord/scripts/enforce-discord-reply-stop-hook.py index 651e310..cdb97b4 100644 --- a/module/channel-adapters/discord/scripts/enforce-discord-reply-stop-hook.py +++ b/module/channel-adapters/discord/scripts/enforce-discord-reply-stop-hook.py @@ -1,4 +1,5 @@ import json +import os import re import sys @@ -6,6 +7,7 @@ r']*\bchat_id="([^"]+)"' ) DISCORD_REPLY_TOOL_NAME = "mcp__plugin_discord_discord__reply" +TRANSCRIPT_REVERSE_READ_CHUNK_BYTES = 65536 def message_content_as_text(transcript_entry): @@ -59,21 +61,47 @@ def chat_id_needing_reply(transcript_entries, stop_hook_active): return chat_id -def read_transcript_entries(transcript_path): - entries = [] +def user_entry_carries_discord_envelope(entry): + return entry.get("type") == "user" and bool( + DISCORD_CHANNEL_ENVELOPE_PATTERN.search(message_content_as_text(entry)) + ) + + +def iterate_transcript_lines_newest_first(transcript_file): + transcript_file.seek(0, os.SEEK_END) + position = transcript_file.tell() + carried_prefix = b"" + while position > 0: + read_size = min(TRANSCRIPT_REVERSE_READ_CHUNK_BYTES, position) + position -= read_size + transcript_file.seek(position) + chunk = transcript_file.read(read_size) + carried_prefix + lines = chunk.split(b"\n") + carried_prefix = lines[0] + for line in reversed(lines[1:]): + yield line + if carried_prefix: + yield carried_prefix + + +def read_transcript_tail_through_latest_discord_turn(transcript_path): + collected_newest_first = [] try: - with open(transcript_path) as transcript_file: - for line in transcript_file: - stripped = line.strip() + with open(transcript_path, "rb") as transcript_file: + for raw_line in iterate_transcript_lines_newest_first(transcript_file): + stripped = raw_line.strip() if not stripped: continue try: - entries.append(json.loads(stripped)) + entry = json.loads(stripped) except json.JSONDecodeError: continue + collected_newest_first.append(entry) + if user_entry_carries_discord_envelope(entry): + break except OSError: return [] - return entries + return list(reversed(collected_newest_first)) def main(): @@ -85,7 +113,7 @@ def main(): if not transcript_path: sys.exit(0) chat_id = chat_id_needing_reply( - read_transcript_entries(transcript_path), + read_transcript_tail_through_latest_discord_turn(transcript_path), bool(hook_input.get("stop_hook_active")), ) if chat_id is None: diff --git a/module/scripts/clawde-service/agent_wrapper_reconcile.py b/module/scripts/clawde-service/agent_wrapper_reconcile.py index cf5a962..ce32b91 100644 --- a/module/scripts/clawde-service/agent_wrapper_reconcile.py +++ b/module/scripts/clawde-service/agent_wrapper_reconcile.py @@ -9,13 +9,19 @@ CONFIG_FILE_ARGUMENT_PATTERN = re.compile(r"--config-file (\S+)") -def read_process_command_line(process_id: int) -> str: +def read_all_process_command_lines() -> list[tuple[int, str]]: result = subprocess.run( - ["ps", "-ww", "-p", str(process_id), "-o", "command="], + ["ps", "-axww", "-o", "pid=,command="], capture_output=True, text=True, ) - return result.stdout.strip() + process_command_lines = [] + for line in result.stdout.splitlines(): + process_id_text, _separator, command_line = line.strip().partition(" ") + if not process_id_text.isdigit(): + continue + process_command_lines.append((int(process_id_text), command_line)) + return process_command_lines def read_tmux_session_from_launch_config(config_file_path: str) -> str | None: @@ -27,17 +33,10 @@ def read_tmux_session_from_launch_config(config_file_path: str) -> str | None: def find_session_agent_wrapper_processes(session_name: str) -> list[dict]: - pgrep_result = subprocess.run( - ["pgrep", "-f", AGENT_WRAPPER_PROCESS_MATCH_PATTERN], - capture_output=True, - text=True, - ) wrapper_processes = [] - for line in pgrep_result.stdout.split(): - if not line.strip().isdigit(): + for process_id, command_line in read_all_process_command_lines(): + if AGENT_WRAPPER_PROCESS_MATCH_PATTERN not in command_line: continue - process_id = int(line) - command_line = read_process_command_line(process_id) agent_name_match = AGENT_NAME_ARGUMENT_PATTERN.search(command_line) config_file_match = CONFIG_FILE_ARGUMENT_PATTERN.search(command_line) if not agent_name_match or not config_file_match: diff --git a/module/scripts/tests/unit/test_agent_wrapper_process_discovery.py b/module/scripts/tests/unit/test_agent_wrapper_process_discovery.py new file mode 100644 index 0000000..6417eba --- /dev/null +++ b/module/scripts/tests/unit/test_agent_wrapper_process_discovery.py @@ -0,0 +1,100 @@ +from clawde_service_test_helpers import FakeCompletedProcess, load_service_module + +service_module = load_service_module() + + +def _process_listing(command_lines_by_process_id, noise_lines=()): + lines = [ + f" {process_id} {command_line}" + for process_id, command_line in command_lines_by_process_id.items() + ] + lines.extend(noise_lines) + return "\n".join(lines) + "\n" + + +def test_find_wrapper_processes_filters_by_session_and_parses_agent_name(monkeypatch): + command_lines_by_process_id = { + 111: "python3 /n/agent-wrapper/wrapper.py --agent-name steward " + "--config-file /c/steward.json", + 222: "python3 /n/agent-wrapper/wrapper.py --agent-name copper " + "--config-file /c/copper.json", + 333: "python3 /n/agent-wrapper/wrapper.py --agent-name bronze " + "--config-file /c/bronze.json", + } + tmux_session_by_config_file_path = { + "/c/steward.json": "clawde", + "/c/copper.json": "copper", + "/c/bronze.json": "clawde", + } + + def fake_subprocess_run(arguments, capture_output, text): + assert arguments[0] == "ps" + return FakeCompletedProcess( + 0, + stdout=_process_listing( + command_lines_by_process_id, + noise_lines=[" 999 vim /n/agent-wrapper/wrapper.py"], + ), + ) + + monkeypatch.setattr( + service_module.agent_wrapper_reconcile.subprocess, "run", fake_subprocess_run + ) + monkeypatch.setattr( + service_module.agent_wrapper_reconcile, + "read_tmux_session_from_launch_config", + lambda config_file_path: tmux_session_by_config_file_path[config_file_path], + ) + + discovered = ( + service_module.agent_wrapper_reconcile.find_session_agent_wrapper_processes( + "clawde" + ) + ) + + assert discovered == [ + {"process_id": 111, "agent_name": "steward"}, + {"process_id": 333, "agent_name": "bronze"}, + ], ( + "discovery must return only wrappers whose --config-file declares tmux_session " + "'clawde', so reconciling one session never terminates another session's agents" + ) + + +def test_find_wrapper_processes_issues_a_single_process_scan_regardless_of_count( + monkeypatch, +): + command_lines_by_process_id = { + 111: "python3 /n/agent-wrapper/wrapper.py --agent-name steward " + "--config-file /c/steward.json", + 222: "python3 /n/agent-wrapper/wrapper.py --agent-name cobalt " + "--config-file /c/cobalt.json", + 333: "python3 /n/agent-wrapper/wrapper.py --agent-name bronze " + "--config-file /c/bronze.json", + } + subprocess_run_call_count = {"count": 0} + + def fake_subprocess_run(arguments, capture_output, text): + subprocess_run_call_count["count"] += 1 + return FakeCompletedProcess( + 0, stdout=_process_listing(command_lines_by_process_id) + ) + + monkeypatch.setattr( + service_module.agent_wrapper_reconcile.subprocess, "run", fake_subprocess_run + ) + monkeypatch.setattr( + service_module.agent_wrapper_reconcile, + "read_tmux_session_from_launch_config", + lambda _config_file_path: "clawde", + ) + + service_module.agent_wrapper_reconcile.find_session_agent_wrapper_processes( + "clawde" + ) + + assert subprocess_run_call_count["count"] == 1, ( + "discovery must scan all processes in a single ps call instead of one pgrep " + "plus one ps per matched pid, so per-cycle subprocess forks stay constant in " + "the number of running wrappers" + ) diff --git a/module/scripts/tests/unit/test_clawde_service_agent_identity_reconcile.py b/module/scripts/tests/unit/test_clawde_service_agent_identity_reconcile.py index 7306393..c966331 100644 --- a/module/scripts/tests/unit/test_clawde_service_agent_identity_reconcile.py +++ b/module/scripts/tests/unit/test_clawde_service_agent_identity_reconcile.py @@ -126,50 +126,3 @@ def test_creates_window_when_no_wrapper_is_running(monkeypatch): assert issued_new_windows == ["steward"] assert terminated_process_ids == [] - - -def test_find_wrapper_processes_filters_by_session_and_parses_agent_name(monkeypatch): - command_lines_by_process_id = { - 111: "python3 /n/agent-wrapper/wrapper.py --agent-name steward " - "--config-file /c/steward.json", - 222: "python3 /n/agent-wrapper/wrapper.py --agent-name copper " - "--config-file /c/copper.json", - 333: "python3 /n/agent-wrapper/wrapper.py --agent-name bronze " - "--config-file /c/bronze.json", - } - tmux_session_by_config_file_path = { - "/c/steward.json": "clawde", - "/c/copper.json": "copper", - "/c/bronze.json": "clawde", - } - - def fake_subprocess_run(arguments, capture_output, text): - if arguments[0] == "pgrep": - return FakeCompletedProcess(0, stdout="111\n222\n333\n") - process_id = int(arguments[arguments.index("-p") + 1]) - return FakeCompletedProcess( - 0, stdout=command_lines_by_process_id[process_id] + "\n" - ) - - monkeypatch.setattr( - service_module.agent_wrapper_reconcile.subprocess, "run", fake_subprocess_run - ) - monkeypatch.setattr( - service_module.agent_wrapper_reconcile, - "read_tmux_session_from_launch_config", - lambda config_file_path: tmux_session_by_config_file_path[config_file_path], - ) - - discovered = ( - service_module.agent_wrapper_reconcile.find_session_agent_wrapper_processes( - "clawde" - ) - ) - - assert discovered == [ - {"process_id": 111, "agent_name": "steward"}, - {"process_id": 333, "agent_name": "bronze"}, - ], ( - "discovery must return only wrappers whose --config-file declares tmux_session " - "'clawde', so reconciling one session never terminates another session's agents" - ) diff --git a/module/scripts/tests/unit/test_enforce_discord_reply_stop_hook.py b/module/scripts/tests/unit/test_enforce_discord_reply_stop_hook.py index 2414d68..d7dca90 100644 --- a/module/scripts/tests/unit/test_enforce_discord_reply_stop_hook.py +++ b/module/scripts/tests/unit/test_enforce_discord_reply_stop_hook.py @@ -1,4 +1,5 @@ import importlib.util +import json import pathlib @@ -83,3 +84,64 @@ def test_enforces_reply_to_newest_discord_message_even_after_earlier_reply(): _assistant_text("answered the new one in terminal"), ] assert hook.chat_id_needing_reply(entries, stop_hook_active=False) == "777" + + +def _write_transcript(tmp_path, entries): + transcript = tmp_path / "transcript.jsonl" + transcript.write_text("".join(json.dumps(entry) + "\n" for entry in entries)) + return transcript + + +def test_tail_reader_returns_suffix_from_newest_discord_turn(tmp_path): + entries = ( + [_assistant_text(f"old work {index}") for index in range(2000)] + + [_user(DISCORD_ENVELOPE)] + + [_assistant_text("answered in terminal only")] + ) + transcript = _write_transcript(tmp_path, entries) + tail = hook.read_transcript_tail_through_latest_discord_turn(str(transcript)) + assert len(tail) == 2 + assert hook.user_entry_carries_discord_envelope(tail[0]) + assert hook.chat_id_needing_reply(tail, stop_hook_active=False) == "555" + + +def test_tail_reader_decision_matches_full_parse_when_reply_present(tmp_path): + entries = ( + [_assistant_text(f"old work {index}") for index in range(500)] + + [_user(DISCORD_ENVELOPE)] + + [_assistant_reply("555")] + ) + transcript = _write_transcript(tmp_path, entries) + tail = hook.read_transcript_tail_through_latest_discord_turn(str(transcript)) + assert hook.chat_id_needing_reply(tail, stop_hook_active=False) is None + + +def test_tail_reader_ignores_older_unanswered_discord_turn(tmp_path): + entries = [ + _user(DISCORD_ENVELOPE), + _assistant_text("never answered the old one"), + _user( + '\nnewer\n' + ), + _assistant_reply("777"), + ] + transcript = _write_transcript(tmp_path, entries) + tail = hook.read_transcript_tail_through_latest_discord_turn(str(transcript)) + assert hook.chat_id_needing_reply(tail, stop_hook_active=False) is None + + +def test_tail_reader_handles_missing_trailing_newline(tmp_path): + transcript = tmp_path / "transcript.jsonl" + transcript.write_text( + json.dumps(_user(DISCORD_ENVELOPE)) + + "\n" + + json.dumps(_assistant_text("answered in terminal only")) + ) + tail = hook.read_transcript_tail_through_latest_discord_turn(str(transcript)) + assert hook.chat_id_needing_reply(tail, stop_hook_active=False) == "555" + + +def test_tail_reader_returns_empty_for_missing_file(tmp_path): + missing = tmp_path / "does-not-exist.jsonl" + assert hook.read_transcript_tail_through_latest_discord_turn(str(missing)) == []