Spaces:
Running
Running
File size: 3,582 Bytes
c0216d4 ed362fe c0216d4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | """Create a reviewable developer queue from canonical cycle proposals.
The agent is intentionally proposal-first: it may prepare documentation and
quality-work artifacts, but it never edits production code or merges a change.
"""
from __future__ import annotations
import json
import os
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
MAX_REPORTS = 100
MAX_PROPOSALS = 50
def _safe(value: Any, limit: int = 500) -> str:
text = " ".join(str(value or "").split())
return text[:limit]
def collect(root: Path) -> list[dict[str, str]]:
proposals: list[dict[str, str]] = []
reports = sorted((root / "reports").glob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True)
for report_path in reports[:MAX_REPORTS]:
try:
report = json.loads(report_path.read_text(encoding="utf-8"))
except (OSError, ValueError):
continue
governance = report.get("governance") if isinstance(report, dict) else {}
candidates = governance.get("open_proposals", []) if isinstance(governance, dict) else []
if not isinstance(candidates, list):
continue
for item in candidates:
if not isinstance(item, dict):
continue
proposals.append({
"title": _safe(item.get("title", "Untitled proposal"), 180),
"rationale": _safe(item.get("rationale", item.get("reason", ""))),
"impact": _safe(item.get("impact_area", item.get("area", "quality")), 120),
"source": report_path.name,
})
if len(proposals) >= MAX_PROPOSALS:
return proposals
return proposals
def render(proposals: list[dict[str, str]]) -> str:
now = datetime.now(timezone.utc).isoformat()
lines = [
"# Autonomous Developer Queue",
"",
f"Generated at `{now}`.",
"",
"> This is a review queue, not an authorization to change production code. Every implementation must pass CI and receive human review through a pull request.",
"",
]
if not proposals:
lines.append("No new reviewable proposals were found.")
return "\n".join(lines) + "\n"
lines.extend(["| Priority | Proposal | Impact | Source |", "|---:|---|---|---|"])
for index, proposal in enumerate(proposals, start=1):
title = proposal["title"].replace("|", "-")
impact = proposal["impact"].replace("|", "-")
source = proposal["source"].replace("|", "-")
lines.append(f"| {index} | {title} | {impact} | `{source}` |")
if proposal["rationale"]:
lines.append(f"| | Rationale: {proposal['rationale'].replace('|', '-') } | | |")
lines.extend([
"",
"## Required review gates",
"",
"1. Confirm the proposal is within peaceful civilian-energy scope.",
"2. Add or update deterministic tests before implementation.",
"3. Run safety, health, compile and regression checks.",
"4. Require human approval before merge or any external write.",
])
return "\n".join(lines) + "\n"
def main() -> int:
root = Path(os.getenv("NI_PROJECT_ROOT", Path.cwd())).resolve()
# Keep the queue in a tracked path; reports/ is intentionally gitignored.
output = root / "knowledge_base" / "developer_queue.md"
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(render(collect(root)), encoding="utf-8")
print(output)
return 0
if __name__ == "__main__":
raise SystemExit(main())
|