Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9312e4dbee |
97
.gitea/workflows/agent-pr-gate.yml
Normal file
97
.gitea/workflows/agent-pr-gate.yml
Normal file
@@ -0,0 +1,97 @@
|
||||
name: Agent PR Gate
|
||||
'on':
|
||||
pull_request:
|
||||
branches: [main]
|
||||
|
||||
jobs:
|
||||
gate:
|
||||
runs-on: ubuntu-latest
|
||||
outputs:
|
||||
syntax_status: ${{ steps.syntax.outcome }}
|
||||
tests_status: ${{ steps.tests.outcome }}
|
||||
criteria_status: ${{ steps.criteria.outcome }}
|
||||
risk_level: ${{ steps.risk.outputs.level }}
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
with:
|
||||
fetch-depth: 0
|
||||
|
||||
- uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: '3.11'
|
||||
|
||||
- name: Install CI dependencies
|
||||
run: |
|
||||
python3 -m pip install --quiet pyyaml pytest
|
||||
|
||||
- id: risk
|
||||
name: Classify PR risk
|
||||
run: |
|
||||
BASE_REF="${GITHUB_BASE_REF:-main}"
|
||||
git fetch origin "$BASE_REF" --depth 1
|
||||
git diff --name-only "origin/$BASE_REF"...HEAD > /tmp/changed_files.txt
|
||||
python3 scripts/agent_pr_gate.py classify-risk --files-file /tmp/changed_files.txt > /tmp/risk.json
|
||||
python3 - <<'PY'
|
||||
import json, os
|
||||
with open('/tmp/risk.json', 'r', encoding='utf-8') as fh:
|
||||
data = json.load(fh)
|
||||
with open(os.environ['GITHUB_OUTPUT'], 'a', encoding='utf-8') as fh:
|
||||
fh.write('level=' + data['risk'] + '\n')
|
||||
PY
|
||||
|
||||
- id: syntax
|
||||
name: Syntax and parse checks
|
||||
continue-on-error: true
|
||||
run: |
|
||||
find . \( -name '*.yml' -o -name '*.yaml' \) | grep -v .gitea | xargs -r python3 -c "import sys,yaml; [yaml.safe_load(open(f)) for f in sys.argv[1:]]"
|
||||
find . -name '*.json' | while read f; do python3 -m json.tool "$f" > /dev/null || exit 1; done
|
||||
find . -name '*.py' | xargs -r python3 -m py_compile
|
||||
find . -name '*.sh' | xargs -r bash -n
|
||||
|
||||
- id: tests
|
||||
name: Test suite
|
||||
continue-on-error: true
|
||||
run: |
|
||||
pytest -q --ignore=uni-wizard/v2/tests/test_author_whitelist.py
|
||||
|
||||
- id: criteria
|
||||
name: PR criteria verification
|
||||
continue-on-error: true
|
||||
run: |
|
||||
python3 scripts/agent_pr_gate.py validate-pr --event-path "$GITHUB_EVENT_PATH"
|
||||
|
||||
- name: Fail gate if any required check failed
|
||||
if: steps.syntax.outcome != 'success' || steps.tests.outcome != 'success' || steps.criteria.outcome != 'success'
|
||||
run: exit 1
|
||||
|
||||
report:
|
||||
needs: gate
|
||||
if: always()
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: '3.11'
|
||||
|
||||
- name: Post PR gate report
|
||||
env:
|
||||
GITEA_TOKEN: ${{ github.token }}
|
||||
run: |
|
||||
python3 scripts/agent_pr_gate.py comment \
|
||||
--event-path "$GITHUB_EVENT_PATH" \
|
||||
--token "$GITEA_TOKEN" \
|
||||
--syntax "${{ needs.gate.outputs.syntax_status }}" \
|
||||
--tests "${{ needs.gate.outputs.tests_status }}" \
|
||||
--criteria "${{ needs.gate.outputs.criteria_status }}" \
|
||||
--risk "${{ needs.gate.outputs.risk_level }}"
|
||||
|
||||
- name: Auto-merge low-risk clean PRs
|
||||
if: needs.gate.result == 'success' && needs.gate.outputs.risk_level == 'low'
|
||||
env:
|
||||
GITEA_TOKEN: ${{ github.token }}
|
||||
run: |
|
||||
python3 scripts/agent_pr_gate.py merge \
|
||||
--event-path "$GITHUB_EVENT_PATH" \
|
||||
--token "$GITEA_TOKEN"
|
||||
@@ -1,5 +1,5 @@
|
||||
name: Smoke Test
|
||||
on:
|
||||
'on':
|
||||
pull_request:
|
||||
push:
|
||||
branches: [main]
|
||||
@@ -11,10 +11,13 @@ jobs:
|
||||
- uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: '3.11'
|
||||
- name: Install parse dependencies
|
||||
run: |
|
||||
python3 -m pip install --quiet pyyaml
|
||||
- name: Parse check
|
||||
run: |
|
||||
find . -name '*.yml' -o -name '*.yaml' | grep -v .gitea | xargs -r python3 -c "import sys,yaml; [yaml.safe_load(open(f)) for f in sys.argv[1:]]"
|
||||
find . -name '*.json' | xargs -r python3 -m json.tool > /dev/null
|
||||
find . \( -name '*.yml' -o -name '*.yaml' \) | grep -v .gitea | xargs -r python3 -c "import sys,yaml; [yaml.safe_load(open(f)) for f in sys.argv[1:]]"
|
||||
find . -name '*.json' | while read f; do python3 -m json.tool "$f" > /dev/null || exit 1; done
|
||||
find . -name '*.py' | xargs -r python3 -m py_compile
|
||||
find . -name '*.sh' | xargs -r bash -n
|
||||
echo "PASS: All files parse"
|
||||
|
||||
@@ -1,61 +0,0 @@
|
||||
# [PHASE-1] Survival - Keep the Lights On
|
||||
|
||||
Phase 1 is the manual-clicker stage of the fleet. The machines exist. The services exist. The human is still the automation loop.
|
||||
|
||||
## Phase Definition
|
||||
|
||||
- Current state: fleet exists, agents run, everything important still depends on human vigilance.
|
||||
- Resources tracked here: Capacity, Uptime.
|
||||
- Next phase: [PHASE-2] Automation - Self-Healing Infrastructure
|
||||
|
||||
## Current Buildings
|
||||
|
||||
- VPS hosts: Ezra, Allegro, Bezalel
|
||||
- Agents: Timmy harness, Code Claw heartbeat, Gemini AI Studio worker
|
||||
- Gitea forge
|
||||
- Evennia worlds
|
||||
|
||||
## Current Resource Snapshot
|
||||
|
||||
- Fleet operational: yes
|
||||
- Uptime baseline: 0.0%
|
||||
- Days at or above 95% uptime: 0
|
||||
- Capacity utilization: 0.0%
|
||||
|
||||
## Next Phase Trigger
|
||||
|
||||
To unlock [PHASE-2] Automation - Self-Healing Infrastructure, the fleet must hold both of these conditions at once:
|
||||
- Uptime >= 95% for 30 consecutive days
|
||||
- Capacity utilization > 60%
|
||||
- Current trigger state: NOT READY
|
||||
|
||||
## Missing Requirements
|
||||
|
||||
- Uptime 0.0% / 95.0%
|
||||
- Days at or above 95% uptime: 0/30
|
||||
- Capacity utilization 0.0% / >60.0%
|
||||
|
||||
## Manual Clicker Interpretation
|
||||
|
||||
Paperclips analogy: Phase 1 = Manual clicker. You ARE the automation.
|
||||
Every restart, every SSH, every check is a manual click.
|
||||
|
||||
## Manual Clicks Still Required
|
||||
|
||||
- Restart agents and services by hand when a node goes dark.
|
||||
- SSH into machines to verify health, disk, and memory.
|
||||
- Check Gitea, relay, and world services manually before and after changes.
|
||||
- Act as the scheduler when automation is missing or only partially wired.
|
||||
|
||||
## Repo Signals Already Present
|
||||
|
||||
- `scripts/fleet_health_probe.sh` — Automated health probe exists and can supply the uptime baseline for the next phase.
|
||||
- `scripts/fleet_milestones.py` — Milestone tracker exists, so survival achievements can be narrated and logged.
|
||||
- `scripts/auto_restart_agent.sh` — Auto-restart tooling already exists as phase-2 groundwork.
|
||||
- `scripts/backup_pipeline.sh` — Backup pipeline scaffold exists for post-survival automation work.
|
||||
- `infrastructure/timmy-bridge/reports/generate_report.py` — Bridge reporting exists and can summarize heartbeat-driven uptime.
|
||||
|
||||
## Notes
|
||||
|
||||
- The fleet is alive, but the human is still the control loop.
|
||||
- Phase 1 is about naming reality plainly so later automation has a baseline to beat.
|
||||
@@ -12,7 +12,6 @@ Quick-reference index for common operational tasks across the Timmy Foundation i
|
||||
| Check fleet health | fleet-ops | `python3 scripts/fleet_readiness.py` |
|
||||
| Agent scorecard | fleet-ops | `python3 scripts/agent_scorecard.py` |
|
||||
| View fleet manifest | fleet-ops | `cat manifest.yaml` |
|
||||
| Render Phase-1 survival report | timmy-home | `python3 scripts/fleet_phase_status.py --output docs/FLEET_PHASE_1_SURVIVAL.md` |
|
||||
|
||||
## the-nexus (Frontend + Brain)
|
||||
|
||||
|
||||
191
scripts/agent_pr_gate.py
Executable file
191
scripts/agent_pr_gate.py
Executable file
@@ -0,0 +1,191 @@
|
||||
#!/usr/bin/env python3
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
API_BASE = "https://forge.alexanderwhitestone.com/api/v1"
|
||||
LOW_RISK_PREFIXES = (
|
||||
'docs/', 'reports/', 'notes/', 'tickets/', 'research/', 'briefings/',
|
||||
'twitter-archive/notes/', 'tests/'
|
||||
)
|
||||
LOW_RISK_SUFFIXES = {'.md', '.txt', '.jsonl'}
|
||||
MEDIUM_RISK_PREFIXES = ('.gitea/workflows/',)
|
||||
HIGH_RISK_PREFIXES = (
|
||||
'scripts/', 'deploy/', 'infrastructure/', 'metrics/', 'heartbeat/',
|
||||
'wizards/', 'evennia/', 'uniwizard/', 'uni-wizard/', 'timmy-local/',
|
||||
'evolution/'
|
||||
)
|
||||
HIGH_RISK_SUFFIXES = {'.py', '.sh', '.ini', '.service'}
|
||||
|
||||
|
||||
def read_changed_files(path):
|
||||
return [line.strip() for line in Path(path).read_text(encoding='utf-8').splitlines() if line.strip()]
|
||||
|
||||
|
||||
def classify_risk(files):
|
||||
if not files:
|
||||
return 'high'
|
||||
level = 'low'
|
||||
for file_path in files:
|
||||
path = file_path.strip()
|
||||
suffix = Path(path).suffix.lower()
|
||||
if path.startswith(LOW_RISK_PREFIXES):
|
||||
continue
|
||||
if path.startswith(HIGH_RISK_PREFIXES) or suffix in HIGH_RISK_SUFFIXES:
|
||||
return 'high'
|
||||
if path.startswith(MEDIUM_RISK_PREFIXES):
|
||||
level = 'medium'
|
||||
continue
|
||||
if path.startswith(LOW_RISK_PREFIXES) or suffix in LOW_RISK_SUFFIXES:
|
||||
continue
|
||||
level = 'high'
|
||||
return level
|
||||
|
||||
|
||||
def validate_pr_body(title, body):
|
||||
details = []
|
||||
combined = f"{title}\n{body}".strip()
|
||||
if not re.search(r'#\d+', combined):
|
||||
details.append('PR body/title must include an issue reference like #562.')
|
||||
if not re.search(r'(^|\n)\s*(verification|tests?)\s*:', body, re.IGNORECASE):
|
||||
details.append('PR body must include a Verification: section.')
|
||||
return (len(details) == 0, details)
|
||||
|
||||
|
||||
def build_comment_body(syntax_status, tests_status, criteria_status, risk_level):
|
||||
statuses = {
|
||||
'syntax': syntax_status,
|
||||
'tests': tests_status,
|
||||
'criteria': criteria_status,
|
||||
}
|
||||
all_clean = all(value == 'success' for value in statuses.values())
|
||||
action = 'auto-merge' if all_clean and risk_level == 'low' else 'human review'
|
||||
lines = [
|
||||
'## Agent PR Gate',
|
||||
'',
|
||||
'| Check | Status |',
|
||||
'|-------|--------|',
|
||||
f"| Syntax / parse | {syntax_status} |",
|
||||
f"| Test suite | {tests_status} |",
|
||||
f"| PR criteria | {criteria_status} |",
|
||||
f"| Risk level | {risk_level} |",
|
||||
'',
|
||||
]
|
||||
failed = [name for name, value in statuses.items() if value != 'success']
|
||||
if failed:
|
||||
lines.append('### Failure details')
|
||||
for name in failed:
|
||||
lines.append(f'- {name} reported failure. Inspect the workflow logs for that step.')
|
||||
else:
|
||||
lines.append('All automated checks passed.')
|
||||
lines.extend([
|
||||
'',
|
||||
f'Recommendation: {action}.',
|
||||
'Low-risk documentation/test-only PRs may be auto-merged. Operational changes stay in human review.',
|
||||
])
|
||||
return '\n'.join(lines)
|
||||
|
||||
|
||||
def _read_event(event_path):
|
||||
data = json.loads(Path(event_path).read_text(encoding='utf-8'))
|
||||
pr = data.get('pull_request') or {}
|
||||
repo = (data.get('repository') or {}).get('full_name') or os.environ.get('GITHUB_REPOSITORY')
|
||||
pr_number = pr.get('number') or data.get('number')
|
||||
title = pr.get('title') or ''
|
||||
body = pr.get('body') or ''
|
||||
return repo, pr_number, title, body
|
||||
|
||||
|
||||
def _request_json(method, url, token, payload=None):
|
||||
data = None if payload is None else json.dumps(payload).encode('utf-8')
|
||||
headers = {'Authorization': f'token {token}', 'Content-Type': 'application/json'}
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||||
return json.loads(resp.read().decode('utf-8'))
|
||||
|
||||
|
||||
def post_comment(repo, pr_number, token, body):
|
||||
url = f'{API_BASE}/repos/{repo}/issues/{pr_number}/comments'
|
||||
return _request_json('POST', url, token, {'body': body})
|
||||
|
||||
|
||||
def merge_pr(repo, pr_number, token):
|
||||
url = f'{API_BASE}/repos/{repo}/pulls/{pr_number}/merge'
|
||||
return _request_json('POST', url, token, {'Do': 'merge'})
|
||||
|
||||
|
||||
def cmd_classify_risk(args):
|
||||
files = list(args.files or [])
|
||||
if args.files_file:
|
||||
files.extend(read_changed_files(args.files_file))
|
||||
print(json.dumps({'risk': classify_risk(files), 'files': files}, indent=2))
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_validate_pr(args):
|
||||
_, _, title, body = _read_event(args.event_path)
|
||||
ok, details = validate_pr_body(title, body)
|
||||
if ok:
|
||||
print('PR body validation passed.')
|
||||
return 0
|
||||
for detail in details:
|
||||
print(detail)
|
||||
return 1
|
||||
|
||||
|
||||
def cmd_comment(args):
|
||||
repo, pr_number, _, _ = _read_event(args.event_path)
|
||||
body = build_comment_body(args.syntax, args.tests, args.criteria, args.risk)
|
||||
post_comment(repo, pr_number, args.token, body)
|
||||
print(f'Commented on PR #{pr_number} in {repo}.')
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_merge(args):
|
||||
repo, pr_number, _, _ = _read_event(args.event_path)
|
||||
merge_pr(repo, pr_number, args.token)
|
||||
print(f'Merged PR #{pr_number} in {repo}.')
|
||||
return 0
|
||||
|
||||
|
||||
def build_parser():
|
||||
parser = argparse.ArgumentParser(description='Agent PR CI helpers for timmy-home.')
|
||||
sub = parser.add_subparsers(dest='command', required=True)
|
||||
|
||||
classify = sub.add_parser('classify-risk')
|
||||
classify.add_argument('--files-file')
|
||||
classify.add_argument('files', nargs='*')
|
||||
classify.set_defaults(func=cmd_classify_risk)
|
||||
|
||||
validate = sub.add_parser('validate-pr')
|
||||
validate.add_argument('--event-path', required=True)
|
||||
validate.set_defaults(func=cmd_validate_pr)
|
||||
|
||||
comment = sub.add_parser('comment')
|
||||
comment.add_argument('--event-path', required=True)
|
||||
comment.add_argument('--token', required=True)
|
||||
comment.add_argument('--syntax', required=True)
|
||||
comment.add_argument('--tests', required=True)
|
||||
comment.add_argument('--criteria', required=True)
|
||||
comment.add_argument('--risk', required=True)
|
||||
comment.set_defaults(func=cmd_comment)
|
||||
|
||||
merge = sub.add_parser('merge')
|
||||
merge.add_argument('--event-path', required=True)
|
||||
merge.add_argument('--token', required=True)
|
||||
merge.set_defaults(func=cmd_merge)
|
||||
return parser
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
parser = build_parser()
|
||||
args = parser.parse_args(argv)
|
||||
return args.func(args)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
@@ -1,224 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Render the current fleet survival phase as a durable report."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from copy import deepcopy
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
PHASE_NAME = "[PHASE-1] Survival - Keep the Lights On"
|
||||
NEXT_PHASE_NAME = "[PHASE-2] Automation - Self-Healing Infrastructure"
|
||||
TARGET_UPTIME_PERCENT = 95.0
|
||||
TARGET_UPTIME_DAYS = 30
|
||||
TARGET_CAPACITY_PERCENT = 60.0
|
||||
|
||||
DEFAULT_BUILDINGS = [
|
||||
"VPS hosts: Ezra, Allegro, Bezalel",
|
||||
"Agents: Timmy harness, Code Claw heartbeat, Gemini AI Studio worker",
|
||||
"Gitea forge",
|
||||
"Evennia worlds",
|
||||
]
|
||||
|
||||
DEFAULT_MANUAL_CLICKS = [
|
||||
"Restart agents and services by hand when a node goes dark.",
|
||||
"SSH into machines to verify health, disk, and memory.",
|
||||
"Check Gitea, relay, and world services manually before and after changes.",
|
||||
"Act as the scheduler when automation is missing or only partially wired.",
|
||||
]
|
||||
|
||||
REPO_SIGNAL_FILES = {
|
||||
"scripts/fleet_health_probe.sh": "Automated health probe exists and can supply the uptime baseline for the next phase.",
|
||||
"scripts/fleet_milestones.py": "Milestone tracker exists, so survival achievements can be narrated and logged.",
|
||||
"scripts/auto_restart_agent.sh": "Auto-restart tooling already exists as phase-2 groundwork.",
|
||||
"scripts/backup_pipeline.sh": "Backup pipeline scaffold exists for post-survival automation work.",
|
||||
"infrastructure/timmy-bridge/reports/generate_report.py": "Bridge reporting exists and can summarize heartbeat-driven uptime.",
|
||||
}
|
||||
|
||||
DEFAULT_SNAPSHOT = {
|
||||
"fleet_operational": True,
|
||||
"resources": {
|
||||
"uptime_percent": 0.0,
|
||||
"days_at_or_above_95_percent": 0,
|
||||
"capacity_utilization_percent": 0.0,
|
||||
},
|
||||
"current_buildings": DEFAULT_BUILDINGS,
|
||||
"manual_clicks": DEFAULT_MANUAL_CLICKS,
|
||||
"notes": [
|
||||
"The fleet is alive, but the human is still the control loop.",
|
||||
"Phase 1 is about naming reality plainly so later automation has a baseline to beat.",
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def default_snapshot() -> dict[str, Any]:
|
||||
return deepcopy(DEFAULT_SNAPSHOT)
|
||||
|
||||
|
||||
def _deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
|
||||
result = deepcopy(base)
|
||||
for key, value in override.items():
|
||||
if isinstance(value, dict) and isinstance(result.get(key), dict):
|
||||
result[key] = _deep_merge(result[key], value)
|
||||
else:
|
||||
result[key] = value
|
||||
return result
|
||||
|
||||
|
||||
def load_snapshot(snapshot_path: Path | None = None) -> dict[str, Any]:
|
||||
snapshot = default_snapshot()
|
||||
if snapshot_path is None:
|
||||
return snapshot
|
||||
override = json.loads(snapshot_path.read_text(encoding="utf-8"))
|
||||
return _deep_merge(snapshot, override)
|
||||
|
||||
|
||||
def collect_repo_signals(repo_root: Path) -> list[str]:
|
||||
signals: list[str] = []
|
||||
for rel_path, description in REPO_SIGNAL_FILES.items():
|
||||
if (repo_root / rel_path).exists():
|
||||
signals.append(f"`{rel_path}` — {description}")
|
||||
return signals
|
||||
|
||||
|
||||
def compute_phase_status(snapshot: dict[str, Any], repo_root: Path | None = None) -> dict[str, Any]:
|
||||
repo_root = repo_root or Path(__file__).resolve().parents[1]
|
||||
resources = snapshot.get("resources", {})
|
||||
uptime_percent = float(resources.get("uptime_percent", 0.0))
|
||||
uptime_days = int(resources.get("days_at_or_above_95_percent", 0))
|
||||
capacity_percent = float(resources.get("capacity_utilization_percent", 0.0))
|
||||
fleet_operational = bool(snapshot.get("fleet_operational", False))
|
||||
|
||||
missing: list[str] = []
|
||||
if not fleet_operational:
|
||||
missing.append("Fleet operational flag is false.")
|
||||
if uptime_percent < TARGET_UPTIME_PERCENT:
|
||||
missing.append(f"Uptime {uptime_percent:.1f}% / {TARGET_UPTIME_PERCENT:.1f}%")
|
||||
if uptime_days < TARGET_UPTIME_DAYS:
|
||||
missing.append(f"Days at or above 95% uptime: {uptime_days}/{TARGET_UPTIME_DAYS}")
|
||||
if capacity_percent <= TARGET_CAPACITY_PERCENT:
|
||||
missing.append(f"Capacity utilization {capacity_percent:.1f}% / >{TARGET_CAPACITY_PERCENT:.1f}%")
|
||||
|
||||
return {
|
||||
"title": PHASE_NAME,
|
||||
"current_phase": "PHASE-1 Survival",
|
||||
"fleet_operational": fleet_operational,
|
||||
"resources": {
|
||||
"uptime_percent": uptime_percent,
|
||||
"days_at_or_above_95_percent": uptime_days,
|
||||
"capacity_utilization_percent": capacity_percent,
|
||||
},
|
||||
"current_buildings": list(snapshot.get("current_buildings", DEFAULT_BUILDINGS)),
|
||||
"manual_clicks": list(snapshot.get("manual_clicks", DEFAULT_MANUAL_CLICKS)),
|
||||
"notes": list(snapshot.get("notes", [])),
|
||||
"repo_signals": collect_repo_signals(repo_root),
|
||||
"next_phase": NEXT_PHASE_NAME,
|
||||
"next_phase_ready": fleet_operational and not missing,
|
||||
"missing_requirements": missing,
|
||||
}
|
||||
|
||||
|
||||
def render_markdown(status: dict[str, Any]) -> str:
|
||||
resources = status["resources"]
|
||||
missing = status["missing_requirements"]
|
||||
ready_line = "READY" if status["next_phase_ready"] else "NOT READY"
|
||||
|
||||
lines = [
|
||||
f"# {status['title']}",
|
||||
"",
|
||||
"Phase 1 is the manual-clicker stage of the fleet. The machines exist. The services exist. The human is still the automation loop.",
|
||||
"",
|
||||
"## Phase Definition",
|
||||
"",
|
||||
"- Current state: fleet exists, agents run, everything important still depends on human vigilance.",
|
||||
"- Resources tracked here: Capacity, Uptime.",
|
||||
f"- Next phase: {status['next_phase']}",
|
||||
"",
|
||||
"## Current Buildings",
|
||||
"",
|
||||
]
|
||||
lines.extend(f"- {item}" for item in status["current_buildings"])
|
||||
|
||||
lines.extend([
|
||||
"",
|
||||
"## Current Resource Snapshot",
|
||||
"",
|
||||
f"- Fleet operational: {'yes' if status['fleet_operational'] else 'no'}",
|
||||
f"- Uptime baseline: {resources['uptime_percent']:.1f}%",
|
||||
f"- Days at or above 95% uptime: {resources['days_at_or_above_95_percent']}",
|
||||
f"- Capacity utilization: {resources['capacity_utilization_percent']:.1f}%",
|
||||
"",
|
||||
"## Next Phase Trigger",
|
||||
"",
|
||||
f"To unlock {status['next_phase']}, the fleet must hold both of these conditions at once:",
|
||||
f"- Uptime >= {TARGET_UPTIME_PERCENT:.0f}% for {TARGET_UPTIME_DAYS} consecutive days",
|
||||
f"- Capacity utilization > {TARGET_CAPACITY_PERCENT:.0f}%",
|
||||
f"- Current trigger state: {ready_line}",
|
||||
"",
|
||||
"## Missing Requirements",
|
||||
"",
|
||||
])
|
||||
if missing:
|
||||
lines.extend(f"- {item}" for item in missing)
|
||||
else:
|
||||
lines.append("- None. Phase 2 can unlock now.")
|
||||
|
||||
lines.extend([
|
||||
"",
|
||||
"## Manual Clicker Interpretation",
|
||||
"",
|
||||
"Paperclips analogy: Phase 1 = Manual clicker. You ARE the automation.",
|
||||
"Every restart, every SSH, every check is a manual click.",
|
||||
"",
|
||||
"## Manual Clicks Still Required",
|
||||
"",
|
||||
])
|
||||
lines.extend(f"- {item}" for item in status["manual_clicks"])
|
||||
|
||||
lines.extend([
|
||||
"",
|
||||
"## Repo Signals Already Present",
|
||||
"",
|
||||
])
|
||||
if status["repo_signals"]:
|
||||
lines.extend(f"- {item}" for item in status["repo_signals"])
|
||||
else:
|
||||
lines.append("- No survival-adjacent repo signals detected.")
|
||||
|
||||
if status["notes"]:
|
||||
lines.extend(["", "## Notes", ""])
|
||||
lines.extend(f"- {item}" for item in status["notes"])
|
||||
|
||||
return "\n".join(lines).rstrip() + "\n"
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="Render the fleet phase-1 survival report")
|
||||
parser.add_argument("--snapshot", help="Optional JSON snapshot overriding the default phase-1 baseline")
|
||||
parser.add_argument("--output", help="Write markdown report to this path")
|
||||
parser.add_argument("--json", action="store_true", help="Print computed status as JSON instead of markdown")
|
||||
args = parser.parse_args()
|
||||
|
||||
snapshot = load_snapshot(Path(args.snapshot).expanduser() if args.snapshot else None)
|
||||
repo_root = Path(__file__).resolve().parents[1]
|
||||
status = compute_phase_status(snapshot, repo_root=repo_root)
|
||||
|
||||
if args.json:
|
||||
rendered = json.dumps(status, indent=2)
|
||||
else:
|
||||
rendered = render_markdown(status)
|
||||
|
||||
if args.output:
|
||||
output_path = Path(args.output).expanduser()
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
output_path.write_text(rendered, encoding="utf-8")
|
||||
print(f"Phase status written to {output_path}")
|
||||
else:
|
||||
print(rendered)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
68
tests/test_agent_pr_gate.py
Normal file
68
tests/test_agent_pr_gate.py
Normal file
@@ -0,0 +1,68 @@
|
||||
import pathlib
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
ROOT = pathlib.Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / 'scripts'))
|
||||
|
||||
import agent_pr_gate # noqa: E402
|
||||
|
||||
|
||||
class TestAgentPrGate(unittest.TestCase):
|
||||
def test_classify_risk_low_for_docs_and_tests_only(self):
|
||||
level = agent_pr_gate.classify_risk([
|
||||
'docs/runbook.md',
|
||||
'reports/daily-summary.md',
|
||||
'tests/test_agent_pr_gate.py',
|
||||
])
|
||||
self.assertEqual(level, 'low')
|
||||
|
||||
def test_classify_risk_high_for_operational_paths(self):
|
||||
level = agent_pr_gate.classify_risk([
|
||||
'scripts/failover_monitor.py',
|
||||
'deploy/playbook.yml',
|
||||
])
|
||||
self.assertEqual(level, 'high')
|
||||
|
||||
def test_validate_pr_body_requires_issue_ref_and_verification(self):
|
||||
ok, details = agent_pr_gate.validate_pr_body(
|
||||
'feat: add thing',
|
||||
'What changed only\n\nNo verification section here.'
|
||||
)
|
||||
self.assertFalse(ok)
|
||||
self.assertIn('issue reference', ' '.join(details).lower())
|
||||
self.assertIn('verification', ' '.join(details).lower())
|
||||
|
||||
def test_validate_pr_body_accepts_issue_ref_and_verification(self):
|
||||
ok, details = agent_pr_gate.validate_pr_body(
|
||||
'feat: add thing (#562)',
|
||||
'Refs #562\n\nVerification:\n- pytest -q\n'
|
||||
)
|
||||
self.assertTrue(ok)
|
||||
self.assertEqual(details, [])
|
||||
|
||||
def test_build_comment_body_reports_failures_and_human_review(self):
|
||||
body = agent_pr_gate.build_comment_body(
|
||||
syntax_status='success',
|
||||
tests_status='failure',
|
||||
criteria_status='success',
|
||||
risk_level='high',
|
||||
)
|
||||
self.assertIn('tests', body.lower())
|
||||
self.assertIn('failure', body.lower())
|
||||
self.assertIn('human review', body.lower())
|
||||
|
||||
def test_changed_files_file_loader_ignores_blanks(self):
|
||||
with tempfile.NamedTemporaryFile('w+', delete=False) as handle:
|
||||
handle.write('docs/one.md\n\nreports/two.md\n')
|
||||
path = handle.name
|
||||
try:
|
||||
files = agent_pr_gate.read_changed_files(path)
|
||||
finally:
|
||||
pathlib.Path(path).unlink(missing_ok=True)
|
||||
self.assertEqual(files, ['docs/one.md', 'reports/two.md'])
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
24
tests/test_agent_pr_workflow.py
Normal file
24
tests/test_agent_pr_workflow.py
Normal file
@@ -0,0 +1,24 @@
|
||||
import pathlib
|
||||
import unittest
|
||||
import yaml
|
||||
|
||||
ROOT = pathlib.Path(__file__).resolve().parents[1]
|
||||
WORKFLOW = ROOT / '.gitea' / 'workflows' / 'agent-pr-gate.yml'
|
||||
|
||||
|
||||
class TestAgentPrWorkflow(unittest.TestCase):
|
||||
def test_workflow_exists(self):
|
||||
self.assertTrue(WORKFLOW.exists(), 'agent-pr-gate workflow should exist')
|
||||
|
||||
def test_workflow_has_pr_gate_and_reporting_jobs(self):
|
||||
data = yaml.safe_load(WORKFLOW.read_text(encoding='utf-8'))
|
||||
self.assertIn('pull_request', data.get('on', {}))
|
||||
jobs = data.get('jobs', {})
|
||||
self.assertIn('gate', jobs)
|
||||
self.assertIn('report', jobs)
|
||||
report_steps = jobs['report']['steps']
|
||||
self.assertTrue(any('Auto-merge low-risk clean PRs' in (step.get('name') or '') for step in report_steps))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -1,67 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPT_PATH = ROOT / "scripts" / "fleet_phase_status.py"
|
||||
DOC_PATH = ROOT / "docs" / "FLEET_PHASE_1_SURVIVAL.md"
|
||||
|
||||
|
||||
def _load_module(path: Path, name: str):
|
||||
assert path.exists(), f"missing {path.relative_to(ROOT)}"
|
||||
spec = importlib.util.spec_from_file_location(name, path)
|
||||
assert spec and spec.loader
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def test_compute_phase_status_tracks_survival_gate_requirements() -> None:
|
||||
mod = _load_module(SCRIPT_PATH, "fleet_phase_status")
|
||||
|
||||
status = mod.compute_phase_status(
|
||||
{
|
||||
"fleet_operational": True,
|
||||
"resources": {
|
||||
"uptime_percent": 94.5,
|
||||
"days_at_or_above_95_percent": 12,
|
||||
"capacity_utilization_percent": 45.0,
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
assert status["current_phase"] == "PHASE-1 Survival"
|
||||
assert status["next_phase_ready"] is False
|
||||
assert any("94.5% / 95.0%" in item for item in status["missing_requirements"])
|
||||
assert any("12/30" in item for item in status["missing_requirements"])
|
||||
assert any("45.0% / >60.0%" in item for item in status["missing_requirements"])
|
||||
|
||||
|
||||
def test_render_markdown_preserves_phase_buildings_and_manual_clicker_language() -> None:
|
||||
mod = _load_module(SCRIPT_PATH, "fleet_phase_status")
|
||||
status = mod.compute_phase_status(mod.default_snapshot())
|
||||
report = mod.render_markdown(status)
|
||||
|
||||
for snippet in (
|
||||
"# [PHASE-1] Survival - Keep the Lights On",
|
||||
"VPS hosts: Ezra, Allegro, Bezalel",
|
||||
"Timmy harness",
|
||||
"Gitea forge",
|
||||
"Evennia worlds",
|
||||
"Every restart, every SSH, every check is a manual click.",
|
||||
):
|
||||
assert snippet in report
|
||||
|
||||
|
||||
def test_repo_contains_generated_phase_1_doc() -> None:
|
||||
assert DOC_PATH.exists(), "missing committed phase-1 survival doc"
|
||||
text = DOC_PATH.read_text(encoding="utf-8")
|
||||
for snippet in (
|
||||
"# [PHASE-1] Survival - Keep the Lights On",
|
||||
"## Current Buildings",
|
||||
"## Next Phase Trigger",
|
||||
"## Manual Clicker Interpretation",
|
||||
):
|
||||
assert snippet in text
|
||||
Reference in New Issue
Block a user