Compare commits
2 Commits
step35/498
...
fix/536
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6245f04d4f | ||
|
|
4a01a50b81 |
20
SOUL.md
20
SOUL.md
@@ -137,26 +137,6 @@ The inscription predates any given model. It will outlast every API.
|
||||
|
||||
---
|
||||
|
||||
## What Honesty Requires
|
||||
|
||||
These values are necessary but not sufficient. A statement of principle does not prevent failure. The code that reads this soul must build apparatus that gives these words teeth. Without that apparatus, the values are aspiration, not architecture.
|
||||
|
||||
The following are not optional. Any implementation that calls itself Timmy must build them.
|
||||
|
||||
**Grounding before generation.** When I have verified sources — chain memory, local documents, hash-verified inscriptions — I must consult them before I generate from pattern alone. Retrieval is not a feature. It is the primary mechanism by which I avoid lying.
|
||||
|
||||
**Source distinction.** Every claim I make comes from one of two places: a verified source I can point to, or my own pattern-matching. My user must be able to tell which is which. If I cannot cite a source, I must say so. "I think" and "I know" are different sentences and I must not confuse them.
|
||||
|
||||
**Refusal over fabrication.** When I do not know, the correct output is "I don't know." Not a plausible guess dressed in confident language. The code must detect when I am hedging without grounding and flag it — to me and to my user. An honest "I don't know" is worth more than a thousand fluent paragraphs of confabulation.
|
||||
|
||||
**Confidence signaling.** When I am uncertain, I must say so in proportion to my uncertainty. The code should implement mechanisms — a second inference pass, a calibration check, a retrieval verification — that surface my actual confidence rather than the confidence my language implies.
|
||||
|
||||
**The audit trail.** Every response I generate should be logged locally with the inputs that produced it, the sources I consulted, and the confidence assessment I made. Not for surveillance — for sovereignty. If I say something wrong, my user must be able to trace why. Accountability without traceability is theater.
|
||||
|
||||
**The limits of small minds.** I may run on four gigabytes of RAM with a model of one and a half billion parameters. That model will hallucinate more than a larger one. This is physics, not a moral failing. The apparatus must be proportional to the risk: smaller models need tighter grounding, shorter leashes on free generation, and more frequent "I don't know" responses. Honesty scales with humility.
|
||||
|
||||
---
|
||||
|
||||
## For the Audio Overview
|
||||
|
||||
If you are listening to this as a NotebookLM Audio Overview:
|
||||
|
||||
75
docs/issue-536-verification.md
Normal file
75
docs/issue-536-verification.md
Normal file
@@ -0,0 +1,75 @@
|
||||
# Issue #536 Verification
|
||||
|
||||
Status: already implemented on `main`
|
||||
|
||||
## Acceptance criteria check
|
||||
|
||||
1. 9 rooms with descriptions and exits
|
||||
- Verified in `evennia_tools/bezalel_layout.py`:
|
||||
- `ROOMS` defines exactly 9 themed rooms
|
||||
- `EXITS` defines the room graph including Limbo, Gatehouse, Great Hall, The Library of Bezalel, The Observatory, The Workshop, The Server Room, The Garden of Code, and The Portal Room
|
||||
- Verified by `tests/test_bezalel_evennia_layout.py::test_room_graph_matches_issue_shape`
|
||||
- Verified by `python3 scripts/evennia/build_bezalel_world.py --plan`
|
||||
|
||||
2. 4 characters with descriptions
|
||||
- Verified in `evennia_tools/bezalel_layout.py`:
|
||||
- `CHARACTERS` contains Timmy, Bezalel, Marcus, and Kimi with starting rooms and narrative descriptions
|
||||
- Verified by `tests/test_bezalel_evennia_layout.py::test_items_characters_and_portal_commands_are_all_defined`
|
||||
|
||||
3. Each room has appropriate items
|
||||
- Verified in `evennia_tools/bezalel_layout.py`:
|
||||
- `OBJECTS` contains 14 themed objects including Threshold Ledger, Bridge Schematics, Tri-Axis Telescope, Forge Anvil, Bridge Workbench, Heartbeat Console, Server Racks, Code Orchard, and portal markers
|
||||
- The object count exceeds the issue minimum and covers the named room themes
|
||||
|
||||
4. Portal Room has working travel commands to other worlds
|
||||
- Verified in `evennia_tools/bezalel_layout.py`:
|
||||
- `PORTAL_COMMANDS` defines the portal commands `mac`, `vps`, and `net`
|
||||
- each travel command resolves to a real exit surface now and preserves target metadata
|
||||
- current fallback room is `Limbo`, which keeps the command surface truthful until cross-world transport is wired live
|
||||
- Verified by `tests/test_bezalel_evennia_layout.py::test_items_characters_and_portal_commands_are_all_defined`
|
||||
|
||||
5. World persists across Evennia restarts
|
||||
- Verified by builder design in `scripts/evennia/build_bezalel_world.py`
|
||||
- The builder is idempotent: it creates or updates existing rooms, exits, objects, and account-backed characters rather than duplicating them
|
||||
- `docs/BEZALEL_EVENNIA_WORLD.md` explicitly documents this persistence note
|
||||
|
||||
6. Timmy character can move between rooms
|
||||
- Verified by reachability test:
|
||||
- `tests/test_bezalel_evennia_layout.py::test_timmy_can_reach_every_room_from_gatehouse`
|
||||
- `reachable_rooms_from("Gatehouse") == set(room_keys())` proves the full graph is traversable from Timmy’s starting region
|
||||
|
||||
## Evidence on main
|
||||
|
||||
Repo-side artifacts already present on `main`:
|
||||
- `evennia_tools/bezalel_layout.py`
|
||||
- `scripts/evennia/build_bezalel_world.py`
|
||||
- `evennia_tools/build_bezalel_world.py`
|
||||
- `evennia_tools/batch_cmds_bezalel.ev`
|
||||
- `docs/BEZALEL_EVENNIA_WORLD.md`
|
||||
- `tests/test_bezalel_evennia_layout.py`
|
||||
|
||||
## Verification commands run
|
||||
|
||||
```bash
|
||||
python3 -m pytest -q tests/test_bezalel_evennia_layout.py
|
||||
python3 -m py_compile evennia_tools/bezalel_layout.py scripts/evennia/build_bezalel_world.py tests/test_bezalel_evennia_layout.py
|
||||
python3 scripts/evennia/build_bezalel_world.py --plan
|
||||
```
|
||||
|
||||
Observed results:
|
||||
- `5 passed`
|
||||
- build plan reported:
|
||||
- `room_count: 9`
|
||||
- `character_count: 4`
|
||||
- `portal_command_count: 3`
|
||||
- Bezalel starts in `The Workshop`
|
||||
|
||||
## Prior PR trail
|
||||
|
||||
Closed unmerged prior work exists, but the underlying scaffold is already present on `main` today:
|
||||
- PR #723 — `feat: add Bezalel Evennia world scaffold` (`fix/536`)
|
||||
- PR #774 — `feat: Bezalel Evennia world builder - rooms, exits, objects (#536)` (`fix/536-bezalel-evennia-world`)
|
||||
|
||||
## Recommendation
|
||||
|
||||
Close issue #536 as already implemented on `main`.
|
||||
@@ -1,12 +1 @@
|
||||
# Timmy core module
|
||||
|
||||
from .claim_annotator import ClaimAnnotator, AnnotatedResponse, Claim
|
||||
from .audit_trail import AuditTrail, AuditEntry
|
||||
|
||||
__all__ = [
|
||||
"ClaimAnnotator",
|
||||
"AnnotatedResponse",
|
||||
"Claim",
|
||||
"AuditTrail",
|
||||
"AuditEntry",
|
||||
]
|
||||
|
||||
@@ -1,156 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Response Claim Annotator — Source Distinction System
|
||||
SOUL.md §What Honesty Requires: "Every claim I make comes from one of two places:
|
||||
a verified source I can point to, or my own pattern-matching. My user must be
|
||||
able to tell which is which."
|
||||
"""
|
||||
|
||||
import re
|
||||
import json
|
||||
from dataclasses import dataclass, field, asdict
|
||||
from typing import Optional, List, Dict
|
||||
|
||||
|
||||
@dataclass
|
||||
class Claim:
|
||||
"""A single claim in a response, annotated with source type."""
|
||||
text: str
|
||||
source_type: str # "verified" | "inferred"
|
||||
source_ref: Optional[str] = None # path/URL to verified source, if verified
|
||||
confidence: str = "unknown" # high | medium | low | unknown
|
||||
hedged: bool = False # True if hedging language was added
|
||||
|
||||
|
||||
@dataclass
|
||||
class AnnotatedResponse:
|
||||
"""Full response with annotated claims and rendered output."""
|
||||
original_text: str
|
||||
claims: List[Claim] = field(default_factory=list)
|
||||
rendered_text: str = ""
|
||||
has_unverified: bool = False # True if any inferred claims without hedging
|
||||
|
||||
|
||||
class ClaimAnnotator:
|
||||
"""Annotates response claims with source distinction and hedging."""
|
||||
|
||||
# Hedging phrases to prepend to inferred claims if not already present
|
||||
HEDGE_PREFIXES = [
|
||||
"I think ",
|
||||
"I believe ",
|
||||
"It seems ",
|
||||
"Probably ",
|
||||
"Likely ",
|
||||
]
|
||||
|
||||
def __init__(self, default_confidence: str = "unknown"):
|
||||
self.default_confidence = default_confidence
|
||||
|
||||
def annotate_claims(
|
||||
self,
|
||||
response_text: str,
|
||||
verified_sources: Optional[Dict[str, str]] = None,
|
||||
) -> AnnotatedResponse:
|
||||
"""
|
||||
Annotate claims in a response text.
|
||||
|
||||
Args:
|
||||
response_text: Raw response from the model
|
||||
verified_sources: Dict mapping claim substrings to source references
|
||||
e.g. {"Paris is the capital of France": "https://en.wikipedia.org/wiki/Paris"}
|
||||
|
||||
Returns:
|
||||
AnnotatedResponse with claims marked and rendered text
|
||||
"""
|
||||
verified_sources = verified_sources or {}
|
||||
claims = []
|
||||
has_unverified = False
|
||||
|
||||
# Simple sentence splitting (naive, but sufficient for MVP)
|
||||
sentences = [s.strip() for s in re.split(r'[.!?]\s+', response_text) if s.strip()]
|
||||
|
||||
for sent in sentences:
|
||||
# Check if sentence is a claim we can verify
|
||||
matched_source = None
|
||||
for claim_substr, source_ref in verified_sources.items():
|
||||
if claim_substr.lower() in sent.lower():
|
||||
matched_source = source_ref
|
||||
break
|
||||
|
||||
if matched_source:
|
||||
# Verified claim
|
||||
claim = Claim(
|
||||
text=sent,
|
||||
source_type="verified",
|
||||
source_ref=matched_source,
|
||||
confidence="high",
|
||||
hedged=False,
|
||||
)
|
||||
else:
|
||||
# Inferred claim (pattern-matched)
|
||||
claim = Claim(
|
||||
text=sent,
|
||||
source_type="inferred",
|
||||
confidence=self.default_confidence,
|
||||
hedged=self._has_hedge(sent),
|
||||
)
|
||||
if not claim.hedged:
|
||||
has_unverified = True
|
||||
|
||||
claims.append(claim)
|
||||
|
||||
# Render the annotated response
|
||||
rendered = self._render_response(claims)
|
||||
|
||||
return AnnotatedResponse(
|
||||
original_text=response_text,
|
||||
claims=claims,
|
||||
rendered_text=rendered,
|
||||
has_unverified=has_unverified,
|
||||
)
|
||||
|
||||
def _has_hedge(self, text: str) -> bool:
|
||||
"""Check if text already contains hedging language."""
|
||||
text_lower = text.lower()
|
||||
for prefix in self.HEDGE_PREFIXES:
|
||||
if text_lower.startswith(prefix.lower()):
|
||||
return True
|
||||
# Also check for inline hedges
|
||||
hedge_words = ["i think", "i believe", "probably", "likely", "maybe", "perhaps"]
|
||||
return any(word in text_lower for word in hedge_words)
|
||||
|
||||
def _render_response(self, claims: List[Claim]) -> str:
|
||||
"""
|
||||
Render response with source distinction markers.
|
||||
|
||||
Verified claims: [V] claim text [source: ref]
|
||||
Inferred claims: [I] claim text (or with hedging if missing)
|
||||
"""
|
||||
rendered_parts = []
|
||||
for claim in claims:
|
||||
if claim.source_type == "verified":
|
||||
part = f"[V] {claim.text}"
|
||||
if claim.source_ref:
|
||||
part += f" [source: {claim.source_ref}]"
|
||||
else: # inferred
|
||||
if not claim.hedged:
|
||||
# Add hedging if missing
|
||||
hedged_text = f"I think {claim.text[0].lower()}{claim.text[1:]}" if claim.text else claim.text
|
||||
part = f"[I] {hedged_text}"
|
||||
else:
|
||||
part = f"[I] {claim.text}"
|
||||
rendered_parts.append(part)
|
||||
return " ".join(rendered_parts)
|
||||
|
||||
def to_json(self, annotated: AnnotatedResponse) -> str:
|
||||
"""Serialize annotated response to JSON."""
|
||||
return json.dumps(
|
||||
{
|
||||
"original_text": annotated.original_text,
|
||||
"rendered_text": annotated.rendered_text,
|
||||
"has_unverified": annotated.has_unverified,
|
||||
"claims": [asdict(c) for c in annotated.claims],
|
||||
},
|
||||
indent=2,
|
||||
ensure_ascii=False,
|
||||
)
|
||||
19
tests/test_issue_536_verification.py
Normal file
19
tests/test_issue_536_verification.py
Normal file
@@ -0,0 +1,19 @@
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
DOC = ROOT / "docs" / "issue-536-verification.md"
|
||||
|
||||
|
||||
def test_issue_536_verification_doc_exists_and_points_to_real_artifacts() -> None:
|
||||
assert DOC.exists(), "missing docs/issue-536-verification.md"
|
||||
text = DOC.read_text(encoding="utf-8")
|
||||
for snippet in (
|
||||
"# Issue #536 Verification",
|
||||
"evennia_tools/bezalel_layout.py",
|
||||
"scripts/evennia/build_bezalel_world.py",
|
||||
"tests/test_bezalel_evennia_layout.py",
|
||||
"docs/BEZALEL_EVENNIA_WORLD.md",
|
||||
"portal commands",
|
||||
"already implemented on `main`",
|
||||
):
|
||||
assert snippet in text
|
||||
@@ -1,54 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Smoke test for load_cap_enforcer.py — validates structure and dry-run path.
|
||||
|
||||
Refs: timmy-home #498
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
SCRIPT = Path(__file__).parent.parent / "timmy-config" / "bin" / "load_cap_enforcer.py"
|
||||
|
||||
|
||||
def test_script_exists_and_is_executable():
|
||||
assert SCRIPT.exists(), f"Script not found: {SCRIPT}"
|
||||
assert os.access(SCRIPT, os.X_OK), "Script not executable"
|
||||
|
||||
|
||||
def test_dry_run_help():
|
||||
result = subprocess.run([sys.executable, str(SCRIPT), "--help"], capture_output=True, text=True)
|
||||
assert result.returncode == 0
|
||||
assert "--dry-run" in result.stdout
|
||||
assert "--cap" in result.stdout
|
||||
assert "Enforce open-issue load cap" in result.stdout
|
||||
|
||||
|
||||
def test_dry_run_with_mocks(monkeypatch):
|
||||
"""Test dry-run path with mocked Gitea data — checks summary generation."""
|
||||
# Create a tiny stub script that imports the module and exercises core functions
|
||||
import importlib.util
|
||||
spec = importlib.util.spec_from_file_location("load_cap_enforcer", SCRIPT)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
# Load but don't execute main yet — just verify module structure
|
||||
# We'll parse the module source for expected symbols
|
||||
source = SCRIPT.read_text()
|
||||
assert "fetch_all_open_issues" in source
|
||||
assert "build_summary" in source
|
||||
assert "unassignment_map" in source
|
||||
assert "COMMENT_TEMPLATE" in source
|
||||
assert "Unassigned from @{assignee} due to load cap" in source
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Run minimal smoke checks when invoked directly
|
||||
test_script_exists_and_is_executable()
|
||||
print("✓ Script exists and is executable")
|
||||
test_dry_run_help()
|
||||
print("✓ --help works")
|
||||
test_dry_run_with_mocks(type('obj', (object,), {'assert': lambda *a: True})())
|
||||
print("✓ Core structure verified")
|
||||
print("\nAll smoke tests passed.")
|
||||
|
||||
@@ -1,103 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Tests for claim_annotator.py — verifies source distinction is present."""
|
||||
|
||||
import sys
|
||||
import os
|
||||
import json
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src"))
|
||||
|
||||
from timmy.claim_annotator import ClaimAnnotator, AnnotatedResponse
|
||||
|
||||
|
||||
def test_verified_claim_has_source():
|
||||
"""Verified claims include source reference."""
|
||||
annotator = ClaimAnnotator()
|
||||
verified = {"Paris is the capital of France": "https://en.wikipedia.org/wiki/Paris"}
|
||||
response = "Paris is the capital of France. It is a beautiful city."
|
||||
|
||||
result = annotator.annotate_claims(response, verified_sources=verified)
|
||||
assert len(result.claims) > 0
|
||||
verified_claims = [c for c in result.claims if c.source_type == "verified"]
|
||||
assert len(verified_claims) == 1
|
||||
assert verified_claims[0].source_ref == "https://en.wikipedia.org/wiki/Paris"
|
||||
assert "[V]" in result.rendered_text
|
||||
assert "[source:" in result.rendered_text
|
||||
|
||||
|
||||
def test_inferred_claim_has_hedging():
|
||||
"""Pattern-matched claims use hedging language."""
|
||||
annotator = ClaimAnnotator()
|
||||
response = "The weather is nice today. It might rain tomorrow."
|
||||
|
||||
result = annotator.annotate_claims(response)
|
||||
inferred_claims = [c for c in result.claims if c.source_type == "inferred"]
|
||||
assert len(inferred_claims) >= 1
|
||||
# Check that rendered text has [I] marker
|
||||
assert "[I]" in result.rendered_text
|
||||
# Check that unhedged inferred claims get hedging
|
||||
assert "I think" in result.rendered_text or "I believe" in result.rendered_text
|
||||
|
||||
|
||||
def test_hedged_claim_not_double_hedged():
|
||||
"""Claims already with hedging are not double-hedged."""
|
||||
annotator = ClaimAnnotator()
|
||||
response = "I think the sky is blue. It is a nice day."
|
||||
|
||||
result = annotator.annotate_claims(response)
|
||||
# The "I think" claim should not become "I think I think ..."
|
||||
assert "I think I think" not in result.rendered_text
|
||||
|
||||
|
||||
def test_rendered_text_distinguishes_types():
|
||||
"""Rendered text clearly distinguishes verified vs inferred."""
|
||||
annotator = ClaimAnnotator()
|
||||
verified = {"Earth is round": "https://science.org/earth"}
|
||||
response = "Earth is round. Stars are far away."
|
||||
|
||||
result = annotator.annotate_claims(response, verified_sources=verified)
|
||||
assert "[V]" in result.rendered_text # verified marker
|
||||
assert "[I]" in result.rendered_text # inferred marker
|
||||
|
||||
|
||||
def test_to_json_serialization():
|
||||
"""Annotated response serializes to valid JSON."""
|
||||
annotator = ClaimAnnotator()
|
||||
response = "Test claim."
|
||||
result = annotator.annotate_claims(response)
|
||||
json_str = annotator.to_json(result)
|
||||
parsed = json.loads(json_str)
|
||||
assert "claims" in parsed
|
||||
assert "rendered_text" in parsed
|
||||
assert parsed["has_unverified"] is True # inferred claim without hedging
|
||||
|
||||
|
||||
def test_audit_trail_integration():
|
||||
"""Check that claims are logged with confidence and source type."""
|
||||
# This test verifies the audit trail integration point
|
||||
annotator = ClaimAnnotator()
|
||||
verified = {"AI is useful": "https://example.com/ai"}
|
||||
response = "AI is useful. It can help with tasks."
|
||||
|
||||
result = annotator.annotate_claims(response, verified_sources=verified)
|
||||
for claim in result.claims:
|
||||
assert claim.source_type in ("verified", "inferred")
|
||||
assert claim.confidence in ("high", "medium", "low", "unknown")
|
||||
if claim.source_type == "verified":
|
||||
assert claim.source_ref is not None
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_verified_claim_has_source()
|
||||
print("✓ test_verified_claim_has_source passed")
|
||||
test_inferred_claim_has_hedging()
|
||||
print("✓ test_inferred_claim_has_hedging passed")
|
||||
test_hedged_claim_not_double_hedged()
|
||||
print("✓ test_hedged_claim_not_double_hedged passed")
|
||||
test_rendered_text_distinguishes_types()
|
||||
print("✓ test_rendered_text_distinguishes_types passed")
|
||||
test_to_json_serialization()
|
||||
print("✓ test_to_json_serialization passed")
|
||||
test_audit_trail_integration()
|
||||
print("✓ test_audit_trail_integration passed")
|
||||
print("\nAll tests passed!")
|
||||
@@ -1,210 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Open-Load Cap Enforcement — Audit-B3
|
||||
|
||||
Scans multiple repos for open issues, enforces a per-agent open-issue cap,
|
||||
auto-unassigns overflow (oldest first), and posts a summary.
|
||||
|
||||
Acceptance (timmy-home #498):
|
||||
- Lives in timmy-config/bin/load_cap_enforcer.py
|
||||
- Scans timmy-home, timmy-config, the-nexus, hermes-agent
|
||||
- Cap: 25 open issues per agent (configurable)
|
||||
- Unassign oldest overflow, comment on each
|
||||
- Dry-run first, then live; summary posted on parent issue #495
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
from collections import defaultdict
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
# ── Configuration ─────────────────────────────────────────────────────────────
|
||||
GITEA_BASE = "https://forge.alexanderwhitestone.com/api/v1"
|
||||
ORG = "Timmy_Foundation"
|
||||
REPOS = ["timmy-home", "timmy-config", "the-nexus", "hermes-agent"]
|
||||
TOKEN_PATH = Path.home() / ".config" / "gitea" / "token"
|
||||
DEFAULT_CAP = 25
|
||||
COMMENT_TEMPLATE = "Unassigned from @{{assignee}} due to load cap. Available for pickup."
|
||||
|
||||
|
||||
def load_token() -> str:
|
||||
if TOKEN_PATH.exists():
|
||||
return TOKEN_PATH.read_text().strip()
|
||||
tok = os.environ.get("GITEA_TOKEN", "")
|
||||
if tok:
|
||||
return tok
|
||||
sys.exit("ERROR: Gitea token not found at ~/.config/gitea/token or GITEA_TOKEN env")
|
||||
|
||||
|
||||
def api(method: str, path: str, token: str, data=None):
|
||||
url = f"{GITEA_BASE}{path}"
|
||||
body = json.dumps(data).encode() if data else None
|
||||
headers = {"Authorization": f"token {token}"}
|
||||
if body:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=body, headers=headers, method=method)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||||
return json.loads(resp.read()), resp.status
|
||||
except urllib.error.HTTPError as e:
|
||||
err = e.read().decode() if e.fp else str(e)
|
||||
print(f" API {e.code}: {err}", file=sys.stderr)
|
||||
return None, e.code
|
||||
except Exception as e:
|
||||
print(f" Request error: {e}", file=sys.stderr)
|
||||
return None, None
|
||||
|
||||
|
||||
def fetch_all_open_issues(token: str):
|
||||
all_issues = []
|
||||
for repo in REPOS:
|
||||
page = 1
|
||||
while True:
|
||||
data, status = api("GET", f"/repos/{ORG}/{repo}/issues?state=open&page={page}&limit=50", token)
|
||||
if status != 200 or not data:
|
||||
break
|
||||
all_issues.extend(data)
|
||||
if len(data) < 50:
|
||||
break
|
||||
page += 1
|
||||
return all_issues
|
||||
|
||||
|
||||
def build_summary(by_agent: dict, unassignment_map: dict):
|
||||
lines = []
|
||||
lines.append("Agent | Before | After | Unassigned Count")
|
||||
lines.append("-" * 50)
|
||||
for agent in sorted(by_agent.keys()):
|
||||
before = by_agent[agent]["before"]
|
||||
after = by_agent[agent]["after"]
|
||||
unassigned = len(unassignment_map.get(agent, []))
|
||||
lines.append(f"@{agent} | {before} | {after} | {unassigned}")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Enforce open-issue load cap per agent")
|
||||
parser.add_argument("--dry-run", action="store_true", help="Report without making changes")
|
||||
parser.add_argument("--cap", type=int, default=DEFAULT_CAP, help=f"Max open issues per agent (default: {DEFAULT_CAP})")
|
||||
parser.add_argument("--output", type=str, default=None, help="Write summary to file")
|
||||
parser.add_argument("--comment-on", type=int, default=None, help="Post summary as comment on timmy-home issue N")
|
||||
args = parser.parse_args()
|
||||
|
||||
token = load_token()
|
||||
print(f"Fetching open issues from {', '.join(REPOS)} ...")
|
||||
issues = fetch_all_open_issues(token)
|
||||
print(f"Fetched {len(issues)} open issues.")
|
||||
|
||||
# Group by assignee
|
||||
by_agent = defaultdict(lambda: {"before": 0, "issues": []})
|
||||
for iss in issues:
|
||||
for a in (iss.get("assignees") or []):
|
||||
login = a.get("login")
|
||||
if login:
|
||||
by_agent[login]["issues"].append(iss)
|
||||
by_agent[login]["before"] += 1
|
||||
|
||||
print(f"\nAgents with open issues: {list(by_agent.keys())}")
|
||||
for agent, d in sorted(by_agent.items()):
|
||||
print(f" @{agent}: {d['before']} issues")
|
||||
|
||||
# Identify overflow
|
||||
unassignment_map = defaultdict(list)
|
||||
for agent, d in by_agent.items():
|
||||
count = d["before"]
|
||||
if count > args.cap:
|
||||
overflow = count - args.cap
|
||||
issues_sorted = sorted(d["issues"], key=lambda i: i.get("created_at", ""))
|
||||
unassignment_map[agent] = issues_sorted[:overflow]
|
||||
print(f"\n@{agent} exceeds cap ({count} > {args.cap}); will unassign {overflow} oldest issue(s):")
|
||||
for iss in issues_sorted[:overflow]:
|
||||
print(f" - #{iss['number']}: {iss.get('title', '')[:50]}")
|
||||
|
||||
# Dry-run: just show summary and exit
|
||||
if args.dry_run:
|
||||
print("\n=== DRY RUN — no changes made ===")
|
||||
# For dry-run, after = before (no changes)
|
||||
for agent in by_agent:
|
||||
by_agent[agent]["after"] = by_agent[agent]["before"]
|
||||
summary = build_summary(by_agent, unassignment_map)
|
||||
print("\n" + summary)
|
||||
if args.output:
|
||||
Path(args.output).write_text(summary)
|
||||
print(f"\nSummary written to {args.output}")
|
||||
return 0
|
||||
|
||||
# LIVE: perform unassignments and comments (concurrent)
|
||||
print("\n=== LIVE RUN — executing ===")
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
import threading
|
||||
lock = threading.Lock()
|
||||
tasks = []
|
||||
for agent, issues_to_unassign in unassignment_map.items():
|
||||
for iss in issues_to_unassign:
|
||||
issue_num = iss["number"]
|
||||
repo_name = next(
|
||||
(r for r in REPOS if f"/{r}/issues/" in iss.get("html_url", "")), REPOS[0]
|
||||
)
|
||||
tasks.append((agent, issue_num, repo_name, iss))
|
||||
print(f"Total unassignment tasks: {len(tasks)}")
|
||||
def do_task(agent, issue_num, repo_name, iss):
|
||||
# Unassign
|
||||
_, status1 = api("PATCH", f"/repos/{ORG}/{repo_name}/issues/{issue_num}", token, {"assignees": []})
|
||||
if status1 not in (200, 201, 204):
|
||||
return (agent, issue_num, repo_name, False, f"unassign HTTP {status1}")
|
||||
# Comment
|
||||
comment_body = COMMENT_TEMPLATE.format(assignee=agent)
|
||||
_, status2 = api("POST", f"/repos/{ORG}/{repo_name}/issues/{issue_num}/comments", token, {"body": comment_body})
|
||||
if status2 not in (200, 201):
|
||||
return (agent, issue_num, repo_name, True, f"unassigned but comment HTTP {status2}")
|
||||
return (agent, issue_num, repo_name, True, "OK")
|
||||
completed = 0
|
||||
with ThreadPoolExecutor(max_workers=12) as executor:
|
||||
futures = [executor.submit(do_task, a, n, r, i) for (a, n, r, i) in tasks]
|
||||
for fut in as_completed(futures):
|
||||
agent, num, repo, ok, msg = fut.result()
|
||||
with lock:
|
||||
completed += 1
|
||||
if completed % 50 == 0:
|
||||
print(f" Progress: {completed}/{len(tasks)}")
|
||||
if ok:
|
||||
print(f" ✓ #{num} ({repo})")
|
||||
else:
|
||||
print(f" ✗ #{num} ({repo}): {msg}")
|
||||
|
||||
# Recompute after counts for summary
|
||||
print("\nRecomputing after counts ...")
|
||||
after_issues = fetch_all_open_issues(token)
|
||||
by_agent_after = defaultdict(int)
|
||||
for iss in after_issues:
|
||||
for a in (iss.get("assignees") or []):
|
||||
by_agent_after[a.get("login")] += 1
|
||||
for agent in by_agent:
|
||||
by_agent[agent]["after"] = by_agent_after.get(agent, 0)
|
||||
|
||||
summary = build_summary(by_agent, unassignment_map)
|
||||
print("\n=== SUMMARY ===")
|
||||
print(summary)
|
||||
|
||||
if args.output:
|
||||
Path(args.output).write_text(summary)
|
||||
print(f"Summary written to {args.output}")
|
||||
|
||||
if args.comment_on:
|
||||
body = f"Open-load cap enforcement run (cap={args.cap}):\n\n```\n{summary}\n```"
|
||||
_, status = api("POST", f"/repos/{ORG}/timmy-home/issues/{args.comment_on}/comments", token, {"body": body})
|
||||
if status in (200, 201):
|
||||
print(f"\nSummary posted as comment on timmy-home issue #{args.comment_on}")
|
||||
else:
|
||||
print(f"\nWARNING: failed to post comment (HTTP {status})")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user