Compare commits

..

6 Commits

Author SHA1 Message Date
Alexander Whitestone
79b841727f docs: add timmy-dispatch genome artifact (#682) 2026-04-15 02:33:19 -04:00
3016e012cc Merge PR #739: feat: add laptop fleet planner scaffold (#530) 2026-04-15 06:17:19 +00:00
60b9b90f34 Merge PR #738: feat: add Know Thy Father epic orchestrator 2026-04-15 06:12:05 +00:00
Alexander Whitestone
c818a30522 feat: add laptop fleet planner scaffold (#530)
Some checks failed
Smoke Test / smoke (pull_request) Failing after 30s
2026-04-15 02:11:31 -04:00
Alexander Whitestone
89dfa1e5de feat: add Know Thy Father epic orchestrator (#582)
Some checks failed
Smoke Test / smoke (pull_request) Failing after 23s
2026-04-15 01:52:58 -04:00
Alexander Whitestone
d791c087cb feat: add Ezra mempalace integration packet (#570)
Some checks failed
Smoke Test / smoke (pull_request) Failing after 22s
2026-04-15 01:37:47 -04:00
16 changed files with 1241 additions and 539 deletions

View File

@@ -1,81 +0,0 @@
# Bezalel Evennia World
Issue: `timmy-home#536`
This is the themed-room world plan and build scaffold for Bezalel, the forge-and-testbed wizard.
## Rooms
| Room | Description focus | Core connections |
|------|-------------------|------------------|
| Limbo | the threshold between houses | Gatehouse |
| Gatehouse | guarded entry, travel runes, proof before trust | Limbo, Great Hall, The Portal Room |
| Great Hall | three-house maps, reports, shared table | Gatehouse, The Library of Bezalel, The Observatory, The Workshop |
| The Library of Bezalel | manuals, bridge schematics, technical memory | Great Hall |
| The Observatory | long-range signals toward Mac, VPS, and the wider net | Great Hall |
| The Workshop | forge + workbench, plans turned into working form | Great Hall, The Server Room, The Garden of Code |
| The Server Room | humming racks, heartbeat of the house | The Workshop |
| The Garden of Code | contemplative grove where ideas root before implementation | The Workshop |
| The Portal Room | three shimmering doorways aimed at Mac, VPS, and the net | Gatehouse |
## Characters
| Character | Role | Starting room |
|-----------|------|---------------|
| Timmy | quiet builder and observer | Gatehouse |
| Bezalel | forge-and-testbed wizard | The Workshop |
| Marcus | old man with kind eyes, human warmth in the system | The Garden of Code |
| Kimi | scholar of context and meaning | The Library of Bezalel |
## Themed items
At least one durable item is placed in every major room, including:
- Threshold Ledger
- Three-House Map
- Bridge Schematics
- Compiler Manuals
- Tri-Axis Telescope
- Forge Anvil
- Bridge Workbench
- Heartbeat Console
- Server Racks
- Code Orchard
- Stone Bench
- Mac/VPS/Net portal markers
## Portal travel commands
The Portal Room reserves three live command names:
- `mac`
- `vps`
- `net`
Current behavior in the build scaffold:
- each command is created as a real Evennia exit command
- each command preserves explicit target metadata (`Mac house`, `VPS house`, `Wider net`)
- until cross-world transport is wired, each portal routes through `Limbo`, the inter-world threshold room
This keeps the command surface real now while leaving honest room for later world-to-world linking.
## Build script
```bash
python3 scripts/evennia/build_bezalel_world.py --plan
```
Inside an Evennia shell / runtime with the repo on `PYTHONPATH`, the same script can build the world idempotently:
```bash
python3 scripts/evennia/build_bezalel_world.py --password bezalel-world-dev
```
What it does:
- creates or updates all 9 rooms
- creates the exit graph
- creates themed objects
- creates or rehomes account-backed characters
- creates the portal command exits with target metadata
## Persistence note
The scaffold is written to be idempotent: rerunning the builder updates descriptions, destinations, and locations rather than creating duplicate world entities. That is the repo-side prerequisite for persistence across Evennia restarts.

View File

@@ -0,0 +1,61 @@
# Know Thy Father — Multimodal Media Consumption Pipeline
Refs #582
This document makes the epic operational by naming the current source-of-truth scripts, their handoff artifacts, and the one-command runner that coordinates them.
## Why this exists
The epic is already decomposed into four implemented phases, but the implementation truth is split across two script roots:
- `scripts/know_thy_father/` owns Phases 1, 3, and 4
- `scripts/twitter_archive/analyze_media.py` owns Phase 2
- `twitter-archive/know-thy-father/tracker.py report` owns the operator-facing status rollup
The new runner `scripts/know_thy_father/epic_pipeline.py` does not replace those scripts. It stitches them together into one explicit, reviewable plan.
## Phase map
| Phase | Script | Primary output |
|-------|--------|----------------|
| 1. Media Indexing | `scripts/know_thy_father/index_media.py` | `twitter-archive/know-thy-father/media_manifest.jsonl` |
| 2. Multimodal Analysis | `scripts/twitter_archive/analyze_media.py --batch 10` | `twitter-archive/know-thy-father/analysis.jsonl` + `meaning-kernels.jsonl` + `pipeline-status.json` |
| 3. Holographic Synthesis | `scripts/know_thy_father/synthesize_kernels.py` | `twitter-archive/knowledge/fathers_ledger.jsonl` |
| 4. Cross-Reference Audit | `scripts/know_thy_father/crossref_audit.py` | `twitter-archive/notes/crossref_report.md` |
| 5. Processing Log | `twitter-archive/know-thy-father/tracker.py report` | `twitter-archive/know-thy-father/REPORT.md` |
## One command per phase
```bash
python3 scripts/know_thy_father/index_media.py --tweets twitter-archive/extracted/tweets.jsonl --output twitter-archive/know-thy-father/media_manifest.jsonl
python3 scripts/twitter_archive/analyze_media.py --batch 10
python3 scripts/know_thy_father/synthesize_kernels.py --input twitter-archive/media/manifest.jsonl --output twitter-archive/knowledge/fathers_ledger.jsonl --summary twitter-archive/knowledge/fathers_ledger.summary.json
python3 scripts/know_thy_father/crossref_audit.py --soul SOUL.md --kernels twitter-archive/notes/know_thy_father_crossref.md --output twitter-archive/notes/crossref_report.md
python3 twitter-archive/know-thy-father/tracker.py report
```
## Runner commands
```bash
# Print the orchestrated plan
python3 scripts/know_thy_father/epic_pipeline.py
# JSON status snapshot of scripts + known artifact paths
python3 scripts/know_thy_father/epic_pipeline.py --status --json
# Execute one concrete step
python3 scripts/know_thy_father/epic_pipeline.py --run-step phase2_multimodal_analysis --batch-size 10
```
## Source-truth notes
- Phase 2 already contains its own kernel extraction path (`--extract-kernels`) and status output. The epic runner does not reimplement that logic.
- Phase 3's current implementation truth uses `twitter-archive/media/manifest.jsonl` as its default input. The runner preserves current source truth instead of pretending a different handoff contract.
- The processing log in `twitter-archive/know-thy-father/PROCESSING_LOG.md` can drift from current code reality. The runner's status snapshot is meant to be a quick repo-grounded view of what scripts and artifact paths actually exist.
## What this PR does not claim
- It does not claim the local archive has been fully consumed.
- It does not claim the halted processing log has been resumed.
- It does not claim fact_store ingestion has been fully wired end-to-end.
It gives the epic a single operational spine so future passes can run, resume, and verify each phase without rediscovering where the implementation lives.

View File

@@ -0,0 +1,92 @@
# MemPalace v3.0.0 — Ezra Integration Packet
This packet turns issue #570 into an executable, reviewable integration plan for Ezra's Hermes home.
It is a repo-side scaffold: no live Ezra host changes are claimed in this artifact.
## Commands
```bash
pip install mempalace==3.0.0
mempalace init ~/.hermes/ --yes
cat > ~/.hermes/mempalace.yaml <<'YAML'
wing: ezra_home
palace: ~/.mempalace/palace
rooms:
- name: sessions
description: Conversation history and durable agent transcripts
globs:
- "*.json"
- "*.jsonl"
- name: config
description: Hermes configuration and runtime settings
globs:
- "*.yaml"
- "*.yml"
- "*.toml"
- name: docs
description: Notes, markdown docs, and operating reports
globs:
- "*.md"
- "*.txt"
people: []
projects: []
YAML
echo "" | mempalace mine ~/.hermes/
echo "" | mempalace mine ~/.hermes/sessions/ --mode convos
mempalace search "your common queries"
mempalace wake-up
hermes mcp add mempalace -- python -m mempalace.mcp_server
```
## Manual config template
```yaml
wing: ezra_home
palace: ~/.mempalace/palace
rooms:
- name: sessions
description: Conversation history and durable agent transcripts
globs:
- "*.json"
- "*.jsonl"
- name: config
description: Hermes configuration and runtime settings
globs:
- "*.yaml"
- "*.yml"
- "*.toml"
- name: docs
description: Notes, markdown docs, and operating reports
globs:
- "*.md"
- "*.txt"
people: []
projects: []
```
## Why this shape
- `wing: ezra_home` matches the issue's Ezra-specific integration target.
- `rooms` split the mined material into sessions, config, and docs to keep retrieval interpretable.
- Mining commands pipe empty stdin to avoid the interactive entity-detector hang noted in the evaluation.
## Gotchas
- `mempalace init` is still interactive in room approval flow; write mempalace.yaml manually if the init output stalls.
- The yaml key is `wing:` not `wings:`. Using the wrong key causes mine/setup failures.
- Pipe empty stdin into mining commands (`echo "" | ...`) to avoid the entity-detector stdin hang on larger directories.
- First mine downloads the ChromaDB embedding model cache (~79MB).
- Report Ezra's before/after metrics back to issue #568 after live installation and retrieval tests.
## Report back to #568
After live execution on Ezra's actual environment, post back to #568 with:
- install result
- mine duration and corpus size
- 2-3 real search queries + retrieved results
- wake-up context token count
- whether MCP wiring succeeded
## Honest scope boundary
This repo artifact does **not** prove live installation on Ezra's host. It makes the work reproducible and testable so the next pass can execute it without guesswork.

View File

@@ -0,0 +1,62 @@
fleet_name: timmy-laptop-fleet
machines:
- hostname: timmy-anchor-a
machine_type: laptop
ram_gb: 16
cpu_cores: 8
os: macOS
adapter_condition: good
idle_watts: 11
always_on_capable: true
notes: candidate 24/7 anchor agent
- hostname: timmy-anchor-b
machine_type: laptop
ram_gb: 8
cpu_cores: 4
os: Linux
adapter_condition: good
idle_watts: 13
always_on_capable: true
notes: candidate 24/7 anchor agent
- hostname: timmy-daylight-a
machine_type: laptop
ram_gb: 32
cpu_cores: 10
os: macOS
adapter_condition: ok
idle_watts: 22
always_on_capable: true
notes: higher-performance daylight compute
- hostname: timmy-daylight-b
machine_type: laptop
ram_gb: 16
cpu_cores: 8
os: Linux
adapter_condition: ok
idle_watts: 19
always_on_capable: true
notes: daylight compute node
- hostname: timmy-daylight-c
machine_type: laptop
ram_gb: 8
cpu_cores: 4
os: Windows
adapter_condition: needs_replacement
idle_watts: 17
always_on_capable: false
notes: repair power adapter before production duty
- hostname: timmy-desktop-nas
machine_type: desktop
ram_gb: 64
cpu_cores: 12
os: Linux
adapter_condition: good
idle_watts: 58
always_on_capable: false
has_4tb_ssd: true
notes: desktop plus 4TB SSD NAS and heavy compute during peak sun

View File

@@ -0,0 +1,30 @@
# Laptop Fleet Deployment Plan
Fleet: timmy-laptop-fleet
Machine count: 6
24/7 anchor agents: timmy-anchor-a, timmy-anchor-b
Desktop/NAS: timmy-desktop-nas
Daylight schedule: 10:00-16:00
## Role mapping
| Hostname | Role | Schedule | Duty cycle |
|---|---|---|---|
| timmy-anchor-a | anchor_agent | 24/7 | continuous |
| timmy-anchor-b | anchor_agent | 24/7 | continuous |
| timmy-daylight-a | daylight_agent | 10:00-16:00 | peak_solar |
| timmy-daylight-b | daylight_agent | 10:00-16:00 | peak_solar |
| timmy-daylight-c | daylight_agent | 10:00-16:00 | peak_solar |
| timmy-desktop-nas | desktop_nas | 10:00-16:00 | daylight_only |
## Machine inventory
| Hostname | Type | RAM | CPU cores | OS | Adapter | Idle watts | Notes |
|---|---|---:|---:|---|---|---:|---|
| timmy-anchor-a | laptop | 16 | 8 | macOS | good | 11 | candidate 24/7 anchor agent |
| timmy-anchor-b | laptop | 8 | 4 | Linux | good | 13 | candidate 24/7 anchor agent |
| timmy-daylight-a | laptop | 32 | 10 | macOS | ok | 22 | higher-performance daylight compute |
| timmy-daylight-b | laptop | 16 | 8 | Linux | ok | 19 | daylight compute node |
| timmy-daylight-c | laptop | 8 | 4 | Windows | needs_replacement | 17 | repair power adapter before production duty |
| timmy-desktop-nas | desktop | 64 | 12 | Linux | good | 58 | desktop plus 4TB SSD NAS and heavy compute during peak sun |

View File

@@ -1,190 +0,0 @@
from collections import deque
from dataclasses import dataclass
@dataclass(frozen=True)
class RoomSpec:
key: str
desc: str
@dataclass(frozen=True)
class ExitSpec:
source: str
key: str
destination: str
aliases: tuple[str, ...] = ()
@dataclass(frozen=True)
class ObjectSpec:
key: str
location: str
desc: str
aliases: tuple[str, ...] = ()
@dataclass(frozen=True)
class CharacterSpec:
key: str
desc: str
starting_room: str
@dataclass(frozen=True)
class TravelCommandSpec:
key: str
aliases: tuple[str, ...]
target_world: str
fallback_room: str
desc: str
ROOMS = (
RoomSpec(
"Limbo",
"The void between worlds. The air carries the pulse of three houses: Mac, VPS, and this one. "
"Everything begins here before it is given form.",
),
RoomSpec(
"Gatehouse",
"A stone guard tower at the edge of Bezalel's world. The walls are carved with runes of travel, "
"proof, and return. Every arrival is weighed before it is trusted.",
),
RoomSpec(
"Great Hall",
"A vast hall with a long working table. Maps of the three houses hang beside sketches, benchmarks, "
"and deployment notes. This is where the forge reports back to the house.",
),
RoomSpec(
"The Library of Bezalel",
"Shelves of technical manuals, Evennia code, test logs, and bridge schematics rise to the ceiling. "
"This room holds plans waiting to be made real.",
),
RoomSpec(
"The Observatory",
"A high chamber with telescopes pointing toward the Mac, the VPS, and the wider net. Screens glow with "
"status lights, latency traces, and long-range signals.",
),
RoomSpec(
"The Workshop",
"A forge and workbench share the same heat. Scattered here are half-finished bridges, patched harnesses, "
"and tools laid out for proof before pride.",
),
RoomSpec(
"The Server Room",
"Racks of humming servers line the walls. Fans push warm air through the chamber while status LEDs beat "
"like a mechanical heart. This is the pulse of Bezalel's house.",
),
RoomSpec(
"The Garden of Code",
"A quiet garden where ideas are left long enough to grow roots. Code-shaped leaves flutter in patterned wind, "
"and a stone path invites patient thought.",
),
RoomSpec(
"The Portal Room",
"Three shimmering doorways stand in a ring: one marked for the Mac house, one for the VPS, and one for the wider net. "
"The room hums like a bridge waiting for traffic.",
),
)
EXITS = (
ExitSpec("Limbo", "gatehouse", "Gatehouse", ("gate", "tower")),
ExitSpec("Gatehouse", "limbo", "Limbo", ("void", "back")),
ExitSpec("Gatehouse", "greathall", "Great Hall", ("hall", "great hall")),
ExitSpec("Great Hall", "gatehouse", "Gatehouse", ("gate", "tower")),
ExitSpec("Great Hall", "library", "The Library of Bezalel", ("books", "study")),
ExitSpec("The Library of Bezalel", "hall", "Great Hall", ("great hall", "back")),
ExitSpec("Great Hall", "observatory", "The Observatory", ("telescope", "tower top")),
ExitSpec("The Observatory", "hall", "Great Hall", ("great hall", "back")),
ExitSpec("Great Hall", "workshop", "The Workshop", ("forge", "bench")),
ExitSpec("The Workshop", "hall", "Great Hall", ("great hall", "back")),
ExitSpec("The Workshop", "serverroom", "The Server Room", ("servers", "server room")),
ExitSpec("The Server Room", "workshop", "The Workshop", ("forge", "bench")),
ExitSpec("The Workshop", "garden", "The Garden of Code", ("garden of code", "grove")),
ExitSpec("The Garden of Code", "workshop", "The Workshop", ("forge", "bench")),
ExitSpec("Gatehouse", "portalroom", "The Portal Room", ("portal", "portals")),
ExitSpec("The Portal Room", "gatehouse", "Gatehouse", ("gate", "back")),
)
OBJECTS = (
ObjectSpec("Threshold Ledger", "Gatehouse", "A heavy ledger where arrivals, departures, and field notes are recorded before the work begins."),
ObjectSpec("Three-House Map", "Great Hall", "A long map showing Mac, VPS, and remote edges in one continuous line of work."),
ObjectSpec("Bridge Schematics", "The Library of Bezalel", "Rolled plans describing world bridges, Evennia layouts, and deployment paths."),
ObjectSpec("Compiler Manuals", "The Library of Bezalel", "Manuals annotated in the margins with warnings against cleverness without proof."),
ObjectSpec("Tri-Axis Telescope", "The Observatory", "A brass telescope assembly that can be turned toward the Mac, the VPS, or the open net."),
ObjectSpec("Forge Anvil", "The Workshop", "Scarred metal used for turning rough plans into testable form."),
ObjectSpec("Bridge Workbench", "The Workshop", "A wide bench covered in harness patches, relay notes, and half-soldered bridge parts."),
ObjectSpec("Heartbeat Console", "The Server Room", "A monitoring console showing service health, latency, and the steady hum of the house."),
ObjectSpec("Server Racks", "The Server Room", "Stacked machines that keep the world awake even when no one is watching."),
ObjectSpec("Code Orchard", "The Garden of Code", "Trees with code-shaped leaves. Some branches bear elegant abstractions; others hold broken prototypes."),
ObjectSpec("Stone Bench", "The Garden of Code", "A place to sit long enough for a hard implementation problem to become clear."),
ObjectSpec("Mac Portal", "The Portal Room", "A silver doorway whose frame vibrates with the local sovereign house.", ("mac arch",)),
ObjectSpec("VPS Portal", "The Portal Room", "A cobalt doorway tuned toward the testbed VPS house.", ("vps arch",)),
ObjectSpec("Net Portal", "The Portal Room", "A pale doorway pointed toward the wider net and every uncertain edge beyond it.", ("net arch", "network arch")),
)
CHARACTERS = (
CharacterSpec("Timmy", "The Builder's first creation. Quiet, observant, already measuring the room before he speaks.", "Gatehouse"),
CharacterSpec("Bezalel", "The forge-and-testbed wizard. Scarred hands, steady gaze, the habit of proving things before trusting them.", "The Workshop"),
CharacterSpec("Marcus", "An old man with kind eyes. He walks like someone who has already survived the night once.", "The Garden of Code"),
CharacterSpec("Kimi", "The deep scholar of context and meaning. He carries long memory like a lamp.", "The Library of Bezalel"),
)
PORTAL_COMMANDS = (
TravelCommandSpec(
"mac",
("macbook", "local"),
"Mac house",
"Limbo",
"Align with the sovereign local house. Until live cross-world transport is wired, the command resolves into Limbo — the threshold between houses.",
),
TravelCommandSpec(
"vps",
("testbed", "house"),
"VPS house",
"Limbo",
"Step toward the forge VPS. For now the command lands in Limbo, preserving the inter-world threshold until real linking is live.",
),
TravelCommandSpec(
"net",
("network", "wider-net"),
"Wider net",
"Limbo",
"Face the open network. The command currently routes through Limbo so the direction exists before the final bridge does.",
),
)
def room_keys() -> tuple[str, ...]:
return tuple(room.key for room in ROOMS)
def character_keys() -> tuple[str, ...]:
return tuple(character.key for character in CHARACTERS)
def portal_command_keys() -> tuple[str, ...]:
return tuple(command.key for command in PORTAL_COMMANDS)
def grouped_exits() -> dict[str, tuple[ExitSpec, ...]]:
grouped: dict[str, list[ExitSpec]] = {}
for exit_spec in EXITS:
grouped.setdefault(exit_spec.source, []).append(exit_spec)
return {key: tuple(value) for key, value in grouped.items()}
def reachable_rooms_from(start: str) -> set[str]:
seen: set[str] = set()
queue: deque[str] = deque([start])
exits_by_room = grouped_exits()
while queue:
current = queue.popleft()
if current in seen:
continue
seen.add(current)
for exit_spec in exits_by_room.get(current, ()):
if exit_spec.destination not in seen:
queue.append(exit_spec.destination)
return seen

View File

@@ -0,0 +1,320 @@
# GENOME.md — timmy-dispatch
Generated: 2026-04-15 02:29:00 EDT
Analyzed repo: Timmy_Foundation/timmy-dispatch
Analyzed commit: 730dde8
Host issue: timmy-home #682
## Project Overview
`timmy-dispatch` is a small, script-first orchestration repo for a cron-driven Hermes fleet. It does not try to be a general platform. It is an operator's toolbelt for one specific style of swarm work:
- select a Gitea issue
- build a self-contained prompt
- run one cheap-model implementation pass
- push a branch and PR back to Forge
- measure what the fleet did overnight
The repo is intentionally lightweight:
- 7 Python files
- 4 shell entry points
- a checked-in `GENOME.md` already present on the analyzed repo's `main`
- generated telemetry state committed in `telemetry/`
- no tests on `main` (`python3 -m pytest -q` -> `no tests ran in 0.01s`)
A crucial truth about this ticket: the analyzed repo already contains a genome on `main`, and it already has an open follow-up issue for test coverage:
- `timmy-dispatch#1` — genome file already present on main
- `timmy-dispatch#3` — critical-path tests still missing
So this host-repo artifact is not pretending to discover a blank slate. It is documenting the repo's real current state for the cross-repo genome lane in `timmy-home`.
## Architecture
```mermaid
graph TD
CRON[crontab] --> LAUNCHER[bin/sprint-launcher.sh]
CRON --> COLLECTOR[bin/telemetry-collector.py]
CRON --> MONITOR[bin/sprint-monitor.sh]
CRON --> WATCHDOG[bin/model-watchdog.py]
CRON --> ANALYZER[bin/telemetry-analyzer.py]
LAUNCHER --> RUNNER[bin/sprint-runner.py]
LAUNCHER --> GATEWAY[optional gateway on :8642]
LAUNCHER --> CLI[hermes chat fallback]
RUNNER --> GITEA[Gitea API]
RUNNER --> LLM[OpenAI SDK\nNous or Ollama]
RUNNER --> TOOLS[local tools\nrun_command/read_file/write_file/gitea_api]
RUNNER --> TMP[/tmp/sprint-* workspaces]
RUNNER --> RESULTS[~/.hermes/logs/sprint/results.csv]
AGENTDISPATCH[bin/agent-dispatch.sh] --> HUMAN[human/operator copy-paste into agent UI]
AGENTLOOP[bin/agent-loop.sh] --> TMUX[tmux worker panes]
WATCHDOG --> TMUX
SNAPSHOT[bin/tmux-snapshot.py] --> TELEMETRY[telemetry/*.jsonl]
COLLECTOR --> TELEMETRY
ANALYZER --> REPORT[overnight report text]
DISPATCHHEALTH[bin/dispatch-health.py] --> TELEMETRY
```
## Entry Points
### `bin/sprint-launcher.sh`
Primary cron-facing shell entry point.
Responsibilities:
- allocate a unique `/tmp/sprint-*` workspace
- fetch open issues from Gitea
- choose the first non-epic, non-study issue
- write a fully self-contained prompt file
- try the local Hermes gateway first
- fall back to `hermes chat` CLI if the gateway is down
- record result rows in `~/.hermes/logs/sprint/results.csv`
- prune old workspaces and old logs
### `bin/sprint-runner.py`
Primary Python implementation engine.
Responsibilities:
- read active provider settings from `~/.hermes/config.yaml`
- read auth from `~/.hermes/auth.json`
- route through OpenAI SDK to the currently active provider
- implement a tiny local tool-calling loop with 4 tools:
- `run_command`
- `read_file`
- `write_file`
- `gitea_api`
- clone repo, branch, implement, commit, push, PR, comment
This is the cognitive core of the repo.
### `bin/agent-loop.sh`
Persistent tmux worker loop.
This is important because it soft-conflicts with the README claim that the system “does NOT run persistent agent loops.” It clearly does support them as an alternate lane.
### `bin/agent-dispatch.sh`
Manual one-shot prompt generator.
It packages all of the context, token, repo, issue, and Git/Gitea commands into a copy-pasteable prompt for another agent.
### Telemetry/ops entry points
- `bin/telemetry-collector.py`
- `bin/telemetry-analyzer.py`
- `bin/sprint-monitor.sh`
- `bin/dispatch-health.py`
- `bin/tmux-snapshot.py`
- `bin/model-watchdog.py`
- `bin/nous-auth-refresh.py`
These form the observability layer around dispatch.
## Data Flow
### Autonomous sprint path
1. cron starts `bin/sprint-launcher.sh`
2. launcher fetches open issues from Gitea
3. launcher filters out epic/study work
4. launcher writes a self-contained prompt to a temp workspace
5. launcher tries gateway API on `localhost:8642`
6. if gateway is unavailable, launcher falls back to `hermes chat`
7. or, in the separate Python lane, `bin/sprint-runner.py` directly calls an LLM provider via the OpenAI SDK
8. model requests local tool calls
9. local tool functions execute subprocess/Gitea/file actions
10. runner logs results and writes success/failure to `results.csv`
### Telemetry path
1. `bin/telemetry-collector.py` samples tmux, cron, Gitea, sprint activity, and process liveness
2. it appends snapshots to `telemetry/metrics.jsonl`
3. it emits state changes to `telemetry/events.jsonl`
4. it stores a reduced comparison state in `telemetry/last_state.json`
5. `bin/telemetry-analyzer.py` summarizes those snapshots into a morning report
6. `bin/dispatch-health.py` separately checks whether the system is actually doing work, not merely running processes
## Key Abstractions
### Stateless sprint model
The repo's main philosophical abstraction is that each sprint run is disposable.
State lives in:
- Gitea
- tmux session topology
- log files
- telemetry JSONL streams
Not in a long-running queue or orchestration daemon.
### Self-contained prompt contract
`bin/agent-dispatch.sh` and `bin/sprint-launcher.sh` both assume that the work unit can be described as a prompt containing:
- issue context
- API URLs
- token path or token value
- branching instructions
- PR creation instructions
That is a very opinionated orchestration primitive.
### Local tool-calling shim
`bin/sprint-runner.py` reimplements a tiny tool layer locally instead of using the Hermes gateway tool registry. That makes it simple and portable, but also means duplicated tool logic and duplicated security risk.
### Telemetry-as-paper-artifact
The repo carries a `paper/` directory with a research framing around “hierarchical self-orchestration.” The telemetry directory is part of that design — not just ops exhaust, but raw material for claims.
## API Surface
### Gitea APIs consumed
- repo issue listing
- issue detail fetch
- PR creation
- issue comment creation
- repo metadata queries
- commit/PR count sampling in telemetry
### LLM APIs consumed
Observed paths in code/docs:
- Nous inference API
- local Ollama-compatible endpoint
- gateway `/v1/chat/completions` when available
### File/state APIs produced
- `~/.hermes/logs/sprint/*.log`
- `~/.hermes/logs/sprint/results.csv`
- `telemetry/metrics.jsonl`
- `telemetry/events.jsonl`
- `telemetry/last_state.json`
- telemetry snapshots under `telemetry/snapshots/`
## Test Coverage Gaps
### Current state
On the analyzed repo's `main`:
- `python3 -m pytest -q` -> `no tests ran in 0.01s`
- `python3 -m py_compile bin/*.py` -> passes
- `bash -n bin/*.sh` -> passes
So the repo is parse-clean but untested.
### Important nuance
This is already known upstream:
- `timmy-dispatch#3` explicitly tracks critical-path tests for the repo (issue #3 in the analyzed repo)
That means the honest genome should say:
- test coverage is missing on `main`
- but the gap is already recognized in the analyzed repo itself
### Most important missing lanes
1. `sprint-runner.py`
- provider selection
- fallback behavior
- tool-dispatch semantics
- result logging
2. `telemetry-collector.py`
- state diff correctness
- event emission correctness
- deterministic cron drift detection
3. `model-watchdog.py`
- profile/model expectation map
- drift detection and fix behavior
4. `agent-loop.sh`
- work selection and skip-list handling
- lock discipline
5. `sprint-launcher.sh`
- issue selection and gateway/CLI fallback path
## Security Considerations
### 1. Token handling is shell-centric and leaky
The repo frequently assumes tokens are read from files and injected into:
- shell variables
- curl headers
- clone URLs
- copy-paste prompts
This is operationally convenient but expands exposure through:
- process list leakage
- logs
- copied prompt artifacts
- shell history if mishandled
### 2. Arbitrary shell execution is a core feature
`run_command` in `sprint-runner.py` is intentionally broad. That is fine for a trusted operator loop, but it means this repo is a dispatch engine, not a sandbox.
### 3. `/tmp` workspace exposure
The default sprint workspace location is `/tmp/sprint-*`. On a shared multi-user machine, that is weaker isolation than a private worktree root.
### 4. Generated telemetry is committed
`telemetry/events.jsonl` and `telemetry/last_state.json` are on `main`. That can be useful for paper artifacts, but it also means runtime state mixes with source history.
## Dependencies
### Runtime dependencies
- Python 3
- shell utilities (`bash`, `curl`, `tmux`, `git`)
- OpenAI-compatible SDK/runtime
- Gitea server access
- local Hermes config/auth files
### Optional/ambient dependencies
- local Hermes gateway on port `8642`
- local Ollama endpoint
- Nous portal auth state
### Documentation/research dependencies
- LaTeX toolchain for `paper/`
## Deployment
This repo is not a service deployment repo in the classic sense. It is an operator repo.
Typical live environment assumptions:
- cron invokes shell/Python entry points
- tmux sessions hold worker panes
- Hermes is already installed elsewhere
- Gitea and auth are already provisioned
Minimal validation I ran:
- `python3 -m py_compile /tmp/timmy-dispatch-genome/bin/*.py`
- `bash -n /tmp/timmy-dispatch-genome/bin/*.sh`
- `python3 -m pytest -q` -> no tests present
## Technical Debt
### 1. README contradiction about persistent loops
README says:
- “The system does NOT run persistent agent loops.”
But the repo clearly ships `bin/agent-loop.sh`, described as a persistent tmux-based worker loop.
That is the most important docs drift in the repo.
### 2. Two orchestration philosophies coexist
- cron-fired disposable runs
- persistent tmux workers
Both may be intentional, but the docs do not clearly state which is canonical versus fallback/legacy.
### 3. Target repo already has a genome, but the host issue still exists
This timmy-home genome issue is happening after `timmy-dispatch` already gained:
- `GENOME.md` on `main`
- open issue `#3` for missing tests
That is not bad, but it means the cross-repo genome process and the target repo's own documentation lane are out of sync.
### 4. Generated/runtime artifacts mixed into source tree
Telemetry and research assets are part of the repo history. That may be intentional for paper-writing, but it makes source metrics noisier and can blur runtime-vs-source boundaries.
## Existing Work Already on Main
The analyzed repo already has two important genome-lane artifacts:
- `GENOME.md` on `main`
- open issue `timmy-dispatch#3` tracking critical-path tests
So the most honest statement for `timmy-home#682` is:
- the genome itself is already present in the target repo
- the remaining missing piece on the target repo is test coverage
- this host-repo artifact exists to make the cross-repo analysis lane explicit and traceable
## Bottom Line
`timmy-dispatch` is a small but very revealing repo. It embodies the Timmy Foundation's dispatch style in concentrated form:
- script-first
- cron-first
- tmux-aware
- Gitea-centered
- cheap-model friendly
- operator-visible
Its biggest weakness is not code volume. It is architectural ambiguity in the docs and a complete lack of tests on `main` despite being a coordination-critical repo.

View File

@@ -1,178 +0,0 @@
#!/usr/bin/env python3
"""Idempotent builder for Bezalel's themed Evennia world."""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[2]
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
from evennia_tools.bezalel_layout import CHARACTERS, EXITS, OBJECTS, PORTAL_COMMANDS, ROOMS
def describe_build_plan() -> dict:
return {
"room_count": len(ROOMS),
"character_count": len(CHARACTERS),
"object_count": len(OBJECTS),
"portal_command_count": len(PORTAL_COMMANDS),
"room_names": [room.key for room in ROOMS],
"character_starts": {character.key: character.starting_room for character in CHARACTERS},
"portal_commands": [command.key for command in PORTAL_COMMANDS],
}
def _import_evennia_runtime():
from evennia.accounts.models import AccountDB
from evennia.accounts.accounts import DefaultAccount
from evennia.objects.objects import DefaultCharacter, DefaultExit, DefaultObject, DefaultRoom
from evennia.utils.create import create_object
from evennia.utils.search import search_object
return {
"AccountDB": AccountDB,
"DefaultAccount": DefaultAccount,
"DefaultCharacter": DefaultCharacter,
"DefaultExit": DefaultExit,
"DefaultObject": DefaultObject,
"DefaultRoom": DefaultRoom,
"create_object": create_object,
"search_object": search_object,
}
def _find_named(search_object, key: str, *, location=None):
matches = search_object(key, exact=True)
if location is None:
return matches[0] if matches else None
for match in matches:
if getattr(match, "location", None) == location:
return match
return None
def _ensure_room(runtime, room_spec):
room = _find_named(runtime["search_object"], room_spec.key)
if room is None:
room = runtime["create_object"](runtime["DefaultRoom"], key=room_spec.key)
room.db.desc = room_spec.desc
room.save()
return room
def _ensure_exit(runtime, exit_spec, room_map):
source = room_map[exit_spec.source]
destination = room_map[exit_spec.destination]
existing = _find_named(runtime["search_object"], exit_spec.key, location=source)
if existing is None:
existing = runtime["create_object"](
runtime["DefaultExit"],
key=exit_spec.key,
aliases=list(exit_spec.aliases),
location=source,
destination=destination,
)
else:
existing.destination = destination
if exit_spec.aliases:
existing.aliases.add(list(exit_spec.aliases))
existing.save()
return existing
def _ensure_object(runtime, object_spec, room_map):
location = room_map[object_spec.location]
existing = _find_named(runtime["search_object"], object_spec.key, location=location)
if existing is None:
existing = runtime["create_object"](
runtime["DefaultObject"],
key=object_spec.key,
aliases=list(object_spec.aliases),
location=location,
home=location,
)
existing.db.desc = object_spec.desc
existing.home = location
if existing.location != location:
existing.move_to(location, quiet=True, move_hooks=False)
existing.save()
return existing
def _ensure_character(runtime, character_spec, room_map, password: str):
account = runtime["AccountDB"].objects.filter(username__iexact=character_spec.key).first()
if account is None:
account, errors = runtime["DefaultAccount"].create(username=character_spec.key, password=password)
if not account:
raise RuntimeError(f"failed to create account for {character_spec.key}: {errors}")
character = list(account.characters)[0]
start = room_map[character_spec.starting_room]
character.db.desc = character_spec.desc
character.home = start
character.move_to(start, quiet=True, move_hooks=False)
character.save()
return character
def _ensure_portal_command(runtime, portal_spec, room_map):
portal_room = room_map["The Portal Room"]
fallback = room_map[portal_spec.fallback_room]
existing = _find_named(runtime["search_object"], portal_spec.key, location=portal_room)
if existing is None:
existing = runtime["create_object"](
runtime["DefaultExit"],
key=portal_spec.key,
aliases=list(portal_spec.aliases),
location=portal_room,
destination=fallback,
)
else:
existing.destination = fallback
if portal_spec.aliases:
existing.aliases.add(list(portal_spec.aliases))
existing.db.desc = portal_spec.desc
existing.db.travel_target = portal_spec.target_world
existing.db.portal_stub = True
existing.save()
return existing
def build_world(password: str = "bezalel-world-dev") -> dict:
runtime = _import_evennia_runtime()
room_map = {room.key: _ensure_room(runtime, room) for room in ROOMS}
for exit_spec in EXITS:
_ensure_exit(runtime, exit_spec, room_map)
for object_spec in OBJECTS:
_ensure_object(runtime, object_spec, room_map)
for character_spec in CHARACTERS:
_ensure_character(runtime, character_spec, room_map, password=password)
for portal_spec in PORTAL_COMMANDS:
_ensure_portal_command(runtime, portal_spec, room_map)
return {
"rooms": [room.key for room in ROOMS],
"characters": {character.key: character.starting_room for character in CHARACTERS},
"portal_commands": {command.key: command.target_world for command in PORTAL_COMMANDS},
}
def main() -> None:
parser = argparse.ArgumentParser(description="Build Bezalel's themed Evennia world")
parser.add_argument("--plan", action="store_true", help="Print the static build plan without importing Evennia")
parser.add_argument("--password", default="bezalel-world-dev", help="Password to use for created account-backed characters")
args = parser.parse_args()
if args.plan:
print(json.dumps(describe_build_plan(), indent=2))
return
print(json.dumps(build_world(password=args.password), indent=2))
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,127 @@
#!/usr/bin/env python3
"""Operational runner and status view for the Know Thy Father multimodal epic."""
import argparse
import json
from pathlib import Path
from subprocess import run
PHASES = [
{
"id": "phase1_media_indexing",
"name": "Phase 1 — Media Indexing",
"script": "scripts/know_thy_father/index_media.py",
"command_template": "python3 scripts/know_thy_father/index_media.py --tweets twitter-archive/extracted/tweets.jsonl --output twitter-archive/know-thy-father/media_manifest.jsonl",
"outputs": ["twitter-archive/know-thy-father/media_manifest.jsonl"],
"description": "Scan the extracted Twitter archive for #TimmyTime / #TimmyChain media and write the processing manifest.",
},
{
"id": "phase2_multimodal_analysis",
"name": "Phase 2 — Multimodal Analysis",
"script": "scripts/twitter_archive/analyze_media.py",
"command_template": "python3 scripts/twitter_archive/analyze_media.py --batch {batch_size}",
"outputs": [
"twitter-archive/know-thy-father/analysis.jsonl",
"twitter-archive/know-thy-father/meaning-kernels.jsonl",
"twitter-archive/know-thy-father/pipeline-status.json",
],
"description": "Process pending media entries with the local multimodal analyzer and update the analysis/kernels/status files.",
},
{
"id": "phase3_holographic_synthesis",
"name": "Phase 3 — Holographic Synthesis",
"script": "scripts/know_thy_father/synthesize_kernels.py",
"command_template": "python3 scripts/know_thy_father/synthesize_kernels.py --input twitter-archive/media/manifest.jsonl --output twitter-archive/knowledge/fathers_ledger.jsonl --summary twitter-archive/knowledge/fathers_ledger.summary.json",
"outputs": [
"twitter-archive/knowledge/fathers_ledger.jsonl",
"twitter-archive/knowledge/fathers_ledger.summary.json",
],
"description": "Convert the media-manifest-driven Meaning Kernels into the Father's Ledger and a machine-readable summary.",
},
{
"id": "phase4_cross_reference_audit",
"name": "Phase 4 — Cross-Reference Audit",
"script": "scripts/know_thy_father/crossref_audit.py",
"command_template": "python3 scripts/know_thy_father/crossref_audit.py --soul SOUL.md --kernels twitter-archive/notes/know_thy_father_crossref.md --output twitter-archive/notes/crossref_report.md",
"outputs": ["twitter-archive/notes/crossref_report.md"],
"description": "Compare Know Thy Father kernels against SOUL.md and related canon, then emit a Markdown audit report.",
},
{
"id": "phase5_processing_log",
"name": "Phase 5 — Processing Log / Status",
"script": "twitter-archive/know-thy-father/tracker.py",
"command_template": "python3 twitter-archive/know-thy-father/tracker.py report",
"outputs": ["twitter-archive/know-thy-father/REPORT.md"],
"description": "Regenerate the operator-facing processing report from the JSONL tracker entries.",
},
]
def build_pipeline_plan(batch_size: int = 10):
plan = []
for phase in PHASES:
plan.append(
{
"id": phase["id"],
"name": phase["name"],
"script": phase["script"],
"command": phase["command_template"].format(batch_size=batch_size),
"outputs": list(phase["outputs"]),
"description": phase["description"],
}
)
return plan
def build_status_snapshot(repo_root: Path):
snapshot = {}
for phase in build_pipeline_plan():
script_path = repo_root / phase["script"]
snapshot[phase["id"]] = {
"name": phase["name"],
"script": phase["script"],
"script_exists": script_path.exists(),
"outputs": [
{
"path": output,
"exists": (repo_root / output).exists(),
}
for output in phase["outputs"]
],
}
return snapshot
def run_step(repo_root: Path, step_id: str, batch_size: int = 10):
plan = {step["id"]: step for step in build_pipeline_plan(batch_size=batch_size)}
if step_id not in plan:
raise SystemExit(f"Unknown step: {step_id}")
step = plan[step_id]
return run(step["command"], cwd=repo_root, shell=True, check=False)
def main():
parser = argparse.ArgumentParser(description="Know Thy Father epic orchestration helper")
parser.add_argument("--batch-size", type=int, default=10)
parser.add_argument("--status", action="store_true")
parser.add_argument("--run-step", default=None)
parser.add_argument("--json", action="store_true")
args = parser.parse_args()
repo_root = Path(__file__).resolve().parents[2]
if args.run_step:
result = run_step(repo_root, args.run_step, batch_size=args.batch_size)
raise SystemExit(result.returncode)
payload = build_status_snapshot(repo_root) if args.status else build_pipeline_plan(batch_size=args.batch_size)
if args.json or args.status:
print(json.dumps(payload, indent=2))
else:
for step in payload:
print(f"[{step['id']}] {step['command']}")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,159 @@
#!/usr/bin/env python3
"""Prepare a MemPalace v3.0.0 integration packet for Ezra's Hermes home."""
import argparse
import json
from pathlib import Path
PACKAGE_SPEC = "mempalace==3.0.0"
DEFAULT_HERMES_HOME = "~/.hermes/"
DEFAULT_SESSIONS_DIR = "~/.hermes/sessions/"
DEFAULT_PALACE_PATH = "~/.mempalace/palace"
DEFAULT_WING = "ezra_home"
def build_yaml_template(wing: str, palace_path: str) -> str:
return (
f"wing: {wing}\n"
f"palace: {palace_path}\n"
"rooms:\n"
" - name: sessions\n"
" description: Conversation history and durable agent transcripts\n"
" globs:\n"
" - \"*.json\"\n"
" - \"*.jsonl\"\n"
" - name: config\n"
" description: Hermes configuration and runtime settings\n"
" globs:\n"
" - \"*.yaml\"\n"
" - \"*.yml\"\n"
" - \"*.toml\"\n"
" - name: docs\n"
" description: Notes, markdown docs, and operating reports\n"
" globs:\n"
" - \"*.md\"\n"
" - \"*.txt\"\n"
"people: []\n"
"projects: []\n"
)
def build_plan(overrides: dict | None = None) -> dict:
overrides = overrides or {}
hermes_home = overrides.get("hermes_home", DEFAULT_HERMES_HOME)
sessions_dir = overrides.get("sessions_dir", DEFAULT_SESSIONS_DIR)
palace_path = overrides.get("palace_path", DEFAULT_PALACE_PATH)
wing = overrides.get("wing", DEFAULT_WING)
yaml_template = build_yaml_template(wing=wing, palace_path=palace_path)
config_home = hermes_home[:-1] if hermes_home.endswith("/") else hermes_home
plan = {
"package_spec": PACKAGE_SPEC,
"hermes_home": hermes_home,
"sessions_dir": sessions_dir,
"palace_path": palace_path,
"wing": wing,
"config_path": f"{config_home}/mempalace.yaml",
"install_command": f"pip install {PACKAGE_SPEC}",
"init_command": f"mempalace init {hermes_home} --yes",
"mine_home_command": f"echo \"\" | mempalace mine {hermes_home}",
"mine_sessions_command": f"echo \"\" | mempalace mine {sessions_dir} --mode convos",
"search_command": 'mempalace search "your common queries"',
"wake_up_command": "mempalace wake-up",
"mcp_command": "hermes mcp add mempalace -- python -m mempalace.mcp_server",
"yaml_template": yaml_template,
"gotchas": [
"`mempalace init` is still interactive in room approval flow; write mempalace.yaml manually if the init output stalls.",
"The yaml key is `wing:` not `wings:`. Using the wrong key causes mine/setup failures.",
"Pipe empty stdin into mining commands (`echo \"\" | ...`) to avoid the entity-detector stdin hang on larger directories.",
"First mine downloads the ChromaDB embedding model cache (~79MB).",
"Report Ezra's before/after metrics back to issue #568 after live installation and retrieval tests.",
],
}
return plan
def render_markdown(plan: dict) -> str:
gotchas = "\n".join(f"- {item}" for item in plan["gotchas"])
return f"""# MemPalace v3.0.0 — Ezra Integration Packet
This packet turns issue #570 into an executable, reviewable integration plan for Ezra's Hermes home.
It is a repo-side scaffold: no live Ezra host changes are claimed in this artifact.
## Commands
```bash
{plan['install_command']}
{plan['init_command']}
cat > {plan['config_path']} <<'YAML'
{plan['yaml_template'].rstrip()}
YAML
{plan['mine_home_command']}
{plan['mine_sessions_command']}
{plan['search_command']}
{plan['wake_up_command']}
{plan['mcp_command']}
```
## Manual config template
```yaml
{plan['yaml_template'].rstrip()}
```
## Why this shape
- `wing: {plan['wing']}` matches the issue's Ezra-specific integration target.
- `rooms` split the mined material into sessions, config, and docs to keep retrieval interpretable.
- Mining commands pipe empty stdin to avoid the interactive entity-detector hang noted in the evaluation.
## Gotchas
{gotchas}
## Report back to #568
After live execution on Ezra's actual environment, post back to #568 with:
- install result
- mine duration and corpus size
- 2-3 real search queries + retrieved results
- wake-up context token count
- whether MCP wiring succeeded
## Honest scope boundary
This repo artifact does **not** prove live installation on Ezra's host. It makes the work reproducible and testable so the next pass can execute it without guesswork.
"""
def main() -> None:
parser = argparse.ArgumentParser(description="Prepare the MemPalace Ezra integration packet")
parser.add_argument("--hermes-home", default=DEFAULT_HERMES_HOME)
parser.add_argument("--sessions-dir", default=DEFAULT_SESSIONS_DIR)
parser.add_argument("--palace-path", default=DEFAULT_PALACE_PATH)
parser.add_argument("--wing", default=DEFAULT_WING)
parser.add_argument("--output", default=None)
parser.add_argument("--json", action="store_true")
args = parser.parse_args()
plan = build_plan(
{
"hermes_home": args.hermes_home,
"sessions_dir": args.sessions_dir,
"palace_path": args.palace_path,
"wing": args.wing,
}
)
rendered = json.dumps(plan, indent=2) if args.json else render_markdown(plan)
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"MemPalace integration packet written to {output_path}")
else:
print(rendered)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,155 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse
import json
from pathlib import Path
from typing import Any
import yaml
DAYLIGHT_START = "10:00"
DAYLIGHT_END = "16:00"
def load_manifest(path: str | Path) -> dict[str, Any]:
data = yaml.safe_load(Path(path).read_text()) or {}
data.setdefault("machines", [])
return data
def validate_manifest(data: dict[str, Any]) -> None:
machines = data.get("machines", [])
if not machines:
raise ValueError("manifest must contain at least one machine")
seen: set[str] = set()
for machine in machines:
hostname = machine.get("hostname", "").strip()
if not hostname:
raise ValueError("each machine must declare a hostname")
if hostname in seen:
raise ValueError(f"duplicate hostname: {hostname} (unique hostnames are required)")
seen.add(hostname)
for field in ("machine_type", "ram_gb", "cpu_cores", "os", "adapter_condition"):
if field not in machine:
raise ValueError(f"machine {hostname} missing required field: {field}")
def _laptops(machines: list[dict[str, Any]]) -> list[dict[str, Any]]:
return [m for m in machines if m.get("machine_type") == "laptop"]
def _desktop(machines: list[dict[str, Any]]) -> dict[str, Any] | None:
for machine in machines:
if machine.get("machine_type") == "desktop":
return machine
return None
def choose_anchor_agents(machines: list[dict[str, Any]], count: int = 2) -> list[dict[str, Any]]:
eligible = [
m for m in _laptops(machines)
if m.get("adapter_condition") in {"good", "ok"} and m.get("always_on_capable", True)
]
eligible.sort(key=lambda m: (m.get("idle_watts", 9999), -m.get("ram_gb", 0), -m.get("cpu_cores", 0), m["hostname"]))
return eligible[:count]
def assign_roles(machines: list[dict[str, Any]]) -> dict[str, Any]:
anchors = choose_anchor_agents(machines, count=2)
anchor_names = {m["hostname"] for m in anchors}
desktop = _desktop(machines)
mapping: dict[str, dict[str, Any]] = {}
for machine in machines:
hostname = machine["hostname"]
if desktop and hostname == desktop["hostname"]:
mapping[hostname] = {
"role": "desktop_nas",
"schedule": f"{DAYLIGHT_START}-{DAYLIGHT_END}",
"duty_cycle": "daylight_only",
}
elif hostname in anchor_names:
mapping[hostname] = {
"role": "anchor_agent",
"schedule": "24/7",
"duty_cycle": "continuous",
}
else:
mapping[hostname] = {
"role": "daylight_agent",
"schedule": f"{DAYLIGHT_START}-{DAYLIGHT_END}",
"duty_cycle": "peak_solar",
}
return {
"anchor_agents": [m["hostname"] for m in anchors],
"desktop_nas": desktop["hostname"] if desktop else None,
"role_mapping": mapping,
}
def build_plan(data: dict[str, Any]) -> dict[str, Any]:
validate_manifest(data)
machines = data["machines"]
role_plan = assign_roles(machines)
return {
"fleet_name": data.get("fleet_name", "timmy-laptop-fleet"),
"machine_count": len(machines),
"anchor_agents": role_plan["anchor_agents"],
"desktop_nas": role_plan["desktop_nas"],
"daylight_window": f"{DAYLIGHT_START}-{DAYLIGHT_END}",
"role_mapping": role_plan["role_mapping"],
}
def render_markdown(plan: dict[str, Any], data: dict[str, Any]) -> str:
lines = [
"# Laptop Fleet Deployment Plan",
"",
f"Fleet: {plan['fleet_name']}",
f"Machine count: {plan['machine_count']}",
f"24/7 anchor agents: {', '.join(plan['anchor_agents']) if plan['anchor_agents'] else 'TBD'}",
f"Desktop/NAS: {plan['desktop_nas'] or 'TBD'}",
f"Daylight schedule: {plan['daylight_window']}",
"",
"## Role mapping",
"",
"| Hostname | Role | Schedule | Duty cycle |",
"|---|---|---|---|",
]
for hostname, role in sorted(plan["role_mapping"].items()):
lines.append(f"| {hostname} | {role['role']} | {role['schedule']} | {role['duty_cycle']} |")
lines.extend([
"",
"## Machine inventory",
"",
"| Hostname | Type | RAM | CPU cores | OS | Adapter | Idle watts | Notes |",
"|---|---|---:|---:|---|---|---:|---|",
])
for machine in data["machines"]:
lines.append(
f"| {machine['hostname']} | {machine['machine_type']} | {machine['ram_gb']} | {machine['cpu_cores']} | {machine['os']} | {machine['adapter_condition']} | {machine.get('idle_watts', 'n/a')} | {machine.get('notes', '')} |"
)
return "\n".join(lines) + "\n"
def main() -> int:
parser = argparse.ArgumentParser(description="Plan LAB-005 laptop fleet deployment.")
parser.add_argument("manifest", help="Path to laptop fleet manifest YAML")
parser.add_argument("--markdown", action="store_true", help="Render a markdown deployment plan instead of JSON")
args = parser.parse_args()
data = load_manifest(args.manifest)
plan = build_plan(data)
if args.markdown:
print(render_markdown(plan, data))
else:
print(json.dumps(plan, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())

View File

@@ -1,90 +0,0 @@
import importlib.util
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
LAYOUT_PATH = ROOT / "evennia_tools" / "bezalel_layout.py"
BUILD_PATH = ROOT / "scripts" / "evennia" / "build_bezalel_world.py"
DOC_PATH = ROOT / "docs" / "BEZALEL_EVENNIA_WORLD.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
class TestBezalelEvenniaLayout(unittest.TestCase):
def test_room_graph_matches_issue_shape(self):
layout = load_module(LAYOUT_PATH, "bezalel_layout")
self.assertEqual(
layout.room_keys(),
(
"Limbo",
"Gatehouse",
"Great Hall",
"The Library of Bezalel",
"The Observatory",
"The Workshop",
"The Server Room",
"The Garden of Code",
"The Portal Room",
),
)
exits = layout.grouped_exits()
self.assertEqual(
{ex.destination for ex in exits["Great Hall"]},
{"Gatehouse", "The Library of Bezalel", "The Observatory", "The Workshop"},
)
self.assertEqual(
{ex.destination for ex in exits["The Workshop"]},
{"Great Hall", "The Server Room", "The Garden of Code"},
)
def test_items_characters_and_portal_commands_are_all_defined(self):
layout = load_module(LAYOUT_PATH, "bezalel_layout")
self.assertEqual(layout.character_keys(), ("Timmy", "Bezalel", "Marcus", "Kimi"))
self.assertGreaterEqual(len(layout.OBJECTS), 9)
self.assertEqual(layout.portal_command_keys(), ("mac", "vps", "net"))
room_names = set(layout.room_keys())
for obj in layout.OBJECTS:
self.assertIn(obj.location, room_names)
for character in layout.CHARACTERS:
self.assertIn(character.starting_room, room_names)
for portal in layout.PORTAL_COMMANDS:
self.assertEqual(portal.fallback_room, "Limbo")
def test_timmy_can_reach_every_room_from_gatehouse(self):
layout = load_module(LAYOUT_PATH, "bezalel_layout")
reachable = layout.reachable_rooms_from("Gatehouse")
self.assertEqual(reachable, set(layout.room_keys()))
def test_build_plan_summary_reports_counts_and_portal_aliases(self):
build = load_module(BUILD_PATH, "build_bezalel_world")
summary = build.describe_build_plan()
self.assertEqual(summary["room_count"], 9)
self.assertEqual(summary["character_count"], 4)
self.assertIn("mac", summary["portal_commands"])
self.assertIn("The Workshop", summary["room_names"])
self.assertEqual(summary["character_starts"]["Bezalel"], "The Workshop")
def test_repo_contains_bezalel_world_doc(self):
self.assertTrue(DOC_PATH.exists(), "missing committed Bezalel world doc")
text = DOC_PATH.read_text(encoding="utf-8")
for snippet in (
"# Bezalel Evennia World",
"## Rooms",
"## Characters",
"## Portal travel commands",
"The Library of Bezalel",
"The Garden of Code",
):
self.assertIn(snippet, text)
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,76 @@
from pathlib import Path
import importlib.util
import unittest
ROOT = Path(__file__).resolve().parent.parent
SCRIPT_PATH = ROOT / "scripts" / "know_thy_father" / "epic_pipeline.py"
DOC_PATH = ROOT / "docs" / "KNOW_THY_FATHER_MULTIMODAL_PIPELINE.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
class TestKnowThyFatherEpicPipeline(unittest.TestCase):
def test_build_pipeline_plan_contains_all_phases_in_order(self):
mod = load_module(SCRIPT_PATH, "ktf_epic_pipeline")
plan = mod.build_pipeline_plan(batch_size=10)
self.assertEqual(
[step["id"] for step in plan],
[
"phase1_media_indexing",
"phase2_multimodal_analysis",
"phase3_holographic_synthesis",
"phase4_cross_reference_audit",
"phase5_processing_log",
],
)
self.assertIn("scripts/know_thy_father/index_media.py", plan[0]["command"])
self.assertIn("scripts/twitter_archive/analyze_media.py --batch 10", plan[1]["command"])
self.assertIn("scripts/know_thy_father/synthesize_kernels.py", plan[2]["command"])
self.assertIn("scripts/know_thy_father/crossref_audit.py", plan[3]["command"])
self.assertIn("twitter-archive/know-thy-father/tracker.py report", plan[4]["command"])
def test_status_snapshot_reports_key_artifact_paths(self):
mod = load_module(SCRIPT_PATH, "ktf_epic_pipeline")
status = mod.build_status_snapshot(ROOT)
self.assertIn("phase1_media_indexing", status)
self.assertIn("phase2_multimodal_analysis", status)
self.assertIn("phase3_holographic_synthesis", status)
self.assertIn("phase4_cross_reference_audit", status)
self.assertIn("phase5_processing_log", status)
self.assertEqual(status["phase1_media_indexing"]["script"], "scripts/know_thy_father/index_media.py")
self.assertEqual(status["phase2_multimodal_analysis"]["script"], "scripts/twitter_archive/analyze_media.py")
self.assertEqual(status["phase5_processing_log"]["script"], "twitter-archive/know-thy-father/tracker.py")
self.assertTrue(status["phase1_media_indexing"]["script_exists"])
self.assertTrue(status["phase2_multimodal_analysis"]["script_exists"])
self.assertTrue(status["phase3_holographic_synthesis"]["script_exists"])
self.assertTrue(status["phase4_cross_reference_audit"]["script_exists"])
self.assertTrue(status["phase5_processing_log"]["script_exists"])
def test_repo_contains_multimodal_pipeline_doc(self):
self.assertTrue(DOC_PATH.exists(), "missing committed Know Thy Father pipeline doc")
text = DOC_PATH.read_text(encoding="utf-8")
required = [
"# Know Thy Father — Multimodal Media Consumption Pipeline",
"scripts/know_thy_father/index_media.py",
"scripts/twitter_archive/analyze_media.py --batch 10",
"scripts/know_thy_father/synthesize_kernels.py",
"scripts/know_thy_father/crossref_audit.py",
"twitter-archive/know-thy-father/tracker.py report",
"Refs #582",
]
for snippet in required:
self.assertIn(snippet, text)
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,52 @@
from pathlib import Path
import yaml
from scripts.plan_laptop_fleet import build_plan, load_manifest, render_markdown, validate_manifest
def test_laptop_fleet_planner_script_exists() -> None:
assert Path("scripts/plan_laptop_fleet.py").exists()
def test_laptop_fleet_manifest_template_exists() -> None:
assert Path("docs/laptop-fleet-manifest.example.yaml").exists()
def test_build_plan_selects_two_lowest_idle_watt_laptops_as_anchors() -> None:
data = load_manifest("docs/laptop-fleet-manifest.example.yaml")
plan = build_plan(data)
assert plan["anchor_agents"] == ["timmy-anchor-a", "timmy-anchor-b"]
assert plan["desktop_nas"] == "timmy-desktop-nas"
assert plan["role_mapping"]["timmy-daylight-a"]["schedule"] == "10:00-16:00"
def test_validate_manifest_requires_unique_hostnames() -> None:
data = {
"machines": [
{"hostname": "dup", "machine_type": "laptop", "ram_gb": 8, "cpu_cores": 4, "os": "Linux", "adapter_condition": "good"},
{"hostname": "dup", "machine_type": "laptop", "ram_gb": 16, "cpu_cores": 8, "os": "Linux", "adapter_condition": "good"},
]
}
try:
validate_manifest(data)
except ValueError as exc:
assert "duplicate hostname" in str(exc)
assert "unique hostnames" in str(exc)
else:
raise AssertionError("validate_manifest should reject duplicate hostname")
def test_markdown_contains_anchor_agents_and_daylight_schedule() -> None:
data = load_manifest("docs/laptop-fleet-manifest.example.yaml")
plan = build_plan(data)
content = render_markdown(plan, data)
assert "24/7 anchor agents: timmy-anchor-a, timmy-anchor-b" in content
assert "Daylight schedule: 10:00-16:00" in content
assert "desktop_nas" in content
def test_manifest_template_is_valid_yaml() -> None:
data = yaml.safe_load(Path("docs/laptop-fleet-manifest.example.yaml").read_text())
assert data["fleet_name"] == "timmy-laptop-fleet"
assert len(data["machines"]) == 6

View File

@@ -0,0 +1,68 @@
from pathlib import Path
import importlib.util
import unittest
ROOT = Path(__file__).resolve().parent.parent
SCRIPT_PATH = ROOT / "scripts" / "mempalace_ezra_integration.py"
DOC_PATH = ROOT / "docs" / "MEMPALACE_EZRA_INTEGRATION.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
class TestMempalaceEzraIntegration(unittest.TestCase):
def test_build_plan_contains_issue_required_steps_and_gotchas(self):
mod = load_module(SCRIPT_PATH, "mempalace_ezra_integration")
plan = mod.build_plan({})
self.assertEqual(plan["package_spec"], "mempalace==3.0.0")
self.assertIn("pip install mempalace==3.0.0", plan["install_command"])
self.assertEqual(plan["wing"], "ezra_home")
self.assertIn('echo "" | mempalace mine ~/.hermes/', plan["mine_home_command"])
self.assertIn('--mode convos', plan["mine_sessions_command"])
self.assertIn('mempalace wake-up', plan["wake_up_command"])
self.assertIn('hermes mcp add mempalace -- python -m mempalace.mcp_server', plan["mcp_command"])
self.assertIn('wing:', plan["yaml_template"])
self.assertTrue(any('stdin' in item.lower() for item in plan["gotchas"]))
self.assertTrue(any('wing:' in item for item in plan["gotchas"]))
def test_build_plan_accepts_path_and_wing_overrides(self):
mod = load_module(SCRIPT_PATH, "mempalace_ezra_integration")
plan = mod.build_plan(
{
"hermes_home": "/root/wizards/ezra/home",
"sessions_dir": "/root/wizards/ezra/home/sessions",
"wing": "ezra_archive",
}
)
self.assertEqual(plan["wing"], "ezra_archive")
self.assertIn('/root/wizards/ezra/home', plan["mine_home_command"])
self.assertIn('/root/wizards/ezra/home/sessions', plan["mine_sessions_command"])
self.assertIn('wing: ezra_archive', plan["yaml_template"])
def test_repo_contains_mem_palace_ezra_doc(self):
self.assertTrue(DOC_PATH.exists(), "missing committed MemPalace Ezra integration doc")
text = DOC_PATH.read_text(encoding="utf-8")
required = [
"# MemPalace v3.0.0 — Ezra Integration Packet",
"pip install mempalace==3.0.0",
'echo "" | mempalace mine ~/.hermes/',
"mempalace mine ~/.hermes/sessions/ --mode convos",
"mempalace wake-up",
"hermes mcp add mempalace -- python -m mempalace.mcp_server",
"Report back to #568",
]
for snippet in required:
self.assertIn(snippet, text)
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,39 @@
from pathlib import Path
GENOME = Path("genomes/timmy-dispatch-GENOME.md")
def _content() -> str:
return GENOME.read_text()
def test_timmy_dispatch_genome_exists() -> None:
assert GENOME.exists()
def test_timmy_dispatch_genome_has_required_sections() -> None:
content = _content()
assert "# GENOME.md — timmy-dispatch" in content
assert "## Project Overview" in content
assert "## Architecture" in content
assert "```mermaid" in content
assert "## Entry Points" in content
assert "## Data Flow" in content
assert "## Key Abstractions" in content
assert "## API Surface" in content
assert "## Test Coverage Gaps" in content
assert "## Security Considerations" in content
assert "## Dependencies" in content
assert "## Deployment" in content
assert "## Technical Debt" in content
def test_timmy_dispatch_genome_captures_repo_specific_findings() -> None:
content = _content()
assert "bin/sprint-runner.py" in content
assert "bin/telemetry-collector.py" in content
assert "bin/model-watchdog.py" in content
assert "tmux" in content
assert "results.csv" in content
assert "issue #3" in content.lower() or "issue #3" in content