From 09c8dc79da7cbb9b4f840fb6c57eda96da712aa5 Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Fri, 8 May 2026 22:19:26 -0700 Subject: [PATCH] feat: add --log flag for non-pausing logpoints Logpoints emit a formatted message at a source line without pausing execution. Useful for inspecting state inside hot loops and async paths where pause-eval-continue is too disruptive. CLI: vibe-debug debug-python script.py \ --log 'script.py:7 | iter={i} value={value}' \ --break script.py:18 Output (text and JSON) gains a 'logs' array containing the formatted messages with file/line attribution. Combines naturally with --break and --eval. DAP plumbing: - DebugSession.set_breakpoints accepts log_messages (per-line) - DebugSession.drain_logpoints filters DAP output events for matches by (source.path, line) when debugpy attributes them or by template-regex fallback when source is empty - MCP debug_set_breakpoints gains an entries[] schema with optional logMessage / condition / hitCondition for DAP parity --- README.md | 10 ++++ src/vibe_debug/cli.py | 65 +++++++++++++++++++-- src/vibe_debug/mcp_server.py | 96 ++++++++++++++++++++++++++---- src/vibe_debug/session.py | 103 +++++++++++++++++++++++++++++++- tests/test_cli.py | 48 ++++++++++++++- tests/test_mcp_server.py | 50 ++++++++++++++++ tests/test_session.py | 110 +++++++++++++++++++++++++++++++++++ 7 files changed, 461 insertions(+), 21 deletions(-) create mode 100644 tests/test_session.py diff --git a/README.md b/README.md index ac82480..5d7e693 100644 --- a/README.md +++ b/README.md @@ -125,6 +125,16 @@ Use `--json` when you want machine-readable output for an agent or script: npx -y github:illscience/vibe-debug debug-python ./buggy_invoice.py --break ./buggy_invoice.py:13 --eval "subtotal * (1 - rate)" --json ``` +Use `--log` for non-pausing logpoints. The message uses debugpy expression interpolation, so `{name}` is evaluated in the breakpoint frame: + +```bash +npx -y github:illscience/vibe-debug debug-python ./buggy_invoice.py \ + --log './buggy_invoice.py:13 | rate={rate} discounted={discounted}' \ + --break ./buggy_invoice.py:20 +``` + +Logpoint hits are returned in the JSON `logs` array and printed in a `Logs:` section for human output. + ## Status This is an alpha release. The first debugger backend is Python via [`debugpy`](https://github.com/microsoft/debugpy); the MCP server is designed to grow to TypeScript/Node and other language runtimes. diff --git a/src/vibe_debug/cli.py b/src/vibe_debug/cli.py index 5d08c6b..3bd32ca 100644 --- a/src/vibe_debug/cli.py +++ b/src/vibe_debug/cli.py @@ -65,6 +65,7 @@ def main(): ## Useful Options - `--break :`: set a line breakpoint before running. Repeat for multiple breakpoints. +- `--log : | `: emit a logpoint message without pausing. Use `{expr}` for substitution. - `--eval ""`: evaluate a side-effect-free expression in the paused frame. Repeat for multiple expressions. - `--arg ""`: pass one argument to the target program. Repeat for multiple program arguments. - `--cwd `: run the target program from a specific working directory. @@ -327,6 +328,18 @@ def _parse_breakpoint(value: str) -> dict[str, object]: return {"file": file_name, "line": line} +def _parse_logpoint(value: str) -> dict[str, object]: + location, separator, message = value.partition("|") + if not separator: + raise argparse.ArgumentTypeError("logpoints must look like 'path/to/file.py:12 | message {expr}'") + breakpoint = _parse_breakpoint(location.strip()) + message = message.strip() + if not message: + raise argparse.ArgumentTypeError("logpoint message must not be empty") + breakpoint["logMessage"] = message + return breakpoint + + def _source_location(location: object) -> str: if not isinstance(location, dict): return "unknown" @@ -399,9 +412,19 @@ def _local_summaries(snapshot: dict[str, object]) -> list[dict[str, object]]: return summaries +def _breakpoints_by_file(items: list[dict[str, object]]) -> dict[str, list[dict[str, object]]]: + grouped: dict[str, list[dict[str, object]]] = {} + for item in items: + file_name = str(item["file"]) + grouped.setdefault(file_name, []).append(item) + return grouped + + def _debug_python_payload(args: argparse.Namespace) -> dict[str, object]: breakpoints = args.breakpoints or [] - stop_on_entry = bool(args.stop_on_entry or not breakpoints) + logpoints = args.logpoints or [] + configured_points = [*breakpoints, *logpoints] + stop_on_entry = bool(args.stop_on_entry or not configured_points) session: DebugSession | None = None try: @@ -413,11 +436,21 @@ def _debug_python_payload(args: argparse.Namespace) -> dict[str, object]: stop_on_entry=stop_on_entry, timeout=float(args.timeout), ) - breakpoint_results = [ - session.set_breakpoints(file=str(item["file"]), lines=[int(item["line"])], cwd=args.cwd) - for item in breakpoints - ] + breakpoint_results: list[dict[str, object]] = [] + for file_name, items in _breakpoints_by_file(configured_points).items(): + breakpoint_results.append( + session.set_breakpoints( + file=file_name, + lines=[int(item["line"]) for item in items], + cwd=args.cwd, + log_messages=[ + str(item["logMessage"]) if isinstance(item.get("logMessage"), str) else None + for item in items + ], + ) + ) stopped = session.continue_execution(timeout=float(args.timeout)) + logs = session.drain_logpoints() snapshot: dict[str, object] = {} evaluations: list[dict[str, object]] = [] @@ -455,6 +488,7 @@ def _debug_python_payload(args: argparse.Namespace) -> dict[str, object]: "cwd": str(Path(args.cwd or os.getcwd()).resolve()), "breakpoints": _breakpoint_summaries(breakpoint_results), "stopped": stopped_summary, + "logs": logs, "locals": _local_summaries(snapshot), "evaluations": evaluations, } @@ -467,6 +501,18 @@ def _debug_python_payload(args: argparse.Namespace) -> dict[str, object]: def _print_debug_python_human(payload: dict[str, object]) -> None: + logs = payload.get("logs") + if isinstance(logs, list) and logs: + print("Logs:") + for item in logs: + if not isinstance(item, dict): + continue + file_name = item.get("file") + line = item.get("line") + message = item.get("message") + if isinstance(file_name, str) and isinstance(line, int) and isinstance(message, str): + print(f" {Path(file_name).name}:{line} | {message}") + stopped = payload.get("stopped") if isinstance(stopped, dict) and stopped.get("state") == "stopped": location = { @@ -845,6 +891,15 @@ def main(argv: list[str] | None = None) -> int: metavar="FILE:LINE", help="Set a line breakpoint before continuing. Repeat for multiple breakpoints.", ) + debug_python.add_argument( + "--log", + dest="logpoints", + action="append", + type=_parse_logpoint, + default=[], + metavar="FILE:LINE | MESSAGE", + help="Set a non-pausing log breakpoint. Use {expr} for substitution.", + ) debug_python.add_argument("--cwd", help="Working directory for the target program.") debug_python.add_argument("--python", help="Python executable for debugpy and the target.") debug_python.add_argument("--timeout", type=float, default=20.0) diff --git a/src/vibe_debug/mcp_server.py b/src/vibe_debug/mcp_server.py index 0c25def..b722c38 100644 --- a/src/vibe_debug/mcp_server.py +++ b/src/vibe_debug/mcp_server.py @@ -55,6 +55,9 @@ def _tool_definitions() -> list[dict[str, Any]]: "properties": { "file": {"type": "string"}, "line": {"type": "integer"}, + "condition": {"type": "string"}, + "hitCondition": {"type": "string"}, + "logMessage": {"type": "string"}, }, "required": ["file", "line"], "additionalProperties": False, @@ -118,15 +121,30 @@ def _tool_definitions() -> list[dict[str, Any]]: }, { "name": "debug_set_breakpoints", - "description": "Set one or more line breakpoints in a Python source file.", + "description": "Set one or more line breakpoints or logpoints in a Python source file.", "inputSchema": _schema( { "sessionId": {"type": "string"}, "file": {"type": "string"}, "lines": {"type": "array", "items": {"type": "integer"}}, + "entries": { + "type": "array", + "description": "Breakpoint objects. Use logMessage for non-pausing logpoints.", + "items": { + "type": "object", + "properties": { + "line": {"type": "integer"}, + "condition": {"type": "string"}, + "hitCondition": {"type": "string"}, + "logMessage": {"type": "string"}, + }, + "required": ["line"], + "additionalProperties": False, + }, + }, "cwd": {"type": "string"}, }, - ["sessionId", "file", "lines"], + ["sessionId", "file"], ), }, { @@ -356,16 +374,10 @@ def _debug_python_repro(self, args: dict[str, Any]) -> dict[str, Any]: session = self.manager.get(session_id) breakpoint_results: list[dict[str, Any]] = [] - for item in args.get("breakpoints") or []: - breakpoint_results.append( - session.set_breakpoints( - file=item["file"], - lines=[int(item["line"])], - cwd=args.get("cwd"), - ) - ) + breakpoint_results = self._set_breakpoint_items(session, args.get("breakpoints") or [], args.get("cwd")) stopped = session.continue_execution(timeout=timeout) + logs = session.drain_logpoints() snapshot: dict[str, Any] = {} if stopped.get("state") == "stopped": snapshot = session.top_frame_locals(limit=int(args.get("locals_limit", 40))) @@ -378,6 +390,7 @@ def _debug_python_repro(self, args: dict[str, Any]) -> dict[str, Any]: "launch": launch, "breakpoints": breakpoint_results, "stopped": stopped, + "logs": logs, "snapshot": snapshot, "nextActions": [ "Use debug_step to move over/into/out from the current line.", @@ -395,14 +408,73 @@ def _debug_attach(self, args: dict[str, Any]) -> dict[str, Any]: ) def _debug_set_breakpoints(self, args: dict[str, Any]) -> dict[str, Any]: + lines = args.get("lines") + entries = args.get("entries") + if (lines is None) == (entries is None): + raise ValueError("provide exactly one of lines or entries") + + if entries is not None: + if not isinstance(entries, list): + raise ValueError("entries must be an array") + return self.manager.get(args["sessionId"]).set_breakpoints( + file=args["file"], + lines=[int(item["line"]) for item in entries], + cwd=args.get("cwd"), + conditions=[ + item.get("condition") if isinstance(item.get("condition"), str) else None for item in entries + ], + hit_conditions=[ + item.get("hitCondition") if isinstance(item.get("hitCondition"), str) else None for item in entries + ], + log_messages=[ + item.get("logMessage") if isinstance(item.get("logMessage"), str) else None for item in entries + ], + ) + return self.manager.get(args["sessionId"]).set_breakpoints( file=args["file"], - lines=[int(line) for line in args["lines"]], + lines=[int(line) for line in lines], cwd=args.get("cwd"), ) def _debug_continue(self, args: dict[str, Any]) -> dict[str, Any]: - return self.manager.get(args["sessionId"]).continue_execution(timeout=float(args.get("timeout", 15))) + session = self.manager.get(args["sessionId"]) + result = session.continue_execution(timeout=float(args.get("timeout", 15))) + result["logs"] = session.drain_logpoints() + return result + + def _set_breakpoint_items( + self, + session: Any, + items: list[dict[str, Any]], + cwd: str | None, + ) -> list[dict[str, Any]]: + grouped: dict[str, list[dict[str, Any]]] = {} + for item in items: + grouped.setdefault(str(item["file"]), []).append(item) + + results: list[dict[str, Any]] = [] + for file_name, file_items in grouped.items(): + results.append( + session.set_breakpoints( + file=file_name, + lines=[int(item["line"]) for item in file_items], + cwd=cwd, + conditions=[ + item.get("condition") if isinstance(item.get("condition"), str) else None + for item in file_items + ], + hit_conditions=[ + item.get("hitCondition") if isinstance(item.get("hitCondition"), str) else None + for item in file_items + ], + log_messages=[ + item.get("logMessage") if isinstance(item.get("logMessage"), str) else None + for item in file_items + ], + ) + ) + return results def _debug_step(self, args: dict[str, Any]) -> dict[str, Any]: return self.manager.get(args["sessionId"]).step( diff --git a/src/vibe_debug/session.py b/src/vibe_debug/session.py index a5c85dc..e097cb3 100644 --- a/src/vibe_debug/session.py +++ b/src/vibe_debug/session.py @@ -1,6 +1,7 @@ from __future__ import annotations import os +import re import signal import socket import subprocess @@ -9,7 +10,7 @@ import uuid from dataclasses import dataclass, field from pathlib import Path -from typing import Any +from typing import Any, Pattern from .dap import DAPClient, DAPEvent @@ -60,6 +61,9 @@ class DebugSession: stopped_thread_id: int | None = None launch_request_seq: int | None = None event_cursor: int = 0 + log_event_cursor: int = 0 + logpoint_locations: set[tuple[str, int]] = field(default_factory=set) + logpoint_templates: list[tuple[str, int, str, Pattern[str]]] = field(default_factory=list) metadata: dict[str, Any] = field(default_factory=dict) @classmethod @@ -151,13 +155,39 @@ def attach( session.state = "configuring" return session - def set_breakpoints(self, file: str, lines: list[int], cwd: str | None = None) -> dict[str, Any]: + def set_breakpoints( + self, + file: str, + lines: list[int], + cwd: str | None = None, + conditions: list[str | None] | None = None, + hit_conditions: list[str | None] | None = None, + log_messages: list[str | None] | None = None, + ) -> dict[str, Any]: path = _normalize_path(file, cwd) + breakpoints: list[dict[str, Any]] = [] + self.logpoint_locations = {item for item in self.logpoint_locations if item[0] != path} + self.logpoint_templates = [item for item in self.logpoint_templates if item[0] != path] + for index, line in enumerate(lines): + breakpoint: dict[str, Any] = {"line": int(line)} + if conditions and index < len(conditions) and conditions[index]: + breakpoint["condition"] = conditions[index] + if hit_conditions and index < len(hit_conditions) and hit_conditions[index]: + breakpoint["hitCondition"] = hit_conditions[index] + if log_messages and index < len(log_messages) and log_messages[index]: + log_message = log_messages[index] + breakpoint["logMessage"] = log_message + self.logpoint_locations.add((path, int(line))) + self.logpoint_templates.append( + (path, int(line), log_message, _logpoint_output_pattern(log_message)) + ) + breakpoints.append(breakpoint) + body = self.client.request( "setBreakpoints", { "source": {"path": path}, - "breakpoints": [{"line": int(line)} for line in lines], + "breakpoints": breakpoints, "sourceModified": False, }, ) @@ -168,6 +198,20 @@ def set_breakpoints(self, file: str, lines: list[int], cwd: str | None = None) - "breakpoints": body.get("breakpoints", []), } + def drain_logpoints(self) -> list[dict[str, Any]]: + events = self.client.events + pending = events[self.log_event_cursor :] + self.log_event_cursor = len(events) + + logs: list[dict[str, Any]] = [] + for event in pending: + if event.event != "output": + continue + log = self._logpoint_summary(event) + if log: + logs.append(log) + return logs + def continue_execution(self, timeout: float = 15.0) -> dict[str, Any]: if self.state == "configuring": start = self.client.event_count() @@ -434,6 +478,59 @@ def _frame_summary(frame: dict[str, Any]) -> dict[str, Any]: }, } + def _logpoint_summary(self, event: DAPEvent) -> dict[str, Any] | None: + category = event.body.get("category") + if category not in {"console", "stdout"}: + return None + + source = event.body.get("source") + output = event.body.get("output") + if not isinstance(output, str): + return None + message = output.rstrip("\r\n") + + if not isinstance(source, dict): + return None + source_path = source.get("path") + line = event.body.get("line") + if not isinstance(source_path, str) or not isinstance(line, int): + return self._template_logpoint_summary(message, event.body) + + path = _normalize_path(source_path) + if (path, line) not in self.logpoint_locations: + return None + + log: dict[str, Any] = { + "file": path, + "line": line, + "message": message, + } + thread_id = event.body.get("threadId") + if isinstance(thread_id, int): + log["thread_id"] = thread_id + return log + + def _template_logpoint_summary(self, message: str, body: dict[str, Any]) -> dict[str, Any] | None: + for path, line, _template, pattern in self.logpoint_templates: + if not pattern.match(message): + continue + log: dict[str, Any] = { + "file": path, + "line": line, + "message": message, + } + thread_id = body.get("threadId") + if isinstance(thread_id, int): + log["thread_id"] = thread_id + return log + return None + + +def _logpoint_output_pattern(message: str) -> Pattern[str]: + parts = re.split(r"(\{[^{}]+\})", message) + pattern = "".join(".*" if part.startswith("{") and part.endswith("}") else re.escape(part) for part in parts) + return re.compile(f"^{pattern}$") + class DebugSessionManager: def __init__(self) -> None: diff --git a/tests/test_cli.py b/tests/test_cli.py index 643642c..73f525c 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,5 +1,6 @@ from __future__ import annotations +import argparse import json import tempfile import unittest @@ -7,7 +8,7 @@ from io import StringIO from pathlib import Path -from vibe_debug.cli import _format_claude_stream, main +from vibe_debug.cli import _format_claude_stream, _parse_logpoint, main def call_cli(args: list[str]) -> tuple[int, str]: @@ -137,6 +138,51 @@ def test_debug_python_stops_and_prints_locals(self) -> None: self.assertIn("y = 42", output) self.assertIn("y -> 42", output) + def test_debug_python_logs_and_stops_at_breakpoint(self) -> None: + with tempfile.TemporaryDirectory() as directory: + script = Path(directory) / "sample.py" + script.write_text( + "\n".join( + [ + "def main():", + " seen = []", + " for i in range(3):", + " value = i * 10", + " seen.append(value)", + " print(seen)", + "", + "if __name__ == '__main__':", + " main()", + "", + ] + ), + encoding="utf-8", + ) + + code, output = call_cli( + [ + "debug-python", + str(script), + "--log", + f"{script}:5 | i={{i}} value={{value}}", + "--break", + f"{script}:6", + "--json", + ] + ) + + self.assertEqual(code, 0) + payload = json.loads(output) + self.assertEqual(payload["stopped"]["line"], 6) + self.assertEqual( + [item["message"] for item in payload["logs"]], + ["i=0 value=0", "i=1 value=10", "i=2 value=20"], + ) + + def test_parse_logpoint_requires_separator(self) -> None: + with self.assertRaises(argparse.ArgumentTypeError): + _parse_logpoint("sample.py:5 missing separator") + def test_claude_progress_formats_debugger_events(self) -> None: events = [ { diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 6d6a0bb..4a1332c 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -1,10 +1,28 @@ from __future__ import annotations import unittest +from typing import Any from vibe_debug.mcp_server import MCPDebuggerServer +class FakeSession: + def __init__(self) -> None: + self.calls: list[dict[str, Any]] = [] + + def set_breakpoints(self, **kwargs: Any) -> dict[str, Any]: + self.calls.append(kwargs) + return {"ok": True} + + +class FakeManager: + def __init__(self) -> None: + self.session = FakeSession() + + def get(self, session_id: str) -> FakeSession: + return self.session + + class MCPServerTests(unittest.TestCase): def test_initialize_and_tool_list(self) -> None: server = MCPDebuggerServer() @@ -28,6 +46,38 @@ def test_initialize_and_tool_list(self) -> None: self.assertIn("debug_step", tools) self.assertIn("debug_variables", tools) + set_breakpoints = next(tool for tool in listed["result"]["tools"] if tool["name"] == "debug_set_breakpoints") + entries = set_breakpoints["inputSchema"]["properties"]["entries"] + self.assertEqual(entries["items"]["properties"]["logMessage"]["type"], "string") + + def test_debug_set_breakpoints_accepts_logpoint_entries(self) -> None: + server = MCPDebuggerServer() + manager = FakeManager() + server.manager = manager # type: ignore[assignment] + + result = server._debug_set_breakpoints( + { + "sessionId": "session-1", + "file": "sample.py", + "entries": [{"line": 5, "logMessage": "x={x}"}], + } + ) + + self.assertEqual(result, {"ok": True}) + self.assertEqual( + manager.session.calls, + [ + { + "file": "sample.py", + "lines": [5], + "cwd": None, + "conditions": [None], + "hit_conditions": [None], + "log_messages": ["x={x}"], + } + ], + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_session.py b/tests/test_session.py new file mode 100644 index 0000000..9a0df1c --- /dev/null +++ b/tests/test_session.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +import unittest +from pathlib import Path +from typing import Any + +from vibe_debug.dap import DAPEvent +from vibe_debug.session import DebugSession + + +class FakeClient: + def __init__(self) -> None: + self.requests: list[tuple[str, dict[str, Any]]] = [] + self._events: list[DAPEvent] = [] + + def request( + self, + command: str, + arguments: dict[str, Any] | None = None, + timeout: float | None = None, + ) -> dict[str, Any]: + self.requests.append((command, arguments or {})) + return {"breakpoints": arguments.get("breakpoints", []) if arguments else []} + + @property + def events(self) -> list[DAPEvent]: + return list(self._events) + + +class DebugSessionTests(unittest.TestCase): + def test_set_breakpoints_sends_log_messages_to_dap(self) -> None: + client = FakeClient() + session = DebugSession(session_id="session-1", client=client) # type: ignore[arg-type] + + session.set_breakpoints( + file="sample.py", + lines=[5, 9], + cwd="/tmp/project", + conditions=[None, "enabled"], + hit_conditions=[None, "3"], + log_messages=["x={x}", None], + ) + + self.assertEqual(client.requests[0][0], "setBreakpoints") + arguments = client.requests[0][1] + self.assertEqual( + arguments["breakpoints"], + [ + {"line": 5, "logMessage": "x={x}"}, + {"line": 9, "condition": "enabled", "hitCondition": "3"}, + ], + ) + + def test_drain_logpoints_returns_known_logpoint_output_events(self) -> None: + client = FakeClient() + session = DebugSession(session_id="session-1", client=client) # type: ignore[arg-type] + path = str((Path("/tmp/project") / "sample.py").resolve()) + session.logpoint_locations.add((path, 5)) + client._events.append( + DAPEvent( + event="output", + body={ + "category": "console", + "output": "x=42\n", + "source": {"path": path}, + "line": 5, + "threadId": 1, + }, + raw={}, + ) + ) + + self.assertEqual( + session.drain_logpoints(), + [{"file": path, "line": 5, "message": "x=42", "thread_id": 1}], + ) + self.assertEqual(session.drain_logpoints(), []) + + def test_drain_logpoints_matches_debugpy_output_with_empty_source(self) -> None: + client = FakeClient() + session = DebugSession(session_id="session-1", client=client) # type: ignore[arg-type] + path = str((Path("/tmp/project") / "sample.py").resolve()) + session.set_breakpoints(file="sample.py", lines=[5], cwd="/tmp/project", log_messages=["x={x}"]) + client._events.append( + DAPEvent( + event="output", + body={"category": "stdout", "output": "x=42\n", "source": {}}, + raw={}, + ) + ) + + self.assertEqual(session.drain_logpoints(), [{"file": path, "line": 5, "message": "x=42"}]) + + def test_drain_logpoints_ignores_program_stdout_without_source(self) -> None: + client = FakeClient() + session = DebugSession(session_id="session-1", client=client) # type: ignore[arg-type] + session.set_breakpoints(file="sample.py", lines=[5], cwd="/tmp/project", log_messages=["value={value}"]) + client._events.append( + DAPEvent( + event="output", + body={"category": "stdout", "output": "value=from print\n"}, + raw={}, + ) + ) + + self.assertEqual(session.drain_logpoints(), []) + + +if __name__ == "__main__": + unittest.main()