Compare commits

..

2 Commits

Author SHA1 Message Date
Alexander Whitestone
06a6b53a37 feat: add tower world-changing items for #514
Some checks failed
Agent PR Gate / gate (pull_request) Failing after 58s
Self-Healing Smoke / self-healing-smoke (pull_request) Failing after 26s
Agent PR Gate / report (pull_request) Successful in 23s
Smoke Test / smoke (pull_request) Failing after 27s
2026-04-22 02:54:56 -04:00
Alexander Whitestone
5ce4fa6cb8 test: define tower world items for #514 2026-04-22 02:43:29 -04:00
6 changed files with 203 additions and 672 deletions

View File

@@ -1,18 +0,0 @@
{
"owner": "Timmy_Foundation",
"repos": [
"timmy-home",
"timmy-config",
"fleet-ops",
"the-beacon",
"the-door",
"the-nexus"
],
"lookback_days": 14,
"alert": {
"recent_days": 7,
"baseline_days": 7,
"minimum_baseline_closed": 4,
"drop_ratio": 0.6
}
}

View File

@@ -1,70 +0,0 @@
# Burn-down Velocity Tracking
Refs #519.
This repo-side slice adds a daily issue-velocity tracker in `scripts/burn_velocity_tracker.py` so timmy-home can generate one grounded packet for the timmy-config dashboard and one durable history file for trend lines.
## What it emits
Daily run outputs:
- `~/.timmy/burn-velocity/latest.json` — machine-readable payload for the timmy-config dashboard
- `~/.timmy/burn-velocity/latest.md` — operator-facing markdown summary
- `~/.timmy/burn-velocity/history.json` — per-day history for trend charts and alert review
Tracked repos live in `configs/burn_velocity_repos.json`.
## Cron command
```bash
cd ~/timmy-home && \
python3 scripts/burn_velocity_tracker.py \
--config configs/burn_velocity_repos.json \
--output-json ~/.timmy/burn-velocity/latest.json \
--output-md ~/.timmy/burn-velocity/latest.md \
--history-file ~/.timmy/burn-velocity/history.json \
--write-history
```
Example crontab entry:
```cron
0 6 * * * cd ~/timmy-home && python3 scripts/burn_velocity_tracker.py --config configs/burn_velocity_repos.json --output-json ~/.timmy/burn-velocity/latest.json --output-md ~/.timmy/burn-velocity/latest.md --history-file ~/.timmy/burn-velocity/history.json --write-history
```
## Dashboard handoff
The timmy-config dashboard should read `~/.timmy/burn-velocity/latest.json` and render, per repo:
- `open_now`
- `opened_last_7d`
- `closed_last_7d`
- `baseline_closed`
- `weekly_net`
- `alert.status`
- `alert.kind`
- `alert.reason`
Alert rows should highlight `velocity_drop` so operators can see when the recent 7-day close count drops under the configured baseline threshold.
## Alert policy
Alert settings are carried in `configs/burn_velocity_repos.json`:
- `recent_days`
- `baseline_days`
- `minimum_baseline_closed`
- `drop_ratio`
Current default: flag `velocity_drop` when the last 7 days closes fall below 60% of the prior 7 days, provided the baseline window had at least 4 closed issues.
## Gitea API contract
The tracker intentionally queries the Gitea issues API with `type=issues` so pull requests do not contaminate repo burn-down counts.
Live collection shape:
- open backlog uses `/repos/{owner}/{repo}/issues?state=open&type=issues`
- recent event scan uses `/repos/{owner}/{repo}/issues?state=all&type=issues&since=...`
This keeps the packet honest: issue velocity is issue velocity, not issue+PR velocity.
## Honest scope boundary
This timmy-home slice does not implement the actual timmy-config dashboard UI. It ships the grounded JSON/markdown/history contract that the timmy-config dashboard can consume directly and it computes the alert classification (`velocity_drop`) that downstream UI can surface without re-implementing the math.

View File

@@ -12,6 +12,29 @@ WORLD_DIR = Path('/Users/apayne/.timmy/evennia/timmy_world')
STATE_FILE = WORLD_DIR / 'game_state.json'
TIMMY_LOG = WORLD_DIR / 'timmy_log.md'
WORLD_ITEMS = {
"foraged key": {"effect": "unlock_tower_cache", "quest_item": True, "consumable": False, "effect_text": "A hidden cache clicks open in the Tower wall."},
"seed packet": {"effect": "grow_garden", "quest_item": False, "consumable": True, "effect_text": "Fresh growth pushes through the Garden soil."},
"notebook": {"effect": "write_notebook_rule", "quest_item": False, "consumable": False, "effect_text": "A new rule joins the whiteboard in the Tower."},
"cloth": {"effect": "patch_bridge", "quest_item": False, "consumable": True, "effect_text": "The Bridge railing is wrapped tight against the weather."},
"oil can": {"effect": "stoke_forge", "quest_item": False, "consumable": True, "effect_text": "The Forge fire answers with a hotter glow."},
"lantern": {"effect": "light_bridge", "quest_item": False, "consumable": False, "effect_text": "A steady lantern glow cuts through the dark over the Bridge."},
"rope spool": {"effect": "secure_bridge", "quest_item": False, "consumable": True, "effect_text": "The Bridge is lashed tight and feels safer underfoot."},
"chalk": {"effect": "mark_threshold", "quest_item": False, "consumable": True, "effect_text": "A chalk mark at the Threshold points wanderers home."},
"weather vane": {"effect": "read_weather", "quest_item": False, "consumable": False, "effect_text": "The weather vane settles and the coming storm makes sense."},
"sunstone": {"effect": "restore_tower_power", "quest_item": False, "consumable": False, "effect_text": "Warm light races through the Tower circuits."},
"iron nails": {"effect": "reinforce_bridge", "quest_item": False, "consumable": True, "effect_text": "The Bridge planks are pinned down against the flood."},
"river stone": {"effect": "water_garden", "quest_item": False, "consumable": True, "effect_text": "Moisture returns to the Garden beds."},
}
ROOM_DISCOVERABLES = {
"Threshold": ["chalk", "sunstone"],
"Tower": ["notebook", "lantern"],
"Forge": ["oil can", "iron nails"],
"Garden": ["seed packet", "foraged key"],
"Bridge": ["cloth", "rope spool", "weather vane", "river stone"],
}
# ============================================================
# NARRATIVE ARC — 4 phases that transform the world
# ============================================================
@@ -143,6 +166,8 @@ class World:
"visitors": [],
},
}
for room_name, items in ROOM_DISCOVERABLES.items():
self.rooms[room_name]["discoverables"] = list(items)
# Characters (not NPCs — they have lives)
self.characters = {
@@ -258,6 +283,14 @@ class World:
"items_crafted": 0,
"conflicts_resolved": 0,
"nights_survived": 0,
"bridge_patched": False,
"bridge_secured": False,
"bridge_reinforced": False,
"bridge_lantern_lit": False,
"tower_cache_unlocked": False,
"threshold_marked": False,
"weather_readable": False,
"sunstone_socketed": False,
}
def tick_time(self):
@@ -376,6 +409,14 @@ class World:
desc += " Rain mists on the dark water below."
if len(self.rooms["Bridge"]["carvings"]) > 1:
desc += f" There are {len(self.rooms['Bridge']['carvings'])} carvings now."
if self.state.get("bridge_patched"):
desc += " Cloth bindings keep the railing from splintering."
if self.state.get("bridge_secured"):
desc += " Rope lines keep the span steady against the flood."
if self.state.get("bridge_reinforced"):
desc += " Fresh iron nails hold the planks tight."
if self.state.get("bridge_lantern_lit"):
desc += " A lantern glows warm over the water."
elif room_name == "Tower":
power = self.state.get("tower_power_low", False)
@@ -384,6 +425,19 @@ class World:
if self.rooms["Tower"]["messages"]:
desc += f" The whiteboard holds {len(self.rooms['Tower']['messages'])} rules."
if self.state.get("tower_cache_unlocked"):
desc += " A hidden cache stands open beneath the whiteboard."
if room_name == "Threshold" and self.state.get("threshold_marked"):
desc += " A chalk arrow points late arrivals toward shelter."
if room_name == "Garden" and self.state.get("weather_readable"):
desc += " The beds are arranged to catch whatever weather comes next."
if room_name == "Tower" and self.state.get("sunstone_socketed"):
desc += " A sunstone keeps the room lit with a stubborn amber glow."
discoverables = room.get("discoverables", [])
if discoverables:
desc += f" Discoverable items: {', '.join(discoverables)}."
# Who's here
here = [n for n, c in self.characters.items() if c["room"] == room_name and n != char_name]
@@ -485,6 +539,10 @@ class ActionSystem:
"cost": 1,
"description": "Take an item from the room",
},
"use": {
"cost": 1,
"description": "Use an item from your inventory to change the world",
},
"examine": {
"cost": 0,
"description": "Examine something in detail",
@@ -530,6 +588,13 @@ class ActionSystem:
available.append("rest")
available.append("examine")
discoverables = world.rooms[room].get("discoverables", [])
for item in discoverables:
available.append(f"take:{item}")
for item in char["inventory"]:
available.append(f"use:{item}")
if char["inventory"]:
available.append("give:item")
@@ -1072,6 +1137,76 @@ class GameEngine:
f.write(f"\n*Began: {datetime.now().strftime('%Y-%m-%d %H:%M')}*\n\n")
f.write("---\n\n")
f.write(message + "\n")
def _take_item(self, item_name, scene):
room_name = self.world.characters["Timmy"]["room"]
discoverables = self.world.rooms[room_name].get("discoverables", [])
if item_name not in discoverables:
scene["log"].append(f"There is no {item_name} here.")
return
discoverables.remove(item_name)
self.world.characters["Timmy"]["inventory"].append(item_name)
scene["log"].append(f"You take the {item_name}.")
if WORLD_ITEMS.get(item_name, {}).get("quest_item"):
scene["world_events"].append(f"The {item_name} feels important. It might open a quest route.")
def _use_item(self, item_name, scene):
inventory = self.world.characters["Timmy"]["inventory"]
if item_name not in inventory:
scene["log"].append(f"You are not carrying {item_name}.")
return
item = WORLD_ITEMS.get(item_name)
if not item:
scene["log"].append(f"The {item_name} doesn't seem to do anything.")
return
effect = item["effect"]
effect_text = item["effect_text"]
if effect == "grow_garden":
self.world.rooms["Garden"]["growth"] = min(5, self.world.rooms["Garden"]["growth"] + 2)
self.world.state["garden_drought"] = False
elif effect == "unlock_tower_cache":
self.world.state["tower_cache_unlocked"] = True
cache_rule = "Rule: Keys open more than doors when the world trusts you."
if cache_rule not in self.world.rooms["Tower"]["messages"]:
self.world.rooms["Tower"]["messages"].append(cache_rule)
elif effect == "write_notebook_rule":
note_rule = f"Rule #{self.world.tick}: A notebook can turn memory into structure."
self.world.rooms["Tower"]["messages"].append(note_rule)
elif effect == "patch_bridge":
self.world.state["bridge_patched"] = True
self.world.state["bridge_flooding"] = False
self.world.rooms["Bridge"]["weather"] = None
self.world.rooms["Bridge"]["rain_ticks"] = 0
elif effect == "stoke_forge":
self.world.rooms["Forge"]["fire"] = "glowing"
self.world.state["forge_fire_dying"] = False
elif effect == "light_bridge":
self.world.state["bridge_lantern_lit"] = True
elif effect == "secure_bridge":
self.world.state["bridge_secured"] = True
self.world.state["bridge_flooding"] = False
self.world.rooms["Bridge"]["weather"] = None
self.world.rooms["Bridge"]["rain_ticks"] = 0
elif effect == "mark_threshold":
self.world.state["threshold_marked"] = True
elif effect == "read_weather":
self.world.state["weather_readable"] = True
self.world.state["garden_drought"] = False
elif effect == "restore_tower_power":
self.world.state["tower_power_low"] = False
self.world.state["sunstone_socketed"] = True
elif effect == "reinforce_bridge":
self.world.state["bridge_reinforced"] = True
self.world.state["bridge_flooding"] = False
elif effect == "water_garden":
self.world.state["garden_drought"] = False
self.world.rooms["Garden"]["growth"] = min(5, self.world.rooms["Garden"]["growth"] + 1)
scene["log"].append(f"You use the {item_name}. {effect_text}")
scene["world_events"].append(effect_text)
if item.get("consumable"):
inventory.remove(item_name)
def run_tick(self, timmy_action="look"):
"""Run one tick. Return the scene and available choices."""
@@ -1100,7 +1235,7 @@ class GameEngine:
action_costs = {
"move": 2, "tend_fire": 3, "write_rule": 2, "carve": 2,
"plant": 2, "study": 2, "forge": 3, "help": 2, "speak": 1,
"listen": 0, "rest": -2, "examine": 0, "give": 0, "take": 1,
"listen": 0, "rest": -2, "examine": 0, "give": 0, "take": 1, "use": 1,
}
# Extract action name
@@ -1360,9 +1495,17 @@ class GameEngine:
elif timmy_action == "examine":
room = self.world.characters["Timmy"]["room"]
room_data = self.world.rooms[room]
items = room_data.get("items", [])
items = room_data.get("items", []) + room_data.get("discoverables", [])
scene["log"].append(f"You examine The {room}. You see: {', '.join(items) if items else 'nothing special'}")
elif timmy_action.startswith("take:"):
item_name = timmy_action.split(":", 1)[1]
self._take_item(item_name, scene)
elif timmy_action.startswith("use:"):
item_name = timmy_action.split(":", 1)[1]
self._use_item(item_name, scene)
elif timmy_action.startswith("help:"):
# Help increases trust
target_name = timmy_action.split(":")[1]

View File

@@ -1,406 +0,0 @@
#!/usr/bin/env python3
"""Burn-down velocity tracker for Timmy Foundation issue throughput.
Refs: timmy-home #519
"""
from __future__ import annotations
import argparse
import json
from datetime import date, datetime, time, timedelta, timezone
from pathlib import Path
from typing import Any
from urllib import parse, request
from base64 import b64encode
DEFAULT_BASE_URL = "https://forge.alexanderwhitestone.com/api/v1"
DEFAULT_OWNER = "Timmy_Foundation"
DEFAULT_TOKEN_FILE = Path.home() / ".config" / "gitea" / "token"
DEFAULT_CONFIG_FILE = Path(__file__).resolve().parent.parent / "configs" / "burn_velocity_repos.json"
DEFAULT_OUTPUT_DIR = Path.home() / ".timmy" / "burn-velocity"
DEFAULT_OUTPUT_JSON = DEFAULT_OUTPUT_DIR / "latest.json"
DEFAULT_OUTPUT_MD = DEFAULT_OUTPUT_DIR / "latest.md"
DEFAULT_HISTORY_FILE = DEFAULT_OUTPUT_DIR / "history.json"
DEFAULT_CONFIG = {
"owner": DEFAULT_OWNER,
"repos": ["timmy-home", "timmy-config", "fleet-ops", "the-beacon", "the-door", "the-nexus"],
"lookback_days": 14,
"alert": {
"recent_days": 7,
"baseline_days": 7,
"minimum_baseline_closed": 4,
"drop_ratio": 0.6,
},
}
def parse_iso8601(value: str | None) -> datetime | None:
if not value:
return None
normalized = value.replace("Z", "+00:00")
parsed = datetime.fromisoformat(normalized)
if parsed.tzinfo is None:
return parsed.replace(tzinfo=timezone.utc)
return parsed.astimezone(timezone.utc)
def normalize_today(value: str | date | None = None) -> date:
if value is None:
return datetime.now(timezone.utc).date()
if isinstance(value, date):
return value
return date.fromisoformat(value)
def build_day_window(today: date, lookback_days: int) -> list[date]:
start = today - timedelta(days=lookback_days - 1)
return [start + timedelta(days=offset) for offset in range(lookback_days)]
def filter_issue_items(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
return [item for item in items if not item.get("pull_request")]
def build_daily_series(items: list[dict[str, Any]], today: date, lookback_days: int) -> list[dict[str, int | str]]:
days = build_day_window(today, lookback_days)
counts = {day.isoformat(): {"opened": 0, "closed": 0} for day in days}
start_day = days[0]
for item in filter_issue_items(items):
created_at = parse_iso8601(item.get("created_at"))
if created_at is not None:
created_day = created_at.date()
if start_day <= created_day <= today:
counts[created_day.isoformat()]["opened"] += 1
closed_at = parse_iso8601(item.get("closed_at"))
if closed_at is not None:
closed_day = closed_at.date()
if start_day <= closed_day <= today:
counts[closed_day.isoformat()]["closed"] += 1
return [
{
"date": day.isoformat(),
"opened": counts[day.isoformat()]["opened"],
"closed": counts[day.isoformat()]["closed"],
}
for day in days
]
def summarize_velocity_alert(
*, recent_closed: int, baseline_closed: int, open_now: int, config: dict[str, Any]
) -> dict[str, Any]:
minimum_baseline = int(config.get("minimum_baseline_closed", 4))
drop_ratio = float(config.get("drop_ratio", 0.6))
if baseline_closed >= minimum_baseline and recent_closed < baseline_closed * drop_ratio:
return {
"status": "drop",
"kind": "velocity_drop",
"recent_closed": recent_closed,
"baseline_closed": baseline_closed,
"reason": (
f"velocity_drop: closed {recent_closed} in the last {config.get('recent_days', 7)}d "
f"vs {baseline_closed} in the prior {config.get('baseline_days', 7)}d"
),
}
if open_now > 0 and baseline_closed >= minimum_baseline and recent_closed == 0:
return {
"status": "drop",
"kind": "velocity_drop",
"recent_closed": recent_closed,
"baseline_closed": baseline_closed,
"reason": "velocity_drop: no issues closed in the recent window while backlog is still open",
}
return {
"status": "ok",
"kind": "none",
"recent_closed": recent_closed,
"baseline_closed": baseline_closed,
"reason": "velocity stable",
}
def _sum_window(daily: list[dict[str, int | str]], field: str, days: int) -> int:
if days <= 0:
return 0
return sum(int(item[field]) for item in daily[-days:])
def _sum_baseline_window(daily: list[dict[str, int | str]], recent_days: int, baseline_days: int) -> int:
if baseline_days <= 0:
return 0
if recent_days <= 0:
return sum(int(item["closed"]) for item in daily[-baseline_days:])
baseline_slice = daily[-(recent_days + baseline_days) : -recent_days]
return sum(int(item["closed"]) for item in baseline_slice)
def build_velocity_report(config: dict[str, Any], snapshot: dict[str, Any], today: str | date | None = None) -> dict[str, Any]:
report_day = normalize_today(today)
generated_at = snapshot.get("generated_at") or datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
owner = config.get("owner", DEFAULT_OWNER)
repos = list(config.get("repos") or sorted((snapshot.get("repos") or {}).keys()))
lookback_days = int(config.get("lookback_days", 14))
alert_config = dict(DEFAULT_CONFIG["alert"])
alert_config.update(config.get("alert") or {})
recent_days = int(alert_config.get("recent_days", 7))
baseline_days = int(alert_config.get("baseline_days", 7))
repo_reports: list[dict[str, Any]] = []
total_open_now = 0
total_closed_last_7d = 0
repos_with_alerts: list[str] = []
for repo_name in repos:
repo_snapshot = (snapshot.get("repos") or {}).get(repo_name, {})
open_issues = filter_issue_items(list(repo_snapshot.get("open_issues") or []))
recent_issues = filter_issue_items(list(repo_snapshot.get("recent_issues") or []))
daily = build_daily_series(recent_issues, report_day, lookback_days)
open_now = len(open_issues)
opened_last_7d = _sum_window(daily, "opened", recent_days)
closed_last_7d = _sum_window(daily, "closed", recent_days)
baseline_closed = _sum_baseline_window(daily, recent_days, baseline_days)
weekly_net = opened_last_7d - closed_last_7d
alert = summarize_velocity_alert(
recent_closed=closed_last_7d,
baseline_closed=baseline_closed,
open_now=open_now,
config=alert_config,
)
repo_report = {
"repo": repo_name,
"open_now": open_now,
"opened_last_7d": opened_last_7d,
"closed_last_7d": closed_last_7d,
"baseline_closed": baseline_closed,
"weekly_net": weekly_net,
"daily": daily,
"alert": alert,
}
repo_reports.append(repo_report)
total_open_now += open_now
total_closed_last_7d += closed_last_7d
if alert["status"] != "ok":
repos_with_alerts.append(repo_name)
return {
"owner": owner,
"generated_at": generated_at,
"generated_day": report_day.isoformat(),
"lookback_days": lookback_days,
"dashboard_contract_version": 1,
"repos": repo_reports,
"summary": {
"total_open_now": total_open_now,
"total_closed_last_7d": total_closed_last_7d,
"repos_with_alerts": repos_with_alerts,
},
}
def render_markdown(report: dict[str, Any]) -> str:
lines = [
"# Burn-down Velocity Tracking",
"",
f"Generated: {report['generated_at']}",
f"Owner: {report['owner']}",
f"Lookback days: {report['lookback_days']}",
"",
"## Per-repo velocity",
"",
"| Repo | Open now | Opened 7d | Closed 7d | Previous 7d | Alert |",
"| --- | ---: | ---: | ---: | ---: | --- |",
]
for repo in report["repos"]:
alert_label = repo["alert"]["kind"] if repo["alert"]["status"] != "ok" else "ok"
lines.append(
f"| {repo['repo']} | {repo['open_now']} | {repo['opened_last_7d']} | {repo['closed_last_7d']} | {repo['baseline_closed']} | {alert_label} |"
)
lines.extend(
[
"",
"## Dashboard handoff for timmy-config",
"",
"The timmy-config dashboard should consume `~/.timmy/burn-velocity/latest.json` and render, for each repo:",
"- `open_now`",
"- `opened_last_7d`",
"- `closed_last_7d`",
"- `baseline_closed`",
"- `alert.status` / `alert.kind` / `alert.reason`",
"",
"Cron should also persist `~/.timmy/burn-velocity/history.json` so timmy-config can plot the daily trend line instead of only the latest snapshot.",
"",
"## Alerts",
"",
]
)
alerts = [repo for repo in report["repos"] if repo["alert"]["status"] != "ok"]
if not alerts:
lines.append("- none")
else:
for repo in alerts:
lines.append(f"- {repo['repo']}: {repo['alert']['reason']}")
return "\n".join(lines) + "\n"
def update_history(history_path: Path, report: dict[str, Any]) -> dict[str, Any]:
if history_path.exists():
history = json.loads(history_path.read_text(encoding="utf-8"))
else:
history = {"days": []}
entry = {
"date": report["generated_day"],
"generated_at": report["generated_at"],
"summary": report["summary"],
"repos": report["repos"],
}
retained = [item for item in history.get("days", []) if item.get("date") != report["generated_day"]]
retained.append(entry)
retained.sort(key=lambda item: item["date"])
history["days"] = retained
history_path.parent.mkdir(parents=True, exist_ok=True)
history_path.write_text(json.dumps(history, indent=2), encoding="utf-8")
return history
class GiteaClient:
def __init__(self, token: str, owner: str = DEFAULT_OWNER, base_url: str = DEFAULT_BASE_URL):
self.token = token
self.owner = owner
self.base_url = base_url.rstrip("/")
def _headers(self) -> list[dict[str, str]]:
return [
{"Authorization": f"token {self.token}", "Accept": "application/json"},
{
"Authorization": "Basic " + b64encode(f"{self.token}:".encode()).decode(),
"Accept": "application/json",
},
]
def _request_json(self, url: str) -> list[dict[str, Any]]:
last_error: Exception | None = None
for headers in self._headers():
try:
req = request.Request(url, headers=headers)
with request.urlopen(req, timeout=30) as response:
return json.loads(response.read().decode())
except Exception as exc: # pragma: no cover - exercised only on live API failure
last_error = exc
if last_error is None: # pragma: no cover - defensive
raise RuntimeError("request failed without an exception")
raise last_error
def list_issues(self, repo: str, *, state: str, since: str | None = None) -> list[dict[str, Any]]:
issues: list[dict[str, Any]] = []
page = 1
while True:
query = {"state": state, "type": "issues", "limit": 100, "page": page}
if since:
query["since"] = since
url = f"{self.base_url}/repos/{self.owner}/{repo}/issues?{parse.urlencode(query)}"
batch = self._request_json(url)
if not batch:
break
issues.extend(filter_issue_items(batch))
page += 1
return issues
def load_json(path: Path, default: Any) -> Any:
if not path.exists():
return default
return json.loads(path.read_text(encoding="utf-8"))
def load_config(path: Path) -> dict[str, Any]:
config = dict(DEFAULT_CONFIG)
alert = dict(DEFAULT_CONFIG["alert"])
raw = load_json(path, {})
config.update(raw)
alert.update(raw.get("alert") or {})
config["alert"] = alert
return config
def collect_live_snapshot(
config: dict[str, Any], *, today: str | date | None = None, token_file: Path = DEFAULT_TOKEN_FILE, base_url: str = DEFAULT_BASE_URL
) -> dict[str, Any]:
token = token_file.read_text(encoding="utf-8").strip()
report_day = normalize_today(today)
since_day = report_day - timedelta(days=int(config.get("lookback_days", 14)) - 1)
since_timestamp = datetime.combine(since_day, time.min, tzinfo=timezone.utc).isoformat().replace("+00:00", "Z")
client = GiteaClient(token=token, owner=config.get("owner", DEFAULT_OWNER), base_url=base_url)
repos = list(config.get("repos") or [])
repo_payload = {}
for repo in repos:
repo_payload[repo] = {
"open_issues": client.list_issues(repo, state="open"),
"recent_issues": client.list_issues(repo, state="all", since=since_timestamp),
}
return {
"generated_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
"repos": repo_payload,
}
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Track per-repo issue burn-down velocity and emit timmy-config dashboard payloads.")
parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG_FILE, help="Repo tracking config JSON")
parser.add_argument("--snapshot-file", type=Path, help="Use a pre-fetched snapshot JSON instead of calling Gitea")
parser.add_argument("--token-file", type=Path, default=DEFAULT_TOKEN_FILE, help="Gitea token file for live collection")
parser.add_argument("--base-url", default=DEFAULT_BASE_URL, help="Gitea API base URL")
parser.add_argument("--today", help="Override report date (YYYY-MM-DD)")
parser.add_argument("--output-json", type=Path, default=DEFAULT_OUTPUT_JSON, help="Path for latest JSON payload")
parser.add_argument("--output-md", type=Path, default=DEFAULT_OUTPUT_MD, help="Path for latest markdown summary")
parser.add_argument("--history-file", type=Path, default=DEFAULT_HISTORY_FILE, help="Path for persisted daily history JSON")
parser.add_argument("--write-history", action="store_true", help="Update the daily history file after generating the report")
parser.add_argument("--json", action="store_true", help="Print JSON instead of markdown to stdout")
return parser.parse_args()
def main() -> None:
args = parse_args()
config = load_config(args.config)
if args.snapshot_file:
snapshot = load_json(args.snapshot_file, {"repos": {}})
else:
snapshot = collect_live_snapshot(config, today=args.today, token_file=args.token_file, base_url=args.base_url)
report = build_velocity_report(config, snapshot, today=args.today)
args.output_json.parent.mkdir(parents=True, exist_ok=True)
args.output_md.parent.mkdir(parents=True, exist_ok=True)
args.output_json.write_text(json.dumps(report, indent=2), encoding="utf-8")
args.output_md.write_text(render_markdown(report), encoding="utf-8")
if args.write_history:
update_history(args.history_file, report)
if args.json:
print(json.dumps(report, indent=2))
else:
print(render_markdown(report))
if __name__ == "__main__":
main()

View File

@@ -1,176 +0,0 @@
from __future__ import annotations
import json
import subprocess
import sys
from datetime import date
from pathlib import Path
from scripts.burn_velocity_tracker import build_velocity_report, render_markdown, update_history
ROOT = Path(__file__).resolve().parent.parent
DOC_PATH = ROOT / "docs" / "BURN_VELOCITY_TRACKING.md"
SNAPSHOT = {
"generated_at": "2026-04-22T12:00:00Z",
"repos": {
"timmy-home": {
"open_issues": [
{"number": 501, "state": "open", "created_at": "2026-04-20T09:00:00Z"},
{"number": 502, "state": "open", "created_at": "2026-04-22T07:00:00Z"},
],
"recent_issues": [
{"number": 401, "state": "closed", "created_at": "2026-04-21T09:00:00Z", "closed_at": "2026-04-22T05:30:00Z"},
{"number": 402, "state": "closed", "created_at": "2026-04-20T09:00:00Z", "closed_at": "2026-04-21T05:30:00Z"},
{"number": 403, "state": "closed", "created_at": "2026-04-19T09:00:00Z", "closed_at": "2026-04-20T05:30:00Z"},
{"number": 404, "state": "closed", "created_at": "2026-04-14T09:00:00Z", "closed_at": "2026-04-15T05:30:00Z"},
{"number": 405, "state": "closed", "created_at": "2026-04-13T09:00:00Z", "closed_at": "2026-04-14T05:30:00Z"},
{"number": 406, "state": "closed", "created_at": "2026-04-12T09:00:00Z", "closed_at": "2026-04-13T05:30:00Z"},
{"number": 407, "state": "closed", "created_at": "2026-04-11T09:00:00Z", "closed_at": "2026-04-12T05:30:00Z"},
{"number": 408, "state": "closed", "created_at": "2026-04-10T09:00:00Z", "closed_at": "2026-04-11T05:30:00Z"},
{"number": 409, "state": "closed", "created_at": "2026-04-09T09:00:00Z", "closed_at": "2026-04-10T05:30:00Z"},
{"number": 410, "state": "closed", "created_at": "2026-04-08T09:00:00Z", "closed_at": "2026-04-09T05:30:00Z"},
{"number": 411, "state": "closed", "created_at": "2026-04-07T09:00:00Z", "closed_at": "2026-04-08T05:30:00Z"},
{"number": 412, "state": "closed", "created_at": "2026-04-06T09:00:00Z", "closed_at": "2026-04-07T05:30:00Z"},
{"number": 413, "state": "closed", "created_at": "2026-04-05T09:00:00Z", "closed_at": "2026-04-06T05:30:00Z"},
{"number": 414, "state": "open", "created_at": "2026-04-22T08:45:00Z", "closed_at": None},
{"number": 415, "state": "open", "created_at": "2026-04-17T08:45:00Z", "closed_at": None},
],
},
"timmy-config": {
"open_issues": [
{"number": 601, "state": "open", "created_at": "2026-04-18T09:00:00Z"},
],
"recent_issues": [
{"number": 602, "state": "closed", "created_at": "2026-04-20T09:00:00Z", "closed_at": "2026-04-21T06:00:00Z"},
{"number": 603, "state": "open", "created_at": "2026-04-22T06:00:00Z", "closed_at": None},
],
},
},
}
CONFIG = {
"owner": "Timmy_Foundation",
"repos": ["timmy-home", "timmy-config"],
"lookback_days": 14,
"alert": {
"recent_days": 7,
"baseline_days": 7,
"minimum_baseline_closed": 4,
"drop_ratio": 0.6,
},
}
def test_build_velocity_report_counts_opened_closed_and_flags_drop_alert() -> None:
report = build_velocity_report(CONFIG, SNAPSHOT, today=date(2026, 4, 22))
assert report["generated_day"] == "2026-04-22"
assert report["summary"]["repos_with_alerts"] == ["timmy-home"]
assert report["summary"]["total_open_now"] == 3
home = report["repos"][0]
assert home["repo"] == "timmy-home"
assert home["open_now"] == 2
assert home["opened_last_7d"] == 5
assert home["closed_last_7d"] == 3
assert home["baseline_closed"] == 7
assert home["weekly_net"] == 2
assert home["alert"]["status"] == "drop"
assert home["alert"]["recent_closed"] == 3
assert home["daily"][-1] == {"date": "2026-04-22", "opened": 1, "closed": 1}
timmy_config = report["repos"][1]
assert timmy_config["repo"] == "timmy-config"
assert timmy_config["open_now"] == 1
assert timmy_config["closed_last_7d"] == 1
assert timmy_config["alert"]["status"] == "ok"
def test_render_markdown_includes_dashboard_handoff_and_alerts() -> None:
report = build_velocity_report(CONFIG, SNAPSHOT, today=date(2026, 4, 22))
rendered = render_markdown(report)
for snippet in (
"# Burn-down Velocity Tracking",
"## Per-repo velocity",
"timmy-home",
"timmy-config",
"## Dashboard handoff for timmy-config",
"velocity_drop",
"## Alerts",
):
assert snippet in rendered
def test_update_history_replaces_same_day_snapshot(tmp_path: Path) -> None:
history_path = tmp_path / "burn-velocity-history.json"
report = build_velocity_report(CONFIG, SNAPSHOT, today=date(2026, 4, 22))
update_history(history_path, report)
updated = json.loads(json.dumps(report))
updated["repos"][0]["open_now"] = 9
updated["summary"]["total_open_now"] = 10
update_history(history_path, updated)
history = json.loads(history_path.read_text(encoding="utf-8"))
assert [item["date"] for item in history["days"]] == ["2026-04-22"]
assert history["days"][0]["summary"]["total_open_now"] == 10
assert history["days"][0]["repos"][0]["open_now"] == 9
def test_cli_writes_json_markdown_and_history_from_snapshot(tmp_path: Path) -> None:
snapshot_path = tmp_path / "snapshot.json"
output_json = tmp_path / "latest.json"
output_md = tmp_path / "latest.md"
history_path = tmp_path / "history.json"
snapshot_path.write_text(json.dumps(SNAPSHOT), encoding="utf-8")
result = subprocess.run(
[
sys.executable,
"-m",
"scripts.burn_velocity_tracker",
"--snapshot-file",
str(snapshot_path),
"--today",
"2026-04-22",
"--output-json",
str(output_json),
"--output-md",
str(output_md),
"--history-file",
str(history_path),
"--write-history",
"--json",
],
check=True,
cwd=ROOT,
capture_output=True,
text=True,
)
payload = json.loads(result.stdout)
assert payload["summary"]["repos_with_alerts"] == ["timmy-home"]
assert output_json.exists()
assert output_md.exists()
assert history_path.exists()
assert "timmy-config" in output_md.read_text(encoding="utf-8")
def test_repo_contains_burn_velocity_tracking_doc() -> None:
text = DOC_PATH.read_text(encoding="utf-8")
required = [
"# Burn-down Velocity Tracking",
"python3 scripts/burn_velocity_tracker.py",
"configs/burn_velocity_repos.json",
"~/.timmy/burn-velocity/latest.json",
"timmy-config dashboard",
"type=issues",
"velocity_drop",
]
for snippet in required:
assert snippet in text

View File

@@ -0,0 +1,58 @@
from importlib.util import module_from_spec, spec_from_file_location
from pathlib import Path
import unittest
ROOT = Path(__file__).resolve().parent.parent
GAME_PATH = ROOT / "evennia" / "timmy_world" / "game.py"
def load_game_module():
spec = spec_from_file_location("tower_game_items", GAME_PATH)
module = module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(module)
module.random.seed(0)
return module
class TestTowerGameWorldItems(unittest.TestCase):
def test_world_has_ten_unique_items_and_a_quest_item(self):
module = load_game_module()
world = module.World()
room_items = {
item
for room in world.rooms.values()
for item in room.get("discoverables", [])
}
self.assertGreaterEqual(len(room_items), 10)
self.assertIn("foraged key", room_items)
self.assertTrue(module.WORLD_ITEMS["foraged key"]["quest_item"])
def test_items_change_world_state_when_used(self):
module = load_game_module()
engine = module.GameEngine()
engine.start_new_game()
engine.world.characters["Timmy"]["energy"] = 10
engine.world.characters["Timmy"]["room"] = "Garden"
initial_growth = engine.world.rooms["Garden"]["growth"]
engine.run_tick("take:seed packet")
use_seed = engine.run_tick("use:seed packet")
self.assertGreater(engine.world.rooms["Garden"]["growth"], initial_growth)
self.assertNotIn("seed packet", engine.world.characters["Timmy"]["inventory"])
self.assertTrue(any("garden" in line.lower() for line in use_seed["world_events"] + use_seed["log"]))
engine.world.characters["Timmy"]["energy"] = 10
engine.run_tick("take:foraged key")
use_key = engine.run_tick("use:foraged key")
self.assertTrue(engine.world.state["tower_cache_unlocked"])
self.assertTrue(any("cache" in line.lower() or "quest" in line.lower() for line in use_key["world_events"] + use_key["log"]))
if __name__ == "__main__":
unittest.main()