Skip to content

feat: add --log flag for non-pausing logpoints #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mvanhorn wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat: add --log flag for non-pausing logpoints #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
65 changes: 60 additions & 5 deletions src/vibe_debug/cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@ def main():
## Useful Options

- `--break <file.py>:<line>`: set a line breakpoint before running. Repeat for multiple breakpoints.
- `--log <file.py>:<line> | <message>`: emit a logpoint message without pausing. Use `{expr}` for substitution.
- `--eval "<expr>"`: evaluate a side-effect-free expression in the paused frame. Repeat for multiple expressions.
- `--arg "<value>"`: pass one argument to the target program. Repeat for multiple program arguments.
- `--cwd <dir>`: run the target program from a specific working directory.
Expand Down Expand Up @@ -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"
Expand Down Expand Up @@ -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:
Expand All @@ -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]] = []

Expand Down Expand Up @@ -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,
}
Expand All @@ -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 = {
Expand Down Expand Up @@ -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)
Expand Down
96 changes: 84 additions & 12 deletions src/vibe_debug/mcp_server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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"],
),
},
{
Expand Down Expand Up @@ -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)))
Expand All @@ -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.",
Expand All @@ -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(
Expand Down
Loading