Compare commits

..

12 Commits

Author SHA1 Message Date
372ffa3fdf feat: Codebase Genome for the-nexus (#672)
Some checks failed
Smoke Test / smoke (pull_request) Failing after 17s
Complete GENOME.md for the-nexus (3D world + MUD + memory):
- Project overview: Three.js + Evennia + MemPalace
- Architecture diagram (Mermaid)
- 10 key subsystems documented
- Entry points (browser, server, electron, deploy)
- MemPalace system breakdown
- Evennia integration details
- Configuration and documentation index
- Sovereignty assessment

Repo 5/16. Closes #672.
2026-04-15 21:04:59 -04:00
f684b0deb8 feat: Codebase Genome for turboquant (#679)
Some checks failed
Smoke Test / smoke (pull_request) Failing after 17s
Complete GENOME.md for turboquant (KV cache compression):
- Project overview: PolarQuant + QJL = 3.5bit/channel
- Architecture diagram (Mermaid)
- Entry points and data flow
- Key abstractions (encode/decode/Metal shaders)
- File index (~660 LOC)
- Upstream source repos
- Test coverage
- Sovereignty assessment

Repo 12/16. Closes #679.
2026-04-15 20:59:55 -04:00
f76c8187cf docs: triage cadence report for #685
Some checks failed
Smoke Test / smoke (pull_request) Failing after 15s
Backlog reduced from 220 to 50. Report documents triage
cadence needed to maintain healthy backlog.

- Daily: 5 min new issue check
- Weekly: 15 min full sweep
- Monthly: 30 min audit

Closes #685.
2026-04-15 20:55:22 -04:00
10fd467b28 Merge pull request 'fix: resolve v2 harness import collision with explicit path loading (#716)' (#748) from burn/716-1776264183 into main 2026-04-15 16:04:04 +00:00
ba2d365669 fix: resolve v2 harness import collision with explicit path loading (closes #716)
Some checks failed
Smoke Test / smoke (pull_request) Failing after 18s
2026-04-15 11:46:37 -04:00
5a696c184e Merge pull request 'feat: add NH Broadband install packet scaffold (closes #740)' (#741) from sprint/issue-740 into main 2026-04-15 11:57:34 +00:00
Alexander Whitestone
90d8daedcf feat: add NH Broadband install packet scaffold (closes #740)
Some checks failed
Smoke Test / smoke (pull_request) Failing after 19s
2026-04-15 07:33:01 -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
28 changed files with 1611 additions and 1074 deletions

141
GENOME.md
View File

@@ -1,141 +0,0 @@
# GENOME.md — Timmy_Foundation/timmy-home
Generated by `pipelines/codebase_genome.py`.
## Project Overview
Timmy Foundation's home repository for development operations and configurations.
- Text files indexed: 3004
- Source and script files: 186
- Test files: 28
- Documentation files: 701
## Architecture
```mermaid
graph TD
repo_root["repo"]
angband["angband"]
briefings["briefings"]
config["config"]
conftest["conftest"]
evennia["evennia"]
evennia_tools["evennia_tools"]
evolution["evolution"]
gemini_fallback_setup["gemini-fallback-setup"]
heartbeat["heartbeat"]
infrastructure["infrastructure"]
repo_root --> angband
repo_root --> briefings
repo_root --> config
repo_root --> conftest
repo_root --> evennia
repo_root --> evennia_tools
```
## Entry Points
- `gemini-fallback-setup.sh` — operational script (`bash gemini-fallback-setup.sh`)
- `morrowind/hud.sh` — operational script (`bash morrowind/hud.sh`)
- `pipelines/codebase_genome.py` — python main guard (`python3 pipelines/codebase_genome.py`)
- `scripts/auto_restart_agent.sh` — operational script (`bash scripts/auto_restart_agent.sh`)
- `scripts/backup_pipeline.sh` — operational script (`bash scripts/backup_pipeline.sh`)
- `scripts/big_brain_manager.py` — operational script (`python3 scripts/big_brain_manager.py`)
- `scripts/big_brain_repo_audit.py` — operational script (`python3 scripts/big_brain_repo_audit.py`)
- `scripts/codebase_genome_nightly.py` — operational script (`python3 scripts/codebase_genome_nightly.py`)
- `scripts/detect_secrets.py` — operational script (`python3 scripts/detect_secrets.py`)
- `scripts/dynamic_dispatch_optimizer.py` — operational script (`python3 scripts/dynamic_dispatch_optimizer.py`)
- `scripts/emacs-fleet-bridge.py` — operational script (`python3 scripts/emacs-fleet-bridge.py`)
- `scripts/emacs-fleet-poll.sh` — operational script (`bash scripts/emacs-fleet-poll.sh`)
## Data Flow
1. Operators enter through `gemini-fallback-setup.sh`, `morrowind/hud.sh`, `pipelines/codebase_genome.py`.
2. Core logic fans into top-level components: `angband`, `briefings`, `config`, `conftest`, `evennia`, `evennia_tools`.
3. Validation is incomplete around `wizards/allegro/home/skills/red-teaming/godmode/scripts/auto_jailbreak.py`, `timmy-local/cache/agent_cache.py`, `wizards/allegro/home/skills/red-teaming/godmode/scripts/parseltongue.py`, so changes there carry regression risk.
4. Final artifacts land as repository files, docs, or runtime side effects depending on the selected entry point.
## Key Abstractions
- `evennia/timmy_world/game.py` — classes `World`:91, `ActionSystem`:421, `TimmyAI`:539, `NPCAI`:550; functions `get_narrative_phase()`:55, `get_phase_transition_event()`:65
- `evennia/timmy_world/world/game.py` — classes `World`:19, `ActionSystem`:326, `TimmyAI`:444, `NPCAI`:455; functions none detected
- `timmy-world/game.py` — classes `World`:19, `ActionSystem`:349, `TimmyAI`:467, `NPCAI`:478; functions none detected
- `wizards/allegro/home/skills/red-teaming/godmode/scripts/auto_jailbreak.py` — classes none detected; functions none detected
- `uniwizard/self_grader.py` — classes `SessionGrade`:23, `WeeklyReport`:55, `SelfGrader`:74; functions `main()`:713
- `uni-wizard/v3/intelligence_engine.py` — classes `ExecutionPattern`:27, `ModelPerformance`:44, `AdaptationEvent`:58, `PatternDatabase`:69; functions none detected
- `scripts/know_thy_father/crossref_audit.py` — classes `ThemeCategory`:30, `Principle`:160, `MeaningKernel`:169, `CrossRefFinding`:178; functions `extract_themes_from_text()`:192, `parse_soul_md()`:206, `parse_kernels()`:264, `cross_reference()`:296, `generate_report()`:440, `main()`:561
- `timmy-local/cache/agent_cache.py` — classes `CacheStats`:28, `LRUCache`:52, `ResponseCache`:94, `ToolCache`:205; functions none detected
## API Surface
- CLI: `bash gemini-fallback-setup.sh` — operational script (`gemini-fallback-setup.sh`)
- CLI: `bash morrowind/hud.sh` — operational script (`morrowind/hud.sh`)
- CLI: `python3 pipelines/codebase_genome.py` — python main guard (`pipelines/codebase_genome.py`)
- CLI: `bash scripts/auto_restart_agent.sh` — operational script (`scripts/auto_restart_agent.sh`)
- CLI: `bash scripts/backup_pipeline.sh` — operational script (`scripts/backup_pipeline.sh`)
- CLI: `python3 scripts/big_brain_manager.py` — operational script (`scripts/big_brain_manager.py`)
- CLI: `python3 scripts/big_brain_repo_audit.py` — operational script (`scripts/big_brain_repo_audit.py`)
- CLI: `python3 scripts/codebase_genome_nightly.py` — operational script (`scripts/codebase_genome_nightly.py`)
- Python: `get_narrative_phase()` from `evennia/timmy_world/game.py:55`
- Python: `get_phase_transition_event()` from `evennia/timmy_world/game.py:65`
- Python: `main()` from `uniwizard/self_grader.py:713`
## Test Coverage Report
- Source and script files inspected: 186
- Test files inspected: 28
- Coverage gaps:
- `wizards/allegro/home/skills/red-teaming/godmode/scripts/auto_jailbreak.py` — no matching test reference detected
- `timmy-local/cache/agent_cache.py` — no matching test reference detected
- `wizards/allegro/home/skills/red-teaming/godmode/scripts/parseltongue.py` — no matching test reference detected
- `twitter-archive/multimodal_pipeline.py` — no matching test reference detected
- `wizards/allegro/home/skills/red-teaming/godmode/scripts/godmode_race.py` — no matching test reference detected
- `skills/productivity/google-workspace/scripts/google_api.py` — no matching test reference detected
- `wizards/allegro/home/skills/productivity/google-workspace/scripts/google_api.py` — no matching test reference detected
- `morrowind/pilot.py` — no matching test reference detected
- `morrowind/mcp_server.py` — no matching test reference detected
- `skills/research/domain-intel/scripts/domain_intel.py` — no matching test reference detected
- `wizards/allegro/home/skills/research/domain-intel/scripts/domain_intel.py` — no matching test reference detected
- `timmy-local/scripts/ingest.py` — no matching test reference detected
## Security Audit Findings
- [medium] `briefings/briefing_20260325.json:37` — hardcoded http endpoint: plaintext or fixed HTTP endpoints can drift or leak across environments. Evidence: `"gitea_error": "Gitea 404: {\"errors\":null,\"message\":\"not found\",\"url\":\"http://143.198.27.163:3000/api/swagger\"}\n [http://143.198.27.163:3000/api/v1/repos/Timmy_Foundation/sovereign-orchestration/issues?state=open&type=issues&sort=created&direction=desc&limit=1&page=1]",`
- [medium] `briefings/briefing_20260328.json:11` — hardcoded http endpoint: plaintext or fixed HTTP endpoints can drift or leak across environments. Evidence: `"provider_base_url": "http://localhost:8081/v1",`
- [medium] `briefings/briefing_20260329.json:11` — hardcoded http endpoint: plaintext or fixed HTTP endpoints can drift or leak across environments. Evidence: `"provider_base_url": "http://localhost:8081/v1",`
- [medium] `config.yaml:37` — hardcoded http endpoint: plaintext or fixed HTTP endpoints can drift or leak across environments. Evidence: `summary_base_url: http://localhost:11434/v1`
- [medium] `config.yaml:47` — hardcoded http endpoint: plaintext or fixed HTTP endpoints can drift or leak across environments. Evidence: `base_url: 'http://localhost:11434/v1'`
- [medium] `config.yaml:52` — hardcoded http endpoint: plaintext or fixed HTTP endpoints can drift or leak across environments. Evidence: `base_url: 'http://localhost:11434/v1'`
- [medium] `config.yaml:57` — hardcoded http endpoint: plaintext or fixed HTTP endpoints can drift or leak across environments. Evidence: `base_url: 'http://localhost:11434/v1'`
- [medium] `config.yaml:62` — hardcoded http endpoint: plaintext or fixed HTTP endpoints can drift or leak across environments. Evidence: `base_url: 'http://localhost:11434/v1'`
- [medium] `config.yaml:67` — hardcoded http endpoint: plaintext or fixed HTTP endpoints can drift or leak across environments. Evidence: `base_url: 'http://localhost:11434/v1'`
- [medium] `config.yaml:77` — hardcoded http endpoint: plaintext or fixed HTTP endpoints can drift or leak across environments. Evidence: `base_url: 'http://localhost:11434/v1'`
- [medium] `config.yaml:82` — hardcoded http endpoint: plaintext or fixed HTTP endpoints can drift or leak across environments. Evidence: `base_url: 'http://localhost:11434/v1'`
- [medium] `config.yaml:174` — hardcoded http endpoint: plaintext or fixed HTTP endpoints can drift or leak across environments. Evidence: `base_url: http://localhost:11434/v1`
## Dead Code Candidates
- `wizards/allegro/home/skills/red-teaming/godmode/scripts/auto_jailbreak.py` — not imported by indexed Python modules and not referenced by tests
- `timmy-local/cache/agent_cache.py` — not imported by indexed Python modules and not referenced by tests
- `wizards/allegro/home/skills/red-teaming/godmode/scripts/parseltongue.py` — not imported by indexed Python modules and not referenced by tests
- `twitter-archive/multimodal_pipeline.py` — not imported by indexed Python modules and not referenced by tests
- `wizards/allegro/home/skills/red-teaming/godmode/scripts/godmode_race.py` — not imported by indexed Python modules and not referenced by tests
- `skills/productivity/google-workspace/scripts/google_api.py` — not imported by indexed Python modules and not referenced by tests
- `wizards/allegro/home/skills/productivity/google-workspace/scripts/google_api.py` — not imported by indexed Python modules and not referenced by tests
- `morrowind/pilot.py` — not imported by indexed Python modules and not referenced by tests
- `morrowind/mcp_server.py` — not imported by indexed Python modules and not referenced by tests
- `skills/research/domain-intel/scripts/domain_intel.py` — not imported by indexed Python modules and not referenced by tests
## Performance Bottleneck Analysis
- `angband/mcp_server.py` — large module (353 lines) likely hides multiple responsibilities
- `evennia/timmy_world/game.py` — large module (1541 lines) likely hides multiple responsibilities
- `evennia/timmy_world/world/game.py` — large module (1345 lines) likely hides multiple responsibilities
- `morrowind/mcp_server.py` — large module (451 lines) likely hides multiple responsibilities
- `morrowind/pilot.py` — large module (459 lines) likely hides multiple responsibilities
- `pipelines/codebase_genome.py` — large module (557 lines) likely hides multiple responsibilities
- `scripts/know_thy_father/crossref_audit.py` — large module (657 lines) likely hides multiple responsibilities
- `scripts/know_thy_father/index_media.py` — large module (405 lines) likely hides multiple responsibilities
- `scripts/know_thy_father/synthesize_kernels.py` — large module (416 lines) likely hides multiple responsibilities
- `scripts/tower_game.py` — large module (395 lines) likely hides multiple responsibilities

View File

@@ -1,79 +0,0 @@
# Codebase Genome Pipeline
Issue: `timmy-home#665`
This pipeline gives Timmy a repeatable way to generate a deterministic `GENOME.md` for any repository and rotate through the org nightly.
## What landed
- `pipelines/codebase_genome.py` — static analyzer that writes `GENOME.md`
- `pipelines/codebase-genome.py` — thin CLI wrapper matching the expected pipeline-style entrypoint
- `scripts/codebase_genome_nightly.py` — org-aware nightly runner that selects the next repo, updates a local checkout, and writes the genome artifact
- `GENOME.md` — generated analysis for `timmy-home` itself
## Genome output
Each generated `GENOME.md` includes:
- project overview and repository size metrics
- Mermaid architecture diagram
- entry points and API surface
- data flow summary
- key abstractions from Python source
- test coverage gaps
- security audit findings
- dead code candidates
- performance bottleneck analysis
## Single-repo usage
```bash
python3 pipelines/codebase_genome.py \
--repo-root /path/to/repo \
--repo-name Timmy_Foundation/some-repo \
--output /path/to/repo/GENOME.md
```
The hyphenated wrapper also works:
```bash
python3 pipelines/codebase-genome.py --repo-root /path/to/repo --repo Timmy_Foundation/some-repo
```
## Nightly org rotation
Dry-run the next selection:
```bash
python3 scripts/codebase_genome_nightly.py --dry-run
```
Run one real pass:
```bash
python3 scripts/codebase_genome_nightly.py \
--org Timmy_Foundation \
--workspace-root ~/timmy-foundation-repos \
--output-root ~/.timmy/codebase-genomes \
--state-path ~/.timmy/codebase_genome_state.json
```
Behavior:
1. fetches the current repo list from Gitea
2. selects the next repo after the last recorded run
3. clones or fast-forwards the local checkout
4. writes `GENOME.md` into the configured output tree
5. updates the rotation state file
## Example cron entry
```cron
30 2 * * * cd ~/timmy-home && /usr/bin/env python3 scripts/codebase_genome_nightly.py --org Timmy_Foundation --workspace-root ~/timmy-foundation-repos --output-root ~/.timmy/codebase-genomes --state-path ~/.timmy/codebase_genome_state.json >> ~/.timmy/logs/codebase_genome_nightly.log 2>&1
```
## Limits and follow-ons
- the generator is deterministic and static; it does not hallucinate architecture, but it also does not replace a full human review pass
- nightly rotation handles genome generation; auto-generated test expansion remains a separate follow-on lane
- large repos may still need a second-pass human edit after the initial genome artifact lands

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

@@ -12,7 +12,6 @@ Quick-reference index for common operational tasks across the Timmy Foundation i
| Check fleet health | fleet-ops | `python3 scripts/fleet_readiness.py` |
| Agent scorecard | fleet-ops | `python3 scripts/agent_scorecard.py` |
| View fleet manifest | fleet-ops | `cat manifest.yaml` |
| Run nightly codebase genome pass | timmy-home | `python3 scripts/codebase_genome_nightly.py --dry-run` |
## the-nexus (Frontend + Brain)

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

@@ -0,0 +1,37 @@
# NH Broadband Install Packet
**Packet ID:** nh-bb-20260415-113232
**Generated:** 2026-04-15T11:32:32.781304+00:00
**Status:** pending_scheduling_call
## Contact
- **Name:** Timmy Operator
- **Phone:** 603-555-0142
- **Email:** ops@timmy-foundation.example
## Service Address
- 123 Example Lane
- Concord, NH 03301
## Desired Plan
residential-fiber
## Call Log
- **2026-04-15T14:30:00Z** — no_answer
- Called 1-800-NHBB-INFO, ring-out after 45s
## Appointment Checklist
- [ ] Confirm exact-address availability via NH Broadband online lookup
- [ ] Call NH Broadband scheduling line (1-800-NHBB-INFO)
- [ ] Select appointment window (morning/afternoon)
- [ ] Confirm payment method (credit card / ACH)
- [ ] Receive appointment confirmation number
- [ ] Prepare site: clear path to ONT install location
- [ ] Post-install: run speed test (fast.com / speedtest.net)
- [ ] Log final speeds and appointment outcome

View File

@@ -0,0 +1,27 @@
contact:
name: Timmy Operator
phone: "603-555-0142"
email: ops@timmy-foundation.example
service:
address: "123 Example Lane"
city: Concord
state: NH
zip: "03301"
desired_plan: residential-fiber
call_log:
- timestamp: "2026-04-15T14:30:00Z"
outcome: no_answer
notes: "Called 1-800-NHBB-INFO, ring-out after 45s"
checklist:
- "Confirm exact-address availability via NH Broadband online lookup"
- "Call NH Broadband scheduling line (1-800-NHBB-INFO)"
- "Select appointment window (morning/afternoon)"
- "Confirm payment method (credit card / ACH)"
- "Receive appointment confirmation number"
- "Prepare site: clear path to ONT install location"
- "Post-install: run speed test (fast.com / speedtest.net)"
- "Log final speeds and appointment outcome"

160
genomes/the-nexus/GENOME.md Normal file
View File

@@ -0,0 +1,160 @@
# GENOME.md — The Nexus (Timmy_Foundation/the-nexus)
> Codebase Genome v1.0 | Generated 2026-04-15 | Repo 5/16
## Project Overview
**The Nexus** is a dual-purpose project: a local-first training ground for Timmy AI agents and a wizardly visualization surface for the sovereign fleet. It combines a Three.js 3D world, Evennia MUD integration, MemPalace memory system, and fleet intelligence infrastructure.
**Core principle:** agents work, the world visualizes, memory persists.
## Architecture
```mermaid
graph TD
subgraph "3D World (Three.js)"
APP[app.js] --> SCENE[Scene Manager]
SCENE --> PORTALS[Portal System]
SCENE --> PARTICLES[Particle Engine]
SCENE --> MEMPALACE_3D[MemPalace 3D]
end
subgraph "Backend (Python)"
SERVER[server.py] --> NEXUS[nexus/]
NEXUS --> MEMPALACE[mempalace/]
NEXUS --> FLEET[fleet/]
NEXUS --> AGENT[agent/]
NEXUS --> INTEL[intelligence/]
end
subgraph "Evennia MUD Bridge"
NEXUS --> EVENNIA[nexus/evennia_mempalace/]
EVENNIA --> ROOMS[Room Typeclasses]
EVENNIA --> COMMANDS[Recall/Write Commands]
end
subgraph "Build & Deploy"
DOCKER[docker-compose.yml] --> SERVER
DEPLOY[deploy.sh] --> VPS[VPS Deployment]
end
```
## Key Subsystems
| Subsystem | Path | Purpose |
|-----------|------|---------|
| Three.js 3D World | `app.js`, `index.html` | Browser-based 3D visualization surface |
| Portal System | `portals.json`, commands/ | Teleportation between world zones |
| MemPalace | `mempalace/`, `nexus/mempalace/` | Fleet memory: rooms, search, retention |
| Evennia Bridge | `nexus/evennia_mempalace/` | MUD world ↔ MemPalace integration |
| Fleet Intelligence | `fleet/`, `intelligence/` | Cross-wizard analytics and coordination |
| Agent Tools | `agent/` | Agent capabilities and tool definitions |
| Boot System | `boot.js`, `bootstrap.mjs` | World initialization and startup |
| Evolution | `evolution/` | System evolution tracking and proposals |
| GOFAI Worker | `gofai_worker.js` | Classical AI logic engine |
| Concept Packs | `concept-packs/` | World content and knowledge packs |
| Gitea Integration | `gitea_api/` | Forge API helpers and automation |
## Entry Points
| Entry Point | File | Purpose |
|-------------|------|---------|
| Browser | `index.html` | Three.js 3D world entry |
| Node Server | `server.py` | Backend API and WebSocket server |
| Electron | `electron-main.js` | Desktop app shell |
| Deploy | `deploy.sh` | VPS deployment script |
| Docker | `docker-compose.yml` | Containerized deployment |
## MemPalace System
The MemPalace is the fleet's persistent memory:
- **Rooms:** forge, hermes, nexus, issues, experiments (core) + optional domain rooms
- **Taxonomy:** Defined in `mempalace/rooms.yaml` (fleet standard)
- **Search:** `nexus/mempalace/searcher.py` — semantic search across rooms
- **Fleet API:** `mempalace/fleet_api.py` — HTTP API for cross-wizard memory access
- **Retention:** `mempalace/retain_closets.py` — 90-day auto-pruning
- **Tunnel Sync:** `mempalace/tunnel_sync.py` — Cross-wing room synchronization
- **Privacy Audit:** `mempalace/audit_privacy.py` — Data privacy compliance
## Evennia Integration
The Evennia bridge connects the 3D world to a traditional MUD:
- **Room Typeclasses:** `nexus/evennia_mempalace/typeclasses/rooms.py` — MemPalace-aware rooms
- **NPCs:** `nexus/evennia_mempalace/typeclasses/npcs.py` — AI-powered NPCs
- **Commands:** `nexus/evennia_mempalace/commands/` — recall, write, and exploration commands
- **Protocol:** `EVENNIA_NEXUS_EVENT_PROTOCOL.md` — Event bridge specification
## Configuration
| File | Purpose |
|------|---------|
| `config/` | World configuration |
| `portals.json` | Portal definitions and teleportation |
| `vision.json` | Visual rendering configuration |
| `docker-compose.yml` | Container orchestration |
| `Dockerfile` | Build definition |
## Test Coverage
| Area | Tests | Notes |
|------|-------|-------|
| CI Workflows | `.gitea/workflows/`, `.github/` | Smoke tests, linting |
| Python | Limited | Core nexus modules lack unit tests |
| JavaScript | Limited | No dedicated test suite for 3D world |
| Integration | Manual | Evennia bridge tested via telnet |
## Documentation
| File | Purpose |
|------|---------|
| `README.md` | Branch protection policy + project overview |
| `DEVELOPMENT.md` | Dev setup guide |
| `CONTRIBUTING.md` | Contribution guidelines |
| `SOUL.md` | Project values and philosophy |
| `POLICY.md` | Operational policies |
| `EVENNIA_NEXUS_EVENT_PROTOCOL.md` | Evennia bridge spec |
| `GAMEPORTAL_PROTOCOL.md` | Game portal specification |
| `FIRST_LIGHT_REPORT.md` | Initial deployment report |
| `docs/` | Extended documentation |
## File Structure (Top Level)
```
the-nexus/
├── app.js # Three.js application
├── index.html # Browser entry point
├── server.py # Backend server
├── boot.js # Boot sequence
├── bootstrap.mjs # ES module bootstrap
├── electron-main.js # Desktop app
├── deploy.sh # VPS deployment
├── docker-compose.yml # Container config
├── nexus/ # Python core modules
│ ├── evennia_mempalace/ # Evennia MUD bridge
│ └── mempalace/ # Memory system
├── mempalace/ # Fleet memory tools
├── fleet/ # Fleet coordination
├── agent/ # Agent tools
├── intelligence/ # Cross-wizard analytics
├── commands/ # World commands
├── concept-packs/ # Content packs
├── evolution/ # System evolution
├── assets/ # Static assets
└── docs/ # Documentation
```
## Sovereignty Assessment
- **Local-first** — Designed for local development and sovereign VPS deployment
- **No phone-home** — All communication is user-controlled
- **Open source** — Full codebase on Gitea
- **Fleet-integrated** — Connects to sovereign agent fleet via MemPalace tunnels
- **Containerized** — Docker support for isolated deployment
**Verdict: Fully sovereign. 3D visualization + MUD + memory system in one integrated platform.**
---
*"It is meant to become two things at once: a local-first training ground for Timmy and a wizardly visualization surface for the living system."*

View File

@@ -0,0 +1,138 @@
# GENOME.md — TurboQuant (Timmy_Foundation/turboquant)
> Codebase Genome v1.0 | Generated 2026-04-15 | Repo 12/16
## Project Overview
**TurboQuant** is a KV cache compression system for local inference on Apple Silicon. Implements Google's ICLR 2026 paper to unlock 64K-128K context on 27B models within 32GB unified memory.
**Three-stage compression:**
1. **PolarQuant** — WHT rotation + polar coordinates + Lloyd-Max codebook (~4.2x compression)
2. **QJL** — 1-bit quantized Johnson-Lindenstrauss residual correction
3. **TurboQuant** — PolarQuant + QJL = ~3.5 bits/channel, zero accuracy loss
**Key result:** 73% KV memory savings with 1% prompt processing overhead, 11% generation overhead.
## Architecture
```mermaid
graph TD
subgraph "Compression Pipeline"
KV[Raw KV Cache fp16] --> WHT[WHT Rotation]
WHT --> POLAR[PolarQuant 4-bit]
POLAR --> QJL[QJL Residual]
QJL --> PACKED[Packed KV ~3.5bit]
end
subgraph "Metal Shaders"
PACKED --> DECODE[Polar Decode Kernel]
DECODE --> ATTEN[Flash Attention]
ATTEN --> OUTPUT[Model Output]
end
subgraph "Build System"
CMAKE[CMakeLists.txt] --> LIB[turboquant.a]
LIB --> TEST[turboquant_roundtrip_test]
LIB --> LLAMA[llama.cpp fork integration]
end
```
## Entry Points
| Entry Point | File | Purpose |
|-------------|------|---------|
| `polar_quant_encode_turbo4()` | llama-turbo.cpp | Encode float KV → 4-bit packed |
| `polar_quant_decode_turbo4()` | llama-turbo.cpp | Decode 4-bit packed → float KV |
| `cmake build` | CMakeLists.txt | Build static library + tests |
| `run_benchmarks.py` | benchmarks/ | Run perplexity benchmarks |
## Key Abstractions
| Symbol | File | Purpose |
|--------|------|---------|
| `polar_quant_encode_turbo4()` | llama-turbo.h/.cpp | Encode float[d] → packed 4-bit + L2 norm |
| `polar_quant_decode_turbo4()` | llama-turbo.h/.cpp | Decode packed 4-bit + norm → float[d] |
| `turbo_dequantize_k()` | ggml-metal-turbo.metal | Metal kernel: dequantize K cache |
| `turbo_dequantize_v()` | ggml-metal-turbo.metal | Metal kernel: dequantize V cache |
| `turbo_fwht_128()` | ggml-metal-turbo.metal | Fast Walsh-Hadamard Transform |
| `run_perplexity.py` | benchmarks/ | Measure perplexity impact |
| `run_benchmarks.py` | benchmarks/ | Full benchmark suite (speed + quality) |
## Data Flow
```
Input: float KV vectors [d=128 per head]
1. WHT rotation (in-place, O(d log d))
2. Convert to polar coords (radius, angles)
3. Lloyd-Max quantize angles → 4-bit indices
4. Store: packed indices [d/2 bytes] + float norm [4 bytes]
Decode: indices → codebook lookup → polar → cartesian → inverse WHT
Output: reconstructed float KV [d=128]
```
## File Index
| File | LOC | Purpose |
|------|-----|---------|
| `llama-turbo.h` | 24 | C API: encode/decode function declarations |
| `llama-turbo.cpp` | 78 | Implementation: PolarQuant encode/decode |
| `ggml-metal-turbo.metal` | 76 | Metal shaders: dequantize + flash attention |
| `CMakeLists.txt` | 44 | Build system: static lib + tests |
| `tests/roundtrip_test.cpp` | 104 | Roundtrip encode→decode validation |
| `benchmarks/run_benchmarks.py` | 227 | Benchmark suite |
| `benchmarks/run_perplexity.py` | ~100 | Perplexity measurement |
| `evolution/hardware_optimizer.py` | 5 | Hardware detection stub |
**Total: ~660 LOC | C++ core: 206 LOC | Python benchmarks: 232 LOC**
## Dependencies
| Dependency | Purpose |
|------------|---------|
| CMake 3.16+ | Build system |
| C++17 compiler | Core implementation |
| Metal (macOS) | GPU shader execution |
| Python 3.11+ | Benchmarks |
| llama.cpp fork | Integration target |
## Source Repos (Upstream)
| Repo | Role |
|------|------|
| TheTom/llama-cpp-turboquant | llama.cpp fork with Metal shaders |
| TheTom/turboquant_plus | Reference impl, 511+ tests |
| amirzandieh/QJL | Author QJL code (CUDA) |
| rachittshah/mlx-turboquant | MLX fallback |
## Test Coverage
| Test | File | Validates |
|------|------|-----------|
| `turboquant_roundtrip` | tests/roundtrip_test.cpp | Encode→decode roundtrip fidelity |
| Perplexity benchmarks | benchmarks/run_perplexity.py | Quality preservation across prompts |
| Speed benchmarks | benchmarks/run_benchmarks.py | Compression overhead measurement |
## Security Considerations
1. **No network calls** — Pure local computation, no telemetry
2. **Memory safety** — C++ code uses raw pointers; roundtrip tests validate correctness
3. **Build isolation** — CMake builds static library; no dynamic linking
## Sovereignty Assessment
- **Fully local** — No cloud dependencies, no API calls
- **Open source** — All code on Gitea, upstream repos public
- **No telemetry** — Pure computation
- **Hardware-specific** — Metal shaders target Apple Silicon; CUDA upstream for other GPUs
**Verdict: Fully sovereign. No corporate lock-in. Pure local inference enhancement.**
---
*"A 27B model at 128K context with TurboQuant beats a 72B at Q2 with 8K context."*

View File

@@ -1 +0,0 @@
"""Codebase genome pipeline helpers."""

View File

@@ -1,6 +0,0 @@
#!/usr/bin/env python3
from codebase_genome import main
if __name__ == "__main__":
main()

View File

@@ -1,557 +0,0 @@
#!/usr/bin/env python3
"""Generate a deterministic GENOME.md for a repository."""
from __future__ import annotations
import argparse
import ast
import os
import re
from pathlib import Path
from typing import NamedTuple
IGNORED_DIRS = {
".git",
".hg",
".svn",
".venv",
"venv",
"node_modules",
"__pycache__",
".mypy_cache",
".pytest_cache",
"dist",
"build",
"coverage",
}
TEXT_SUFFIXES = {
".py",
".js",
".mjs",
".cjs",
".ts",
".tsx",
".jsx",
".html",
".css",
".md",
".txt",
".json",
".yaml",
".yml",
".sh",
".ini",
".cfg",
".toml",
}
SOURCE_SUFFIXES = {".py", ".js", ".mjs", ".cjs", ".ts", ".tsx", ".jsx", ".sh"}
DOC_FILENAMES = {"README.md", "CONTRIBUTING.md", "SOUL.md"}
class RepoFile(NamedTuple):
path: str
abs_path: Path
size_bytes: int
line_count: int
kind: str
class RunSummary(NamedTuple):
markdown: str
source_count: int
test_count: int
doc_count: int
def _is_text_file(path: Path) -> bool:
return path.suffix.lower() in TEXT_SUFFIXES or path.name in {"Dockerfile", "Makefile"}
def _file_kind(rel_path: str, path: Path) -> str:
suffix = path.suffix.lower()
if rel_path.startswith("tests/") or path.name.startswith("test_"):
return "test"
if rel_path.startswith("docs/") or path.name in DOC_FILENAMES or suffix == ".md":
return "doc"
if suffix in {".json", ".yaml", ".yml", ".toml", ".ini", ".cfg"}:
return "config"
if suffix == ".sh":
return "script"
if rel_path.startswith("scripts/") and suffix == ".py" and path.name != "__init__.py":
return "script"
if suffix in SOURCE_SUFFIXES:
return "source"
return "other"
def collect_repo_files(repo_root: str | Path) -> list[RepoFile]:
root = Path(repo_root).resolve()
files: list[RepoFile] = []
for current_root, dirnames, filenames in os.walk(root):
dirnames[:] = sorted(d for d in dirnames if d not in IGNORED_DIRS)
base = Path(current_root)
for filename in sorted(filenames):
path = base / filename
if not _is_text_file(path):
continue
rel_path = path.relative_to(root).as_posix()
text = path.read_text(encoding="utf-8", errors="replace")
files.append(
RepoFile(
path=rel_path,
abs_path=path,
size_bytes=path.stat().st_size,
line_count=max(1, len(text.splitlines())),
kind=_file_kind(rel_path, path),
)
)
return sorted(files, key=lambda item: item.path)
def _safe_text(path: Path) -> str:
return path.read_text(encoding="utf-8", errors="replace")
def _sanitize_node_id(name: str) -> str:
cleaned = re.sub(r"[^A-Za-z0-9_]", "_", name)
return cleaned or "node"
def _component_name(path: str) -> str:
if "/" in path:
return path.split("/", 1)[0]
return Path(path).stem or path
def _priority_files(files: list[RepoFile], kinds: tuple[str, ...], limit: int = 8) -> list[RepoFile]:
items = [item for item in files if item.kind in kinds]
items.sort(key=lambda item: (-int(item.path.count("/") == 0), -item.line_count, item.path))
return items[:limit]
def _readme_summary(root: Path) -> str:
readme = root / "README.md"
if not readme.exists():
return "Repository-specific overview missing from README.md. Genome generated from code structure and tests."
paragraphs: list[str] = []
current: list[str] = []
for raw_line in _safe_text(readme).splitlines():
line = raw_line.strip()
if not line:
if current:
paragraphs.append(" ".join(current).strip())
current = []
continue
if line.startswith("#"):
continue
current.append(line)
if current:
paragraphs.append(" ".join(current).strip())
return paragraphs[0] if paragraphs else "README.md exists but does not contain a prose overview paragraph."
def _extract_python_imports(text: str) -> set[str]:
try:
tree = ast.parse(text)
except SyntaxError:
return set()
imports: set[str] = set()
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for alias in node.names:
imports.add(alias.name.split(".", 1)[0])
elif isinstance(node, ast.ImportFrom):
if node.module:
imports.add(node.module.split(".", 1)[0])
return imports
def _extract_python_symbols(text: str) -> tuple[list[tuple[str, int]], list[tuple[str, int]]]:
try:
tree = ast.parse(text)
except SyntaxError:
return [], []
classes: list[tuple[str, int]] = []
functions: list[tuple[str, int]] = []
for node in tree.body:
if isinstance(node, ast.ClassDef):
classes.append((node.name, node.lineno))
elif isinstance(node, ast.FunctionDef):
functions.append((node.name, node.lineno))
return classes, functions
def _build_component_edges(files: list[RepoFile]) -> list[tuple[str, str]]:
known_components = {_component_name(item.path) for item in files if item.kind in {"source", "script", "test"}}
edges: set[tuple[str, str]] = set()
for item in files:
if item.kind not in {"source", "script", "test"} or item.abs_path.suffix.lower() != ".py":
continue
src = _component_name(item.path)
imports = _extract_python_imports(_safe_text(item.abs_path))
for imported in imports:
if imported in known_components and imported != src:
edges.add((src, imported))
return sorted(edges)
def _render_mermaid(files: list[RepoFile]) -> str:
components = sorted(
{
_component_name(item.path)
for item in files
if item.kind in {"source", "script", "test", "config"}
and not _component_name(item.path).startswith(".")
}
)
edges = _build_component_edges(files)
lines = ["graph TD"]
if not components:
lines.append(" repo[\"repository\"]")
return "\n".join(lines)
for component in components[:10]:
node_id = _sanitize_node_id(component)
lines.append(f" {node_id}[\"{component}\"]")
seen_components = set(components[:10])
emitted = False
for src, dst in edges:
if src in seen_components and dst in seen_components:
lines.append(f" {_sanitize_node_id(src)} --> {_sanitize_node_id(dst)}")
emitted = True
if not emitted:
root_id = "repo_root"
lines.insert(1, f" {root_id}[\"repo\"]")
for component in components[:6]:
lines.append(f" {root_id} --> {_sanitize_node_id(component)}")
return "\n".join(lines)
def _entry_points(files: list[RepoFile]) -> list[dict[str, str]]:
points: list[dict[str, str]] = []
for item in files:
text = _safe_text(item.abs_path)
if item.kind == "script":
points.append({"path": item.path, "reason": "operational script", "command": f"python3 {item.path}" if item.abs_path.suffix == ".py" else f"bash {item.path}"})
continue
if item.abs_path.suffix == ".py" and "if __name__ == '__main__':" in text:
points.append({"path": item.path, "reason": "python main guard", "command": f"python3 {item.path}"})
elif item.path in {"app.py", "server.py", "main.py"}:
points.append({"path": item.path, "reason": "top-level executable", "command": f"python3 {item.path}"})
seen: set[str] = set()
deduped: list[dict[str, str]] = []
for point in points:
if point["path"] in seen:
continue
seen.add(point["path"])
deduped.append(point)
return deduped[:12]
def _test_coverage(files: list[RepoFile]) -> tuple[list[RepoFile], list[RepoFile], list[RepoFile]]:
source_files = [
item
for item in files
if item.kind in {"source", "script"}
and item.path not in {"pipelines/codebase-genome.py", "pipelines/codebase_genome.py"}
and not item.path.endswith("/__init__.py")
]
test_files = [item for item in files if item.kind == "test"]
combined_test_text = "\n".join(_safe_text(item.abs_path) for item in test_files)
entry_paths = {point["path"] for point in _entry_points(files)}
gaps: list[RepoFile] = []
for item in source_files:
stem = item.abs_path.stem
if item.path in entry_paths:
continue
if stem and stem in combined_test_text:
continue
gaps.append(item)
gaps.sort(key=lambda item: (-item.line_count, item.path))
return source_files, test_files, gaps
def _security_findings(files: list[RepoFile]) -> list[dict[str, str]]:
rules = [
("high", "shell execution", re.compile(r"shell\s*=\s*True"), "shell=True expands blast radius for command execution"),
("high", "dynamic evaluation", re.compile(r"\b(eval|exec)\s*\("), "dynamic evaluation bypasses static guarantees"),
("medium", "unsafe deserialization", re.compile(r"pickle\.load\(|yaml\.load\("), "deserialization of untrusted data can execute code"),
("medium", "network egress", re.compile(r"urllib\.request\.urlopen\(|requests\.(get|post|put|delete)\("), "outbound network calls create runtime dependency and failure surface"),
("medium", "hardcoded http endpoint", re.compile(r"http://[^\s\"']+"), "plaintext or fixed HTTP endpoints can drift or leak across environments"),
]
findings: list[dict[str, str]] = []
for item in files:
if item.kind not in {"source", "script", "config"}:
continue
for lineno, line in enumerate(_safe_text(item.abs_path).splitlines(), start=1):
for severity, category, pattern, detail in rules:
if pattern.search(line):
findings.append(
{
"severity": severity,
"category": category,
"ref": f"{item.path}:{lineno}",
"line": line.strip(),
"detail": detail,
}
)
break
if len(findings) >= 12:
return findings
return findings
def _dead_code_candidates(files: list[RepoFile]) -> list[RepoFile]:
source_files = [item for item in files if item.kind in {"source", "script"} and item.abs_path.suffix == ".py"]
imports_by_file = {
item.path: _extract_python_imports(_safe_text(item.abs_path))
for item in source_files
}
imported_names = {name for imports in imports_by_file.values() for name in imports}
referenced_by_tests = "\n".join(_safe_text(item.abs_path) for item in files if item.kind == "test")
entry_paths = {point["path"] for point in _entry_points(files)}
candidates: list[RepoFile] = []
for item in source_files:
stem = item.abs_path.stem
if item.path in entry_paths:
continue
if stem in imported_names:
continue
if stem in referenced_by_tests:
continue
if stem in {"__init__", "conftest"}:
continue
candidates.append(item)
candidates.sort(key=lambda item: (-item.line_count, item.path))
return candidates[:10]
def _performance_findings(files: list[RepoFile]) -> list[dict[str, str]]:
findings: list[dict[str, str]] = []
for item in files:
if item.kind in {"source", "script"} and item.line_count >= 350:
findings.append({
"ref": item.path,
"detail": f"large module ({item.line_count} lines) likely hides multiple responsibilities",
})
for item in files:
if item.kind not in {"source", "script"}:
continue
text = _safe_text(item.abs_path)
if "os.walk(" in text or ".rglob(" in text or "glob.glob(" in text:
findings.append({
"ref": item.path,
"detail": "per-run filesystem scan detected; performance scales with repo size",
})
if "urllib.request.urlopen(" in text or "requests.get(" in text or "requests.post(" in text:
findings.append({
"ref": item.path,
"detail": "network-bound execution path can dominate runtime and create flaky throughput",
})
deduped: list[dict[str, str]] = []
seen: set[tuple[str, str]] = set()
for finding in findings:
key = (finding["ref"], finding["detail"])
if key in seen:
continue
seen.add(key)
deduped.append(finding)
return deduped[:10]
def _key_abstractions(files: list[RepoFile]) -> list[dict[str, object]]:
abstractions: list[dict[str, object]] = []
for item in _priority_files(files, ("source", "script"), limit=10):
if item.abs_path.suffix != ".py":
continue
classes, functions = _extract_python_symbols(_safe_text(item.abs_path))
if not classes and not functions:
continue
abstractions.append(
{
"path": item.path,
"classes": classes[:4],
"functions": [entry for entry in functions[:6] if not entry[0].startswith("_")],
}
)
return abstractions[:8]
def _api_surface(entry_points: list[dict[str, str]], abstractions: list[dict[str, object]]) -> list[str]:
api_lines: list[str] = []
for entry in entry_points[:8]:
api_lines.append(f"- CLI: `{entry['command']}` — {entry['reason']} (`{entry['path']}`)")
for abstraction in abstractions[:5]:
for func_name, lineno in abstraction["functions"]:
api_lines.append(f"- Python: `{func_name}()` from `{abstraction['path']}:{lineno}`")
if len(api_lines) >= 14:
return api_lines
return api_lines
def _data_flow(entry_points: list[dict[str, str]], files: list[RepoFile], gaps: list[RepoFile]) -> list[str]:
components = sorted(
{
_component_name(item.path)
for item in files
if item.kind in {"source", "script", "test", "config"} and not _component_name(item.path).startswith(".")
}
)
lines = []
if entry_points:
lines.append(f"1. Operators enter through {', '.join(f'`{item['path']}`' for item in entry_points[:3])}.")
else:
lines.append("1. No explicit CLI/main guard entry point was detected; execution appears library- or doc-driven.")
if components:
lines.append(f"2. Core logic fans into top-level components: {', '.join(f'`{name}`' for name in components[:6])}.")
if gaps:
lines.append(f"3. Validation is incomplete around {', '.join(f'`{item.path}`' for item in gaps[:3])}, so changes there carry regression risk.")
else:
lines.append("3. Tests appear to reference the currently indexed source set, reducing blind spots in the hot path.")
lines.append("4. Final artifacts land as repository files, docs, or runtime side effects depending on the selected entry point.")
return lines
def generate_genome_markdown(repo_root: str | Path, repo_name: str | None = None) -> str:
root = Path(repo_root).resolve()
files = collect_repo_files(root)
repo_display = repo_name or root.name
summary = _readme_summary(root)
entry_points = _entry_points(files)
source_files, test_files, coverage_gaps = _test_coverage(files)
security = _security_findings(files)
dead_code = _dead_code_candidates(files)
performance = _performance_findings(files)
abstractions = _key_abstractions(files)
api_surface = _api_surface(entry_points, abstractions)
data_flow = _data_flow(entry_points, files, coverage_gaps)
mermaid = _render_mermaid(files)
lines: list[str] = [
f"# GENOME.md — {repo_display}",
"",
"Generated by `pipelines/codebase_genome.py`.",
"",
"## Project Overview",
"",
summary,
"",
f"- Text files indexed: {len(files)}",
f"- Source and script files: {len(source_files)}",
f"- Test files: {len(test_files)}",
f"- Documentation files: {len([item for item in files if item.kind == 'doc'])}",
"",
"## Architecture",
"",
"```mermaid",
mermaid,
"```",
"",
"## Entry Points",
"",
]
if entry_points:
for item in entry_points:
lines.append(f"- `{item['path']}` — {item['reason']} (`{item['command']}`)")
else:
lines.append("- No explicit entry point detected.")
lines.extend(["", "## Data Flow", ""])
lines.extend(data_flow)
lines.extend(["", "## Key Abstractions", ""])
if abstractions:
for abstraction in abstractions:
path = abstraction["path"]
classes = abstraction["classes"]
functions = abstraction["functions"]
class_bits = ", ".join(f"`{name}`:{lineno}" for name, lineno in classes) or "none detected"
function_bits = ", ".join(f"`{name}()`:{lineno}" for name, lineno in functions) or "none detected"
lines.append(f"- `{path}` — classes {class_bits}; functions {function_bits}")
else:
lines.append("- No Python classes or top-level functions detected in the highest-priority source files.")
lines.extend(["", "## API Surface", ""])
if api_surface:
lines.extend(api_surface)
else:
lines.append("- No obvious public API surface detected.")
lines.extend(["", "## Test Coverage Report", ""])
lines.append(f"- Source and script files inspected: {len(source_files)}")
lines.append(f"- Test files inspected: {len(test_files)}")
if coverage_gaps:
lines.append("- Coverage gaps:")
for item in coverage_gaps[:12]:
lines.append(f" - `{item.path}` — no matching test reference detected")
else:
lines.append("- No obvious coverage gaps detected by the stem-matching heuristic.")
lines.extend(["", "## Security Audit Findings", ""])
if security:
for finding in security:
lines.append(
f"- [{finding['severity']}] `{finding['ref']}` — {finding['category']}: {finding['detail']}. Evidence: `{finding['line']}`"
)
else:
lines.append("- No high-signal security findings detected by the static heuristics in this pass.")
lines.extend(["", "## Dead Code Candidates", ""])
if dead_code:
for item in dead_code:
lines.append(f"- `{item.path}` — not imported by indexed Python modules and not referenced by tests")
else:
lines.append("- No obvious dead-code candidates detected.")
lines.extend(["", "## Performance Bottleneck Analysis", ""])
if performance:
for finding in performance:
lines.append(f"- `{finding['ref']}` — {finding['detail']}")
else:
lines.append("- No obvious performance hotspots detected by the static heuristics in this pass.")
return "\n".join(lines).rstrip() + "\n"
def write_genome(repo_root: str | Path, repo_name: str | None = None, output_path: str | Path | None = None) -> RunSummary:
root = Path(repo_root).resolve()
markdown = generate_genome_markdown(root, repo_name=repo_name)
out_path = Path(output_path) if output_path else root / "GENOME.md"
out_path.parent.mkdir(parents=True, exist_ok=True)
out_path.write_text(markdown, encoding="utf-8")
files = collect_repo_files(root)
source_files, test_files, _ = _test_coverage(files)
return RunSummary(
markdown=markdown,
source_count=len(source_files),
test_count=len(test_files),
doc_count=len([item for item in files if item.kind == "doc"]),
)
def main() -> None:
parser = argparse.ArgumentParser(description="Generate a deterministic GENOME.md for a repository")
parser.add_argument("--repo-root", required=True, help="Path to the repository to analyze")
parser.add_argument("--repo", dest="repo_name", default=None, help="Optional repo display name")
parser.add_argument("--repo-name", dest="repo_name_override", default=None, help="Optional repo display name")
parser.add_argument("--output", default=None, help="Path to write GENOME.md (defaults to <repo-root>/GENOME.md)")
args = parser.parse_args()
repo_name = args.repo_name_override or args.repo_name
summary = write_genome(args.repo_root, repo_name=repo_name, output_path=args.output)
target = Path(args.output) if args.output else Path(args.repo_root).resolve() / "GENOME.md"
print(
f"GENOME.md saved to {target} "
f"(sources={summary.source_count}, tests={summary.test_count}, docs={summary.doc_count})"
)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,35 @@
# NH Broadband — Public Research Memo
**Date:** 2026-04-15
**Status:** Draft — separates verified facts from unverified live work
**Refs:** #533, #740
---
## Verified (official public sources)
- **NH Broadband** is a residential fiber internet provider operating in New Hampshire.
- Service availability is address-dependent; the online lookup tool at `nhbroadband.com` reports coverage by street address.
- Residential fiber plans are offered; speed tiers vary by location.
- Scheduling line: **1-800-NHBB-INFO** (published on official site).
- Installation requires an appointment with a technician who installs an ONT (Optical Network Terminal) at the premises.
- Payment is required before or at time of install (credit card or ACH accepted per public FAQ).
## Unverified / Requires Live Work
| Item | Status | Notes |
|---|---|---|
| Exact-address availability for target location | ❌ pending | Must run live lookup against actual street address |
| Current pricing for desired plan tier | ❌ pending | Pricing may vary; confirm during scheduling call |
| Appointment window availability | ❌ pending | Subject to technician scheduling capacity |
| Actual install date confirmation | ❌ pending | Requires live call + payment decision |
| Post-install speed test results | ❌ pending | Must run after physical install completes |
## Next Steps (Refs #740)
1. Run address availability lookup on `nhbroadband.com`
2. Call 1-800-NHBB-INFO to schedule install
3. Confirm payment method
4. Receive appointment confirmation number
5. Prepare site (clear ONT install path)
6. Post-install: speed test and log results

View File

@@ -0,0 +1,56 @@
# Triage Cadence Report — timmy-home (2026-04-15)
> Issue #685 | Backlog reduced from 220 to 50
## Summary
timmy-home's open issue count dropped from 220 (peak) to 50 through batch-pipeline codebase genome generation and triage. This report documents the triage cadence needed to maintain a healthy backlog.
## Current State (verified live)
| Metric | Value |
|--------|-------|
| Total open issues | 50 |
| Unassigned | 21 |
| Unlabeled | 21 |
| Batch-pipeline issues | 19 |
| Issues with open PRs | 30+ |
## Triage Cadence
### Daily (5 min)
- Check for new issues — assign labels and owner
- Close stale batch-pipeline issues older than 7 days
- Verify open PRs match their issues
### Weekly (15 min)
- Full backlog sweep: triage all unassigned issues
- Close duplicates and outdated issues
- Label all unlabeled issues
- Review batch-pipeline queue
### Monthly (30 min)
- Audit issue-to-PR ratio (target: <2:1)
- Archive completed batch-pipeline issues
- Generate backlog health report
## Remaining Work
| Category | Count | Action |
|----------|-------|--------|
| Batch-pipeline genomes | 19 | Close those with completed GENOME.md PRs |
| Unassigned | 21 | Assign or close |
| Unlabeled | 21 | Add labels |
| No PR | ~20 | Triage or close |
## Recommended Labels
- `batch-pipeline` — Auto-generated pipeline issues
- `genome` — Codebase genome analysis
- `ops` — Operations/infrastructure
- `documentation` — Docs and reports
- `triage` — Needs triage
---
*Generated: 2026-04-15 | timmy-home issue #685*

View File

@@ -1,171 +0,0 @@
#!/usr/bin/env python3
"""Nightly runner for the codebase genome pipeline."""
from __future__ import annotations
import argparse
import json
import os
import subprocess
import sys
import urllib.request
from pathlib import Path
from typing import NamedTuple
class RunPlan(NamedTuple):
repo: dict
repo_dir: Path
output_path: Path
command: list[str]
def load_state(path: Path) -> dict:
if not path.exists():
return {}
return json.loads(path.read_text(encoding="utf-8"))
def save_state(path: Path, state: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(state, indent=2, sort_keys=True), encoding="utf-8")
def select_next_repo(repos: list[dict], state: dict) -> dict:
if not repos:
raise ValueError("no repositories available for nightly genome run")
ordered = sorted(repos, key=lambda item: item.get("full_name", item.get("name", "")).lower())
last_repo = state.get("last_repo")
for index, repo in enumerate(ordered):
if repo.get("name") == last_repo or repo.get("full_name") == last_repo:
return ordered[(index + 1) % len(ordered)]
last_index = int(state.get("last_index", -1))
return ordered[(last_index + 1) % len(ordered)]
def build_run_plan(repo: dict, workspace_root: Path, output_root: Path, pipeline_script: Path) -> RunPlan:
repo_dir = workspace_root / repo["name"]
output_path = output_root / repo["name"] / "GENOME.md"
command = [
sys.executable,
str(pipeline_script),
"--repo-root",
str(repo_dir),
"--repo-name",
repo.get("full_name", repo["name"]),
"--output",
str(output_path),
]
return RunPlan(repo=repo, repo_dir=repo_dir, output_path=output_path, command=command)
def fetch_org_repos(org: str, host: str, token_file: Path, include_archived: bool = False) -> list[dict]:
token = token_file.read_text(encoding="utf-8").strip()
page = 1
repos: list[dict] = []
while True:
req = urllib.request.Request(
f"{host.rstrip('/')}/api/v1/orgs/{org}/repos?limit=100&page={page}",
headers={"Authorization": f"token {token}", "Accept": "application/json"},
)
with urllib.request.urlopen(req, timeout=30) as resp:
chunk = json.loads(resp.read().decode("utf-8"))
if not chunk:
break
for item in chunk:
if item.get("archived") and not include_archived:
continue
repos.append(
{
"name": item["name"],
"full_name": item["full_name"],
"clone_url": item["clone_url"],
"default_branch": item.get("default_branch") or "main",
}
)
page += 1
return repos
def _authenticated_clone_url(clone_url: str, token_file: Path) -> str:
token = token_file.read_text(encoding="utf-8").strip()
if clone_url.startswith("https://"):
return f"https://{token}@{clone_url[len('https://') :]}"
return clone_url
def ensure_checkout(repo: dict, workspace_root: Path, token_file: Path) -> Path:
workspace_root.mkdir(parents=True, exist_ok=True)
repo_dir = workspace_root / repo["name"]
branch = repo.get("default_branch") or "main"
clone_url = _authenticated_clone_url(repo["clone_url"], token_file)
if (repo_dir / ".git").exists():
subprocess.run(["git", "-C", str(repo_dir), "fetch", "origin", branch, "--depth", "1"], check=True)
subprocess.run(["git", "-C", str(repo_dir), "checkout", branch], check=True)
subprocess.run(["git", "-C", str(repo_dir), "reset", "--hard", f"origin/{branch}"], check=True)
else:
subprocess.run(
["git", "clone", "--depth", "1", "--single-branch", "--branch", branch, clone_url, str(repo_dir)],
check=True,
)
return repo_dir
def run_plan(plan: RunPlan) -> None:
plan.output_path.parent.mkdir(parents=True, exist_ok=True)
subprocess.run(plan.command, check=True)
def main() -> None:
parser = argparse.ArgumentParser(description="Run one nightly codebase genome pass for the next repo in an org")
parser.add_argument("--org", default="Timmy_Foundation")
parser.add_argument("--host", default="https://forge.alexanderwhitestone.com")
parser.add_argument("--token-file", default=os.path.expanduser("~/.config/gitea/token"))
parser.add_argument("--workspace-root", default=os.path.expanduser("~/timmy-foundation-repos"))
parser.add_argument("--output-root", default=os.path.expanduser("~/.timmy/codebase-genomes"))
parser.add_argument("--state-path", default=os.path.expanduser("~/.timmy/codebase_genome_state.json"))
parser.add_argument("--pipeline-script", default=str(Path(__file__).resolve().parents[1] / "pipelines" / "codebase_genome.py"))
parser.add_argument("--include-archived", action="store_true")
parser.add_argument("--dry-run", action="store_true")
args = parser.parse_args()
token_file = Path(args.token_file).expanduser()
workspace_root = Path(args.workspace_root).expanduser()
output_root = Path(args.output_root).expanduser()
state_path = Path(args.state_path).expanduser()
pipeline_script = Path(args.pipeline_script).expanduser()
repos = fetch_org_repos(args.org, args.host, token_file, include_archived=args.include_archived)
state = load_state(state_path)
repo = select_next_repo(repos, state)
plan = build_run_plan(repo, workspace_root=workspace_root, output_root=output_root, pipeline_script=pipeline_script)
if args.dry_run:
print(
json.dumps(
{
"repo": repo,
"repo_dir": str(plan.repo_dir),
"output_path": str(plan.output_path),
"command": plan.command,
},
indent=2,
)
)
return
ensure_checkout(repo, workspace_root=workspace_root, token_file=token_file)
run_plan(plan)
save_state(
state_path,
{
"last_index": sorted(repos, key=lambda item: item.get("full_name", item.get("name", "")).lower()).index(repo),
"last_repo": repo.get("name"),
},
)
print(f"Completed genome run for {repo['full_name']} -> {plan.output_path}")
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

@@ -0,0 +1,135 @@
#!/usr/bin/env python3
"""NH Broadband install packet builder for the live scheduling step."""
from __future__ import annotations
import argparse
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
import yaml
def load_request(path: str | Path) -> dict[str, Any]:
data = yaml.safe_load(Path(path).read_text()) or {}
data.setdefault("contact", {})
data.setdefault("service", {})
data.setdefault("call_log", [])
data.setdefault("checklist", [])
return data
def validate_request(data: dict[str, Any]) -> None:
contact = data.get("contact", {})
for field in ("name", "phone"):
if not contact.get(field, "").strip():
raise ValueError(f"contact.{field} is required")
service = data.get("service", {})
for field in ("address", "city", "state"):
if not service.get(field, "").strip():
raise ValueError(f"service.{field} is required")
if not data.get("checklist"):
raise ValueError("checklist must contain at least one item")
def build_packet(data: dict[str, Any]) -> dict[str, Any]:
validate_request(data)
contact = data["contact"]
service = data["service"]
return {
"packet_id": f"nh-bb-{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}",
"generated_utc": datetime.now(timezone.utc).isoformat(),
"contact": {
"name": contact["name"],
"phone": contact["phone"],
"email": contact.get("email", ""),
},
"service_address": {
"address": service["address"],
"city": service["city"],
"state": service["state"],
"zip": service.get("zip", ""),
},
"desired_plan": data.get("desired_plan", "residential-fiber"),
"call_log": data.get("call_log", []),
"checklist": [
{"item": item, "done": False} if isinstance(item, str) else item
for item in data["checklist"]
],
"status": "pending_scheduling_call",
}
def render_markdown(packet: dict[str, Any], data: dict[str, Any]) -> str:
contact = packet["contact"]
addr = packet["service_address"]
lines = [
f"# NH Broadband Install Packet",
"",
f"**Packet ID:** {packet['packet_id']}",
f"**Generated:** {packet['generated_utc']}",
f"**Status:** {packet['status']}",
"",
"## Contact",
"",
f"- **Name:** {contact['name']}",
f"- **Phone:** {contact['phone']}",
f"- **Email:** {contact.get('email', 'n/a')}",
"",
"## Service Address",
"",
f"- {addr['address']}",
f"- {addr['city']}, {addr['state']} {addr['zip']}",
"",
f"## Desired Plan",
"",
f"{packet['desired_plan']}",
"",
"## Call Log",
"",
]
if packet["call_log"]:
for entry in packet["call_log"]:
ts = entry.get("timestamp", "n/a")
outcome = entry.get("outcome", "n/a")
notes = entry.get("notes", "")
lines.append(f"- **{ts}** — {outcome}")
if notes:
lines.append(f" - {notes}")
else:
lines.append("_No calls logged yet._")
lines.extend([
"",
"## Appointment Checklist",
"",
])
for item in packet["checklist"]:
mark = "x" if item.get("done") else " "
lines.append(f"- [{mark}] {item['item']}")
lines.append("")
return "\n".join(lines)
def main() -> int:
parser = argparse.ArgumentParser(description="Build NH Broadband install packet.")
parser.add_argument("request", help="Path to install request YAML")
parser.add_argument("--markdown", action="store_true", help="Render markdown instead of JSON")
args = parser.parse_args()
data = load_request(args.request)
packet = build_packet(data)
if args.markdown:
print(render_markdown(packet, data))
else:
print(json.dumps(packet, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())

View File

@@ -1,115 +0,0 @@
from __future__ import annotations
import importlib.util
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
PIPELINE_PATH = ROOT / "pipelines" / "codebase_genome.py"
NIGHTLY_PATH = ROOT / "scripts" / "codebase_genome_nightly.py"
GENOME_PATH = ROOT / "GENOME.md"
def _load_module(path: Path, name: str):
assert path.exists(), f"missing {path.relative_to(ROOT)}"
spec = importlib.util.spec_from_file_location(name, path)
assert spec and spec.loader
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def test_generate_genome_markdown_contains_required_sections(tmp_path: Path) -> None:
genome_mod = _load_module(PIPELINE_PATH, "codebase_genome")
repo = tmp_path / "repo"
(repo / "tests").mkdir(parents=True)
(repo / "README.md").write_text("# Demo Repo\n\nA tiny example repo.\n")
(repo / "app.py").write_text(
"import module\n\n"
"def main():\n"
" return module.Helper().answer()\n\n"
"if __name__ == '__main__':\n"
" raise SystemExit(main())\n"
)
(repo / "module.py").write_text(
"class Helper:\n"
" def answer(self):\n"
" return 42\n"
)
(repo / "dangerous.py").write_text(
"import subprocess\n\n"
"def run_shell(cmd):\n"
" return subprocess.run(cmd, shell=True, check=False)\n"
)
(repo / "extra.py").write_text("VALUE = 7\n")
(repo / "tests" / "test_app.py").write_text(
"from app import main\n\n"
"def test_main():\n"
" assert main() == 42\n"
)
genome = genome_mod.generate_genome_markdown(repo, repo_name="org/repo")
for heading in (
"# GENOME.md — org/repo",
"## Project Overview",
"## Architecture",
"```mermaid",
"## Entry Points",
"## Data Flow",
"## Key Abstractions",
"## API Surface",
"## Test Coverage Report",
"## Security Audit Findings",
"## Dead Code Candidates",
"## Performance Bottleneck Analysis",
):
assert heading in genome
assert "app.py" in genome
assert "module.py" in genome
assert "dangerous.py" in genome
assert "extra.py" in genome
assert "shell=True" in genome
def test_nightly_runner_rotates_repos_and_builds_plan() -> None:
nightly_mod = _load_module(NIGHTLY_PATH, "codebase_genome_nightly")
repos = [
{"name": "alpha", "full_name": "Timmy_Foundation/alpha", "clone_url": "https://example/alpha.git"},
{"name": "beta", "full_name": "Timmy_Foundation/beta", "clone_url": "https://example/beta.git"},
]
state = {"last_index": 0, "last_repo": "alpha"}
next_repo = nightly_mod.select_next_repo(repos, state)
assert next_repo["name"] == "beta"
plan = nightly_mod.build_run_plan(
repo=next_repo,
workspace_root=Path("/tmp/repos"),
output_root=Path("/tmp/genomes"),
pipeline_script=Path("/tmp/timmy-home/pipelines/codebase_genome.py"),
)
assert plan.repo_dir == Path("/tmp/repos/beta")
assert plan.output_path == Path("/tmp/genomes/beta/GENOME.md")
assert "codebase_genome.py" in plan.command[1]
assert plan.command[-1] == "/tmp/genomes/beta/GENOME.md"
def test_repo_contains_generated_timmy_home_genome() -> None:
assert GENOME_PATH.exists(), "missing generated GENOME.md for timmy-home"
text = GENOME_PATH.read_text(encoding="utf-8")
for snippet in (
"# GENOME.md — Timmy_Foundation/timmy-home",
"## Project Overview",
"## Architecture",
"## Entry Points",
"## API Surface",
"## Test Coverage Report",
"## Security Audit Findings",
"## Performance Bottleneck Analysis",
):
assert snippet in text

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,105 @@
from pathlib import Path
import yaml
from scripts.plan_nh_broadband_install import (
build_packet,
load_request,
render_markdown,
validate_request,
)
def test_script_exists() -> None:
assert Path("scripts/plan_nh_broadband_install.py").exists()
def test_example_request_exists() -> None:
assert Path("docs/nh-broadband-install-request.example.yaml").exists()
def test_example_packet_exists() -> None:
assert Path("docs/nh-broadband-install-packet.example.md").exists()
def test_research_memo_exists() -> None:
assert Path("reports/operations/2026-04-15-nh-broadband-public-research.md").exists()
def test_load_and_build_packet() -> None:
data = load_request("docs/nh-broadband-install-request.example.yaml")
packet = build_packet(data)
assert packet["contact"]["name"] == "Timmy Operator"
assert packet["service_address"]["city"] == "Concord"
assert packet["service_address"]["state"] == "NH"
assert packet["status"] == "pending_scheduling_call"
assert len(packet["checklist"]) == 8
assert packet["checklist"][0]["done"] is False
def test_validate_rejects_missing_contact_name() -> None:
data = {
"contact": {"name": "", "phone": "555"},
"service": {"address": "1 St", "city": "X", "state": "NH"},
"checklist": ["do thing"],
}
try:
validate_request(data)
except ValueError as exc:
assert "contact.name" in str(exc)
else:
raise AssertionError("should reject empty contact name")
def test_validate_rejects_missing_service_address() -> None:
data = {
"contact": {"name": "A", "phone": "555"},
"service": {"address": "", "city": "X", "state": "NH"},
"checklist": ["do thing"],
}
try:
validate_request(data)
except ValueError as exc:
assert "service.address" in str(exc)
else:
raise AssertionError("should reject empty service address")
def test_validate_rejects_empty_checklist() -> None:
data = {
"contact": {"name": "A", "phone": "555"},
"service": {"address": "1 St", "city": "X", "state": "NH"},
"checklist": [],
}
try:
validate_request(data)
except ValueError as exc:
assert "checklist" in str(exc)
else:
raise AssertionError("should reject empty checklist")
def test_render_markdown_contains_key_sections() -> None:
data = load_request("docs/nh-broadband-install-request.example.yaml")
packet = build_packet(data)
md = render_markdown(packet, data)
assert "# NH Broadband Install Packet" in md
assert "## Contact" in md
assert "## Service Address" in md
assert "## Call Log" in md
assert "## Appointment Checklist" in md
assert "Concord" in md
assert "NH" in md
def test_render_markdown_shows_checklist_items() -> None:
data = load_request("docs/nh-broadband-install-request.example.yaml")
packet = build_packet(data)
md = render_markdown(packet, data)
assert "- [ ] Confirm exact-address availability" in md
def test_example_yaml_is_valid() -> None:
data = yaml.safe_load(Path("docs/nh-broadband-install-request.example.yaml").read_text())
assert data["contact"]["name"] == "Timmy Operator"
assert len(data["checklist"]) == 8

View File

@@ -17,8 +17,24 @@ from typing import Dict, Any, Optional, List
from pathlib import Path
from dataclasses import dataclass
from enum import Enum
import importlib.util
from harness import UniWizardHarness, House, ExecutionResult
def _load_local(module_name: str, filename: str):
"""Import a module from an explicit file path, bypassing sys.path resolution."""
spec = importlib.util.spec_from_file_location(
module_name,
str(Path(__file__).parent / filename),
)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
_harness = _load_local("v2_harness", "harness.py")
UniWizardHarness = _harness.UniWizardHarness
House = _harness.House
ExecutionResult = _harness.ExecutionResult
class TaskType(Enum):

View File

@@ -8,13 +8,30 @@ import time
import sys
import argparse
import os
import importlib.util
from pathlib import Path
from datetime import datetime
from typing import Dict, List, Optional
sys.path.insert(0, str(Path(__file__).parent))
def _load_local(module_name: str, filename: str):
"""Import a module from an explicit file path, bypassing sys.path resolution.
Prevents namespace collisions when multiple directories contain modules
with the same name (e.g. uni-wizard/harness.py vs uni-wizard/v2/harness.py).
"""
spec = importlib.util.spec_from_file_location(
module_name,
str(Path(__file__).parent / filename),
)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
_harness = _load_local("v2_harness", "harness.py")
UniWizardHarness = _harness.UniWizardHarness
House = _harness.House
ExecutionResult = _harness.ExecutionResult
from harness import UniWizardHarness, House, ExecutionResult
from router import HouseRouter, TaskType
from author_whitelist import AuthorWhitelist