generated from coulomb/repo-seed
WP-0016 finished: interactive registry maintain with llm-connect automation
Some checks failed
ci / validate-registry (push) Has been cancelled
Some checks failed
ci / validate-registry (push) Has been cancelled
Closes the registry maintenance loop from inside each domain repo: interactive prompting for judgment calls, full automation for safe and high-confidence changes, both backed by the llm-connect HTTP bridge. - New modules: maintain.py, maintain_llm.py, patches.py, interactive.py - Schema: schemas/registry-patch.schema.json - CLI: reuse-surface maintain; establish --scaffold --hook - Sibling templates: Makefile fragment, pre-commit hook - Deterministic signal collectors extended; validate cwd auto-detect - Docs, gap priority 28, SCOPE update - Tests: test_maintain.py, test_interactive.py (59 pytest total) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
119
reuse_surface/interactive.py
Normal file
119
reuse_surface/interactive.py
Normal file
@@ -0,0 +1,119 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Any, Literal
|
||||
|
||||
from reuse_surface.patches import is_safe_patch
|
||||
|
||||
PromptAction = Literal["apply", "skip", "edit", "quit", "apply_all_safe"]
|
||||
|
||||
|
||||
class NonInteractiveError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
def is_tty() -> bool:
|
||||
return sys.stdin.isatty() and sys.stdout.isatty()
|
||||
|
||||
|
||||
def format_patch_summary(patch: dict[str, Any]) -> str:
|
||||
lines = [
|
||||
f" capability: {patch['capability_id']}",
|
||||
f" kind: {patch['kind']}",
|
||||
f" confidence: {patch.get('confidence', 'n/a')}",
|
||||
f" rationale: {patch.get('rationale', '')}",
|
||||
]
|
||||
for key in ("append", "value", "field_path", "dimension", "from_level", "to_level"):
|
||||
if patch.get(key) is not None:
|
||||
lines.append(f" {key}: {patch[key]}")
|
||||
if patch.get("evidence_citations"):
|
||||
lines.append(f" evidence: {', '.join(patch['evidence_citations'])}")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def emit_event(event: str, payload: dict[str, Any]) -> None:
|
||||
print(json.dumps({"event": event, **payload}, sort_keys=True))
|
||||
|
||||
|
||||
def prompt_patch(patch: dict[str, Any]) -> PromptAction:
|
||||
print("\n--- Registry patch ---")
|
||||
print(format_patch_summary(patch))
|
||||
while True:
|
||||
choice = input("[a]pply [s]kip [e]dit [q]uit [A]pply all safe? ").strip().lower()
|
||||
if choice in {"a", "apply"}:
|
||||
return "apply"
|
||||
if choice in {"s", "skip"}:
|
||||
return "skip"
|
||||
if choice in {"e", "edit"}:
|
||||
return "edit"
|
||||
if choice in {"q", "quit"}:
|
||||
return "quit"
|
||||
if choice == "":
|
||||
continue
|
||||
if choice.upper() == "A" or choice == "apply all safe":
|
||||
return "apply_all_safe"
|
||||
print("Invalid choice.")
|
||||
|
||||
|
||||
def edit_patch(patch: dict[str, Any]) -> dict[str, Any]:
|
||||
editor = os.environ.get("EDITOR", "nano")
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".yaml", delete=False) as handle:
|
||||
import yaml
|
||||
|
||||
yaml.safe_dump(patch, handle, sort_keys=False)
|
||||
temp_path = handle.name
|
||||
subprocess.run([editor, temp_path], check=False)
|
||||
import yaml
|
||||
|
||||
edited = yaml.safe_load(Path(temp_path).read_text(encoding="utf-8"))
|
||||
Path(temp_path).unlink(missing_ok=True)
|
||||
if not isinstance(edited, dict):
|
||||
return patch
|
||||
return edited
|
||||
|
||||
|
||||
def prompt_batch(
|
||||
patches: list[dict[str, Any]],
|
||||
*,
|
||||
assume_yes: bool = False,
|
||||
auto_mode: bool = False,
|
||||
emit_json: bool = False,
|
||||
) -> list[dict[str, Any]]:
|
||||
if auto_mode or assume_yes:
|
||||
return list(patches)
|
||||
|
||||
if not is_tty():
|
||||
if emit_json:
|
||||
for patch in patches:
|
||||
emit_event("suggestion", {"patch": patch, "default": "skip"})
|
||||
raise NonInteractiveError(
|
||||
"non-interactive stdin; use --auto or --yes to apply patches"
|
||||
)
|
||||
raise NonInteractiveError(
|
||||
"non-interactive stdin; use --auto or --yes to apply patches"
|
||||
)
|
||||
|
||||
selected: list[dict[str, Any]] = []
|
||||
index = 0
|
||||
while index < len(patches):
|
||||
patch = patches[index]
|
||||
action = prompt_patch(patch)
|
||||
if action == "apply_all_safe":
|
||||
selected.extend(p for p in patches[index:] if is_safe_patch(p))
|
||||
index = len(patches)
|
||||
break
|
||||
if action == "quit":
|
||||
break
|
||||
if action == "skip":
|
||||
index += 1
|
||||
continue
|
||||
if action == "edit":
|
||||
patch = edit_patch(patch)
|
||||
selected.append(patch)
|
||||
index += 1
|
||||
return selected
|
||||
Reference in New Issue
Block a user