mirror of
https://github.com/kuhyx/utils.git
synced 2026-11-10 21:27:14 +01:00
Krzysztof kuhy Rudnicki
aec96eefd3
Python 3.14 stdlib package wiring git to the todo MCP server: - closer (post-commit): runs the done_check of each Todo: trailer task and close_task's it only on exit 0. It FAILS OPEN: the commit already happened, so a down server, bad token or no gh prints one warning and exits 0 instead of failing the hook. - sweep: nightly drift report (systemd units shipped, not enabled). - import-pr: a merged PR's '## Todo:' items become tasks, deduped. - A done_check runs only if its sha256 is in ~/.config/todo-hooks/trusted-checks; 'Check: manual' notes are never auto-closed. - Hooks exposed in .pre-commit-hooks.yaml (post-commit, post-merge, post-rewrite). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
63 lines
1.6 KiB
Python
63 lines
1.6 KiB
Python
"""Run a done_check with a hard timeout, in its own process group."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import os
|
|
import signal
|
|
import subprocess
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from pathlib import Path
|
|
|
|
CHECK_TIMEOUT = 55.0
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CheckResult:
|
|
"""Exit code, or None on timeout or failure to start."""
|
|
|
|
code: int | None
|
|
timed_out: bool
|
|
|
|
@property
|
|
def passed(self) -> bool:
|
|
return self.code == 0
|
|
|
|
def describe(self) -> str:
|
|
if self.timed_out:
|
|
return f"timed out after {CHECK_TIMEOUT:g}s"
|
|
return "could not start" if self.code is None else f"exit {self.code}"
|
|
|
|
|
|
def run_check(
|
|
command: str,
|
|
cwd: Path,
|
|
*,
|
|
timeout: float = CHECK_TIMEOUT,
|
|
wrapper: Path | None = None,
|
|
) -> CheckResult:
|
|
"""``bash -c command`` from `cwd` without GIT_* env; kill the group on timeout."""
|
|
argv = [str(wrapper)] if wrapper else []
|
|
argv += ["bash", "-c", command]
|
|
env = {k: v for k, v in os.environ.items() if not k.startswith("GIT_")}
|
|
try:
|
|
proc = subprocess.Popen(
|
|
argv,
|
|
cwd=cwd,
|
|
env=env,
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
start_new_session=True,
|
|
)
|
|
except OSError:
|
|
return CheckResult(None, timed_out=False)
|
|
try:
|
|
return CheckResult(proc.wait(timeout=timeout), timed_out=False)
|
|
except subprocess.TimeoutExpired:
|
|
os.killpg(proc.pid, signal.SIGKILL)
|
|
proc.wait()
|
|
return CheckResult(None, timed_out=True)
|