Compare commits

...

18 Commits

Author SHA1 Message Date
76a886334b test: add tests for describe_image, describe_image_sharegpt, generate_training_pairs
Addresses review item #5 (test coverage gaps):
- describe_image: oversize skip, valid JSON parse, truncated JSON repair, fallback
- describe_image_sharegpt: oversize skip, natural language return
- generate_training_pairs: JSONL format, ShareGPT format, dry run
2026-04-15 11:20:35 +00:00
e1abecbc54 fix: address all 7 review items from Timmy's code review
1. Add MAX_FILE_SIZE check (50MB) to prevent unbounded memory usage
2. Add 'lighting' to JSON repair field list
3. Seek 2s into video before extracting frame + log ffmpeg stderr
4. Fix --model default to None so auto-detect actually works
6. Wrap temp frame cleanup in try/finally for crash safety
7. Reduce inter-file sleep to 0.1s (Ollama is local, no rate limit needed)

Test coverage gap (item #5) tracked for follow-up.
2026-04-15 11:17:24 +00:00
b3f5a2f21c docs: add visual evidence for scene description generator #689
Some checks are pending
Architecture Lint / Linter Tests (pull_request) Waiting to run
Architecture Lint / Lint Repository (pull_request) Blocked by required conditions
PR Checklist / pr-checklist (pull_request) Waiting to run
Smoke Test / smoke (pull_request) Waiting to run
Validate Config / YAML Lint (pull_request) Waiting to run
Validate Config / JSON Validate (pull_request) Waiting to run
Validate Config / Python Syntax & Import Check (pull_request) Waiting to run
Validate Config / Python Test Suite (pull_request) Blocked by required conditions
Validate Config / Shell Script Lint (pull_request) Waiting to run
Validate Config / Cron Syntax Check (pull_request) Waiting to run
Validate Config / Deploy Script Dry Run (pull_request) Waiting to run
Validate Config / Playbook Schema Validation (pull_request) Waiting to run
2026-04-15 10:03:16 +00:00
e176fadef5 fix: make --input optional when --check-model is used 2026-04-15 09:54:22 +00:00
7ca2ebe6b5 fix: increase token limit and improve JSON repair for truncated responses 2026-04-15 09:50:31 +00:00
e9d2cb5e56 fix: improve JSON parsing fallback and default to gemma4 2026-04-15 09:42:27 +00:00
990676fb02 feat: enhance scene description generator for #689
- Auto-detect best vision model (gemma4:latest preferred)
- Add --check-model flag for model availability
- Add --format sharegpt for training pipeline integration
- Add retry logic for Ollama API calls
- Add video frame extraction with ffmpeg
- Structured JSON output with validation
- Comprehensive CLI with --help

Closes #689
2026-04-15 09:37:25 +00:00
3ad934febd test: add tests for scene description generator 2026-04-15 09:35:27 +00:00
35a191f7b1 Merge PR #725: feat: Provider health monitor with auto-switch (#509) 2026-04-15 06:10:45 +00:00
e987e1b870 Merge PR #726: feat: Pre-flight provider check for session launch (#508) 2026-04-15 06:10:41 +00:00
19278513b4 Merge PR #727: feat: Three.js-specific glitch detection patterns (#543) 2026-04-15 06:10:38 +00:00
1088bf8983 test: add Three.js pattern tests and update assertions (#543)
Some checks are pending
Architecture Lint / Linter Tests (pull_request) Waiting to run
Architecture Lint / Lint Repository (pull_request) Blocked by required conditions
PR Checklist / pr-checklist (pull_request) Waiting to run
Smoke Test / smoke (pull_request) Waiting to run
Validate Config / YAML Lint (pull_request) Waiting to run
Validate Config / JSON Validate (pull_request) Waiting to run
Validate Config / Python Syntax & Import Check (pull_request) Waiting to run
Validate Config / Python Test Suite (pull_request) Blocked by required conditions
Validate Config / Shell Script Lint (pull_request) Waiting to run
Validate Config / Cron Syntax Check (pull_request) Waiting to run
Validate Config / Deploy Script Dry Run (pull_request) Waiting to run
Validate Config / Playbook Schema Validation (pull_request) Waiting to run
- Added TestThreeJsPatterns class with 14 tests
- Tests cover: pattern existence, severity inference, vision prompt
- Updated pattern count assertion (14+ patterns now)
- Updated demo test (6 glitches: 4 original + 2 Three.js)
2026-04-15 05:37:17 +00:00
94f0a132d4 feat: add get_threejs_patterns() filter function (#543) 2026-04-15 05:34:17 +00:00
279356bed6 feat: add --threejs flag and Three.js-aware severity inference (#543)
- Added --threejs flag for focused Three.js pattern scanning
- Updated _infer_severity with shader_failure, texture_placeholder,
  uv_mapping_error, frustum_culling, shadow_map_artifact categories
- Added Three.js demo detections (shader failure, shadow map)
- Bumped detector version to 0.2.0
2026-04-15 05:34:16 +00:00
511ff863c2 feat: add Three.js-specific glitch detection patterns (#543)
Adds 6 new Three.js-specific glitch categories and patterns:
- SHADER_FAILURE: Solid black materials from shader compilation errors
- TEXTURE_PLACEHOLDER: 1x1 white pixel stretched surfaces
- UV_MAPPING_ERROR: BufferGeometry UV coordinate errors
- FRUSTUM_CULLING: Objects popping at screen edges
- SHADOW_MAP_ARTIFACT: Pixelated/blocky shadow maps
- BLOOM_OVERFLOW: Excessive post-processing bloom bleed

Closes #543
2026-04-15 05:32:25 +00:00
b6e3a647b0 feat: add pre-flight provider check script (#508)
Some checks are pending
Architecture Lint / Linter Tests (pull_request) Waiting to run
Architecture Lint / Lint Repository (pull_request) Blocked by required conditions
PR Checklist / pr-checklist (pull_request) Waiting to run
Smoke Test / smoke (pull_request) Waiting to run
Validate Config / Deploy Script Dry Run (pull_request) Waiting to run
Validate Config / Playbook Schema Validation (pull_request) Waiting to run
Validate Config / YAML Lint (pull_request) Waiting to run
Validate Config / JSON Validate (pull_request) Waiting to run
Validate Config / Python Syntax & Import Check (pull_request) Waiting to run
Validate Config / Python Test Suite (pull_request) Blocked by required conditions
Validate Config / Shell Script Lint (pull_request) Waiting to run
Validate Config / Cron Syntax Check (pull_request) Waiting to run
- Checks OpenRouter balance via /api/v1/auth/key
- Tests Nous and Anthropic API keys
- Verifies Ollama is running
- Pre-flight check before session launch
- Returns exit code for automation

Closes #508
2026-04-15 03:55:04 +00:00
e14158676d feat: add provider health monitor script (#509)
Some checks are pending
Smoke Test / smoke (pull_request) Waiting to run
Architecture Lint / Linter Tests (pull_request) Waiting to run
Architecture Lint / Lint Repository (pull_request) Blocked by required conditions
PR Checklist / pr-checklist (pull_request) Waiting to run
Validate Config / Cron Syntax Check (pull_request) Waiting to run
Validate Config / Deploy Script Dry Run (pull_request) Waiting to run
Validate Config / Playbook Schema Validation (pull_request) Waiting to run
Validate Config / YAML Lint (pull_request) Waiting to run
Validate Config / JSON Validate (pull_request) Waiting to run
Validate Config / Python Syntax & Import Check (pull_request) Waiting to run
Validate Config / Python Test Suite (pull_request) Blocked by required conditions
Validate Config / Shell Script Lint (pull_request) Waiting to run
- Tests all configured providers
- Maintains health map in tmux-state.json
- Auto-switches profiles to working providers
- Supports --daemon and --status modes

Closes #509
2026-04-15 03:48:37 +00:00
26e39d8949 feat: add autonomous cron supervisor job (#513)
- Runs every 7 minutes
- Checks dev and timmy sessions
- Loads tmux-supervisor skill
- Telegram only on actionable events
- Silent when all agents busy
2026-04-15 03:33:43 +00:00
9 changed files with 1815 additions and 9 deletions

View File

@@ -31,6 +31,14 @@ class GlitchCategory(Enum):
WATER_REFLECTION = "water_reflection"
SKYBOX_SEAM = "skybox_seam"
# Three.js-specific categories (ref: timmy-config#543)
SHADER_FAILURE = "shader_failure"
TEXTURE_PLACEHOLDER = "texture_placeholder"
UV_MAPPING_ERROR = "uv_mapping_error"
FRUSTUM_CULLING = "frustum_culling"
SHADOW_MAP_ARTIFACT = "shadow_map_artifact"
BLOOM_OVERFLOW = "bloom_overflow"
@dataclass
class GlitchPattern:
@@ -241,6 +249,123 @@ MATRIX_GLITCH_PATTERNS: list[GlitchPattern] = [
],
confidence_threshold=0.45,
),
# --- Three.js-Specific Glitch Patterns (ref: timmy-config#543) ---
GlitchPattern(
category=GlitchCategory.SHADER_FAILURE,
name="Shader Compilation Failure",
description="Three.js shader failed to compile, rendering the material as solid black. "
"Common when custom ShaderMaterial has syntax errors or missing uniforms.",
severity=GlitchSeverity.CRITICAL,
detection_prompts=[
"Look for objects or surfaces rendered as pure black (#000000) that should have visible textures or materials.",
"Identify geometry that appears completely dark while surrounding objects are normally lit.",
"Check for objects where the material seems to 'absorb all light' — flat black with no shading gradient.",
],
visual_indicators=[
"solid black object with no shading",
"geometry rendered as silhouette",
"material appears to absorb light entirely",
"black patch inconsistent with scene lighting",
],
confidence_threshold=0.7,
),
GlitchPattern(
category=GlitchCategory.TEXTURE_PLACEHOLDER,
name="Three.js Texture Not Loaded",
description="Three.js failed to load the texture asset, rendering a 1x1 white pixel "
"stretched across the entire surface. Distinguished from missing-texture by "
"the uniform white/grey appearance rather than magenta.",
severity=GlitchSeverity.CRITICAL,
detection_prompts=[
"Look for surfaces that are uniformly white or light grey with no texture detail, even on large geometry.",
"Identify objects where the texture appears as a single solid color stretched across complex UVs.",
"Check for surfaces that look 'blank' or 'unloaded' — flat white/grey where detail should exist.",
],
visual_indicators=[
"uniform white or light grey surface",
"no texture detail on large geometry",
"stretched single-color appearance",
"1x1 pixel placeholder stretched to fill UV space",
],
confidence_threshold=0.65,
),
GlitchPattern(
category=GlitchCategory.UV_MAPPING_ERROR,
name="BufferGeometry UV Mapping Error",
description="Three.js BufferGeometry has incorrect UV coordinates, causing textures to "
"appear stretched, compressed, or mapped to the wrong faces.",
severity=GlitchSeverity.HIGH,
detection_prompts=[
"Look for textures that appear dramatically stretched in one direction on specific faces.",
"Identify surfaces where the texture pattern is distorted but other nearby surfaces look correct.",
"Check for faces where the texture seems 'smeared' or mapped with incorrect aspect ratio.",
],
visual_indicators=[
"texture stretching on specific faces",
"distorted pattern on geometry",
"smeared texture appearance",
"aspect ratio mismatch between texture and surface",
],
confidence_threshold=0.6,
),
GlitchPattern(
category=GlitchCategory.FRUSTUM_CULLING,
name="Frustum Culling Artifact",
description="Three.js frustum culling incorrectly marks objects as outside the camera "
"frustum, causing them to pop in/out of existence at screen edges.",
severity=GlitchSeverity.MEDIUM,
detection_prompts=[
"Look for objects that are partially visible at the edge of the frame — half-rendered or cut off unnaturally.",
"Identify geometry that seems to 'pop' into existence as the view angle changes.",
"Check screen edges for objects that appear suddenly rather than smoothly entering the viewport.",
],
visual_indicators=[
"half-visible object at screen edge",
"object popping into frame",
"abrupt appearance of geometry",
"bounding box visible but mesh missing",
],
confidence_threshold=0.55,
),
GlitchPattern(
category=GlitchCategory.SHADOW_MAP_ARTIFACT,
name="Shadow Map Resolution Artifact",
description="Three.js shadow map has insufficient resolution, causing pixelated, "
"blocky shadows with visible texel edges instead of smooth shadow gradients.",
severity=GlitchSeverity.MEDIUM,
detection_prompts=[
"Look for shadows with visible blocky or pixelated edges instead of smooth gradients.",
"Identify shadow maps where individual texels (texture pixels) are clearly visible.",
"Check for shadows that appear as jagged stair-stepped patterns rather than soft edges.",
],
visual_indicators=[
"blocky shadow edges",
"visible texel grid in shadows",
"stair-stepped shadow boundary",
"pixelated shadow gradient",
],
confidence_threshold=0.55,
),
GlitchPattern(
category=GlitchCategory.BLOOM_OVERFLOW,
name="Post-Processing Bloom Overflow",
description="Three.js UnrealBloomPass or similar post-processing bloom effect is too "
"intense, causing bright areas to bleed glow into surrounding geometry.",
severity=GlitchSeverity.LOW,
detection_prompts=[
"Look for bright areas that have an unusually large, soft glow bleeding into adjacent surfaces.",
"Identify scenes where light sources appear to have a 'halo' that extends beyond physical plausibility.",
"Check for bright objects whose glow color bleeds onto nearby unrelated geometry.",
],
visual_indicators=[
"excessive glow bleeding from bright surfaces",
"halo around light sources",
"bloom color tinting adjacent geometry",
"glow bleeding beyond object boundaries",
],
confidence_threshold=0.5,
),
]
@@ -289,6 +414,23 @@ def build_vision_prompt(patterns: list[GlitchPattern] | None = None) -> str:
)
# Three.js-specific category set for filtering (ref: timmy-config#543)
THREEJS_CATEGORIES = {
GlitchCategory.SHADER_FAILURE,
GlitchCategory.TEXTURE_PLACEHOLDER,
GlitchCategory.UV_MAPPING_ERROR,
GlitchCategory.FRUSTUM_CULLING,
GlitchCategory.SHADOW_MAP_ARTIFACT,
GlitchCategory.BLOOM_OVERFLOW,
}
def get_threejs_patterns() -> list[GlitchPattern]:
"""Return only Three.js-specific glitch patterns."""
return [p for p in MATRIX_GLITCH_PATTERNS if p.category in THREEJS_CATEGORIES]
if __name__ == "__main__":
import json
print(f"Loaded {len(MATRIX_GLITCH_PATTERNS)} glitch patterns:\n")

View File

@@ -9,7 +9,7 @@ Usage:
python matrix_glitch_detector.py <url> [--angles 4] [--output report.json]
python matrix_glitch_detector.py --demo # Run with synthetic test data
Ref: timmy-config#491
Ref: timmy-config#491, timmy-config#543
"""
import argparse
@@ -33,6 +33,7 @@ from glitch_patterns import (
MATRIX_GLITCH_PATTERNS,
build_vision_prompt,
get_patterns_by_severity,
get_threejs_patterns,
)
@@ -345,14 +346,17 @@ def _parse_vision_response(
def _infer_severity(category: str, confidence: float) -> str:
"""Infer severity from category and confidence when not provided."""
critical_cats = {"missing_textures", "clipping"}
high_cats = {"floating_assets", "broken_normals"}
critical_cats = {"missing_textures", "clipping", "shader_failure", "texture_placeholder"}
high_cats = {"floating_assets", "broken_normals", "uv_mapping_error"}
medium_cats = {"frustum_culling", "shadow_map_artifact"}
cat_lower = category.lower()
if any(c in cat_lower for c in critical_cats):
return "critical" if confidence > 0.7 else "high"
if any(c in cat_lower for c in high_cats):
return "high" if confidence > 0.7 else "medium"
if any(c in cat_lower for c in medium_cats):
return "medium" if confidence > 0.6 else "low"
return "medium" if confidence > 0.6 else "low"
@@ -389,9 +393,9 @@ def build_report(
),
},
metadata={
"detector_version": "0.1.0",
"detector_version": "0.2.0",
"pattern_count": len(MATRIX_GLITCH_PATTERNS),
"reference": "timmy-config#491",
"reference": "timmy-config#491, timmy-config#543",
},
)
@@ -460,6 +464,30 @@ def run_demo(output_path: Optional[Path] = None) -> ScanResult:
screenshot_index=3,
screenshot_angle="left",
),
DetectedGlitch(
id=str(uuid.uuid4())[:8],
category="shader_failure",
name="Black Material on Portal Frame",
description="Portal frame rendered as solid black — shader compilation failed (missing uniform u_time)",
severity="critical",
confidence=0.91,
location_x=45.0,
location_y=30.0,
screenshot_index=0,
screenshot_angle="front",
),
DetectedGlitch(
id=str(uuid.uuid4())[:8],
category="shadow_map_artifact",
name="Pixelated Character Shadow",
description="Character shadow shows visible texel grid — shadow map resolution too low (512x512)",
severity="medium",
confidence=0.78,
location_x=52.0,
location_y=75.0,
screenshot_index=1,
screenshot_angle="right",
),
]
print(f"[*] Detected {len(demo_glitches)} glitches")
@@ -496,6 +524,11 @@ Examples:
help="Minimum severity to include in report",
)
parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output")
parser.add_argument(
"--threejs",
action="store_true",
help="Focus on Three.js-specific glitch patterns only (shader, texture, UV, culling, shadow, bloom)",
)
args = parser.parse_args()
@@ -525,9 +558,13 @@ Examples:
screenshots = capture_screenshots(args.url, angles, screenshots_dir)
print(f"[*] Captured {len(screenshots)} screenshots")
# Filter patterns by severity
# Filter patterns by severity and type
min_sev = GlitchSeverity(args.min_severity)
patterns = get_patterns_by_severity(min_sev)
if args.threejs:
threejs_patterns = get_threejs_patterns()
patterns = [p for p in patterns if p in threejs_patterns]
print(f"[*] Three.js-focused mode: {len(patterns)} patterns")
# Analyze with vision AI
print(f"[*] Analyzing with vision AI ({len(patterns)} patterns)...")

View File

@@ -0,0 +1,271 @@
#!/usr/bin/env python3
"""
Pre-Flight Provider Check Script
Issue #508: [Robustness] Credential drain detection — provider health checks
Pre-flight check before session launch: verifies provider credentials and balance.
Usage:
python3 preflight-provider-check.py # Check all providers
python3 preflight-provider-check.py --launch # Check and return exit code
python3 preflight-provider-check.py --balance # Check OpenRouter balance
"""
import os, sys, json, yaml, urllib.request
from datetime import datetime, timezone
from pathlib import Path
# Configuration
HERMES_HOME = Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes"))
LOG_DIR = Path.home() / ".local" / "timmy" / "fleet-health"
LOG_FILE = LOG_DIR / "preflight-check.log"
def log(msg):
"""Log message to file and optionally console."""
timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
log_entry = "[" + timestamp + "] " + msg
LOG_DIR.mkdir(parents=True, exist_ok=True)
with open(LOG_FILE, "a") as f:
f.write(log_entry + "\n")
if "--quiet" not in sys.argv:
print(log_entry)
def get_provider_api_key(provider):
"""Get API key for a provider from .env or environment."""
env_file = HERMES_HOME / ".env"
if env_file.exists():
with open(env_file) as f:
for line in f:
line = line.strip()
if line.startswith(provider.upper() + "_API_KEY="):
return line.split("=", 1)[1].strip().strip("'\"")
return os.environ.get(provider.upper() + "_API_KEY")
def check_openrouter_balance(api_key):
"""Check OpenRouter balance via /api/v1/auth/key."""
if not api_key:
return False, "No API key", 0
try:
req = urllib.request.Request(
"https://openrouter.ai/api/v1/auth/key",
headers={"Authorization": "Bearer " + api_key}
)
resp = urllib.request.urlopen(req, timeout=10)
data = json.loads(resp.read())
# Check for credits
credits = data.get("data", {}).get("limit", 0)
usage = data.get("data", {}).get("usage", 0)
remaining = credits - usage if credits else None
if remaining is not None and remaining <= 0:
return False, "No credits remaining", 0
elif remaining is not None:
return True, "Credits available", remaining
else:
return True, "Unlimited or unknown balance", None
except urllib.error.HTTPError as e:
if e.code == 401:
return False, "Invalid API key", 0
else:
return False, "HTTP " + str(e.code), 0
except Exception as e:
return False, str(e)[:100], 0
def check_nous_key(api_key):
"""Check Nous API key with minimal test call."""
if not api_key:
return False, "No API key"
try:
req = urllib.request.Request(
"https://inference.nousresearch.com/v1/models",
headers={"Authorization": "Bearer " + api_key}
)
resp = urllib.request.urlopen(req, timeout=10)
if resp.status == 200:
return True, "Valid key"
else:
return False, "HTTP " + str(resp.status)
except urllib.error.HTTPError as e:
if e.code == 401:
return False, "Invalid API key"
elif e.code == 403:
return False, "Forbidden"
else:
return False, "HTTP " + str(e.code)
except Exception as e:
return False, str(e)[:100]
def check_anthropic_key(api_key):
"""Check Anthropic API key with minimal test call."""
if not api_key:
return False, "No API key"
try:
req = urllib.request.Request(
"https://api.anthropic.com/v1/models",
headers={
"x-api-key": api_key,
"anthropic-version": "2023-06-01"
}
)
resp = urllib.request.urlopen(req, timeout=10)
if resp.status == 200:
return True, "Valid key"
else:
return False, "HTTP " + str(resp.status)
except urllib.error.HTTPError as e:
if e.code == 401:
return False, "Invalid API key"
elif e.code == 403:
return False, "Forbidden"
else:
return False, "HTTP " + str(e.code)
except Exception as e:
return False, str(e)[:100]
def check_ollama():
"""Check if Ollama is running."""
try:
req = urllib.request.Request("http://localhost:11434/api/tags")
resp = urllib.request.urlopen(req, timeout=5)
if resp.status == 200:
data = json.loads(resp.read())
models = data.get("models", [])
return True, str(len(models)) + " models loaded"
else:
return False, "HTTP " + str(resp.status)
except Exception as e:
return False, str(e)[:100]
def get_configured_provider():
"""Get the configured provider from global config."""
config_file = HERMES_HOME / "config.yaml"
if not config_file.exists():
return None
try:
with open(config_file) as f:
config = yaml.safe_load(f)
model_config = config.get("model", {})
if isinstance(model_config, dict):
return model_config.get("provider")
except:
pass
return None
def run_preflight_check():
"""Run pre-flight check on all providers."""
log("=== Pre-Flight Provider Check ===")
results = {}
# Check OpenRouter
or_key = get_provider_api_key("openrouter")
or_ok, or_msg, or_balance = check_openrouter_balance(or_key)
results["openrouter"] = {"healthy": or_ok, "message": or_msg, "balance": or_balance}
# Check Nous
nous_key = get_provider_api_key("nous")
nous_ok, nous_msg = check_nous_key(nous_key)
results["nous"] = {"healthy": nous_ok, "message": nous_msg}
# Check Anthropic
anthropic_key = get_provider_api_key("anthropic")
anthropic_ok, anthropic_msg = check_anthropic_key(anthropic_key)
results["anthropic"] = {"healthy": anthropic_ok, "message": anthropic_msg}
# Check Ollama
ollama_ok, ollama_msg = check_ollama()
results["ollama"] = {"healthy": ollama_ok, "message": ollama_msg}
# Get configured provider
configured = get_configured_provider()
# Summary
healthy_count = sum(1 for r in results.values() if r["healthy"])
total_count = len(results)
log("Results: " + str(healthy_count) + "/" + str(total_count) + " providers healthy")
for provider, result in results.items():
status = "HEALTHY" if result["healthy"] else "UNHEALTHY"
extra = ""
if provider == "openrouter" and result.get("balance") is not None:
extra = " (balance: " + str(result["balance"]) + ")"
log(" " + provider + ": " + status + " - " + result["message"] + extra)
if configured:
log("Configured provider: " + configured)
if configured in results and not results[configured]["healthy"]:
log("WARNING: Configured provider " + configured + " is UNHEALTHY!")
return results, configured
def check_launch_readiness():
"""Check if we're ready to launch sessions."""
results, configured = run_preflight_check()
# Check if configured provider is healthy
if configured and configured in results:
if not results[configured]["healthy"]:
log("LAUNCH BLOCKED: Configured provider " + configured + " is unhealthy")
return False, configured + " is unhealthy"
# Check if at least one provider is healthy
healthy_providers = [p for p, r in results.items() if r["healthy"]]
if not healthy_providers:
log("LAUNCH BLOCKED: No healthy providers available")
return False, "No healthy providers"
log("LAUNCH READY: " + str(len(healthy_providers)) + " healthy providers available")
return True, "Ready"
def show_balance():
"""Show OpenRouter balance."""
api_key = get_provider_api_key("openrouter")
if not api_key:
print("No OpenRouter API key found")
return
ok, msg, balance = check_openrouter_balance(api_key)
if ok:
if balance is not None:
print("OpenRouter balance: " + str(balance) + " credits")
else:
print("OpenRouter: " + msg)
else:
print("OpenRouter: " + msg)
def main():
if "--balance" in sys.argv:
show_balance()
elif "--launch" in sys.argv:
ready, message = check_launch_readiness()
if ready:
print("READY")
sys.exit(0)
else:
print("BLOCKED: " + message)
sys.exit(1)
else:
run_preflight_check()
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,411 @@
#!/usr/bin/env python3
"""
Provider Health Monitor Script
Issue #509: [Robustness] Provider-aware profile config — auto-switch on failure
Monitors provider health and automatically switches profiles to working providers.
Usage:
python3 provider-health-monitor.py # Run once
python3 provider-health-monitor.py --daemon # Run continuously
python3 provider-health-monitor.py --status # Show provider health
"""
import os, sys, json, yaml, urllib.request, time
from datetime import datetime, timezone
from pathlib import Path
# Configuration
HERMES_HOME = Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes"))
PROFILES_DIR = HERMES_HOME / "profiles"
LOG_DIR = Path.home() / ".local" / "timmy" / "fleet-health"
STATE_FILE = LOG_DIR / "tmux-state.json"
LOG_FILE = LOG_DIR / "provider-health.log"
# Provider test endpoints
PROVIDER_TESTS = {
"openrouter": {
"url": "https://openrouter.ai/api/v1/models",
"method": "GET",
"headers": lambda api_key: {"Authorization": "Bearer " + api_key},
"timeout": 10
},
"anthropic": {
"url": "https://api.anthropic.com/v1/models",
"method": "GET",
"headers": lambda api_key: {"x-api-key": api_key, "anthropic-version": "2023-06-01"},
"timeout": 10
},
"nous": {
"url": "https://inference.nousresearch.com/v1/models",
"method": "GET",
"headers": lambda api_key: {"Authorization": "Bearer " + api_key},
"timeout": 10
},
"kimi-coding": {
"url": "https://api.kimi.com/coding/v1/models",
"method": "GET",
"headers": lambda api_key: {"x-api-key": api_key, "x-api-provider": "kimi-coding"},
"timeout": 10
},
"ollama": {
"url": "http://localhost:11434/api/tags",
"method": "GET",
"headers": lambda api_key: {},
"timeout": 5
}
}
def log(msg):
"""Log message to file and optionally console."""
timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
log_entry = "[" + timestamp + "] " + msg
LOG_DIR.mkdir(parents=True, exist_ok=True)
with open(LOG_FILE, "a") as f:
f.write(log_entry + "\n")
if "--quiet" not in sys.argv:
print(log_entry)
def get_provider_api_key(provider):
"""Get API key for a provider from .env or environment."""
env_file = HERMES_HOME / ".env"
if env_file.exists():
with open(env_file) as f:
for line in f:
line = line.strip()
if line.startswith(provider.upper() + "_API_KEY="):
return line.split("=", 1)[1].strip().strip("'\"")
return os.environ.get(provider.upper() + "_API_KEY")
def test_provider(provider, api_key=None):
"""Test if a provider is healthy."""
config = PROVIDER_TESTS.get(provider)
if not config:
return False, "Unknown provider: " + provider
headers = config["headers"](api_key or "")
try:
req = urllib.request.Request(
config["url"],
headers=headers,
method=config["method"]
)
resp = urllib.request.urlopen(req, timeout=config["timeout"])
if resp.status == 200:
return True, "Healthy"
else:
return False, "HTTP " + str(resp.status)
except urllib.error.HTTPError as e:
if e.code == 401:
return False, "Unauthorized (401)"
elif e.code == 403:
return False, "Forbidden (403)"
elif e.code == 429:
return True, "Rate limited but accessible"
else:
return False, "HTTP " + str(e.code)
except Exception as e:
return False, str(e)[:100]
def get_all_providers():
"""Get all providers from profiles and global config."""
providers = set()
# Global config
global_config = HERMES_HOME / "config.yaml"
if global_config.exists():
try:
with open(global_config) as f:
config = yaml.safe_load(f)
# Primary model provider
model_config = config.get("model", {})
if isinstance(model_config, dict):
provider = model_config.get("provider", "")
if provider:
providers.add(provider)
# Auxiliary providers
auxiliary = config.get("auxiliary", {})
for aux_config in auxiliary.values():
if isinstance(aux_config, dict):
provider = aux_config.get("provider", "")
if provider and provider != "auto":
providers.add(provider)
except:
pass
# Profile configs
if PROFILES_DIR.exists():
for profile_dir in PROFILES_DIR.iterdir():
if profile_dir.is_dir():
config_file = profile_dir / "config.yaml"
if config_file.exists():
try:
with open(config_file) as f:
config = yaml.safe_load(f)
model_config = config.get("model", {})
if isinstance(model_config, dict):
provider = model_config.get("provider", "")
if provider:
providers.add(provider)
auxiliary = config.get("auxiliary", {})
for aux_config in auxiliary.values():
if isinstance(aux_config, dict):
provider = aux_config.get("provider", "")
if provider and provider != "auto":
providers.add(provider)
except:
pass
# Add common providers even if not configured
providers.update(["openrouter", "nous", "ollama"])
return list(providers)
def build_health_map():
"""Build a health map of all providers."""
providers = get_all_providers()
health_map = {}
log("Testing " + str(len(providers)) + " providers...")
for provider in providers:
api_key = get_provider_api_key(provider)
healthy, message = test_provider(provider, api_key)
health_map[provider] = {
"healthy": healthy,
"message": message,
"last_test": datetime.now(timezone.utc).isoformat(),
"api_key_present": bool(api_key)
}
status = "HEALTHY" if healthy else "UNHEALTHY"
log(" " + provider + ": " + status + " - " + message)
return health_map
def get_fallback_providers(health_map):
"""Get list of healthy providers in priority order."""
# Priority order: nous, openrouter, ollama, others
priority_order = ["nous", "openrouter", "ollama", "anthropic", "kimi-coding"]
healthy = []
for provider in priority_order:
if provider in health_map and health_map[provider]["healthy"]:
healthy.append(provider)
# Add any other healthy providers not in priority list
for provider, info in health_map.items():
if info["healthy"] and provider not in healthy:
healthy.append(provider)
return healthy
def update_profile_config(profile_name, new_provider):
"""Update a profile's config to use a new provider."""
config_file = PROFILES_DIR / profile_name / "config.yaml"
if not config_file.exists():
return False, "Config file not found"
try:
with open(config_file) as f:
config = yaml.safe_load(f)
# Update model provider
if "model" not in config:
config["model"] = {}
old_provider = config["model"].get("provider", "unknown")
config["model"]["provider"] = new_provider
# Update auxiliary providers if they were using the old provider
auxiliary = config.get("auxiliary", {})
for aux_name, aux_config in auxiliary.items():
if isinstance(aux_config, dict) and aux_config.get("provider") == old_provider:
aux_config["provider"] = new_provider
# Write back
with open(config_file, "w") as f:
yaml.dump(config, f, default_flow_style=False)
log("Updated " + profile_name + ": " + old_provider + " -> " + new_provider)
return True, "Updated"
except Exception as e:
return False, str(e)
def check_profiles(health_map):
"""Check all profiles and update unhealthy providers."""
if not PROFILES_DIR.exists():
return
fallback_providers = get_fallback_providers(health_map)
if not fallback_providers:
log("CRITICAL: No healthy providers available!")
return
updated_profiles = []
for profile_dir in PROFILES_DIR.iterdir():
if not profile_dir.is_dir():
continue
profile_name = profile_dir.name
config_file = profile_dir / "config.yaml"
if not config_file.exists():
continue
try:
with open(config_file) as f:
config = yaml.safe_load(f)
model_config = config.get("model", {})
if not isinstance(model_config, dict):
continue
current_provider = model_config.get("provider", "")
if not current_provider:
continue
# Check if current provider is healthy
if current_provider in health_map and health_map[current_provider]["healthy"]:
continue # Provider is healthy, no action needed
# Find best fallback
best_fallback = None
for provider in fallback_providers:
if provider != current_provider:
best_fallback = provider
break
if not best_fallback:
log("No fallback for " + profile_name + " (current: " + current_provider + ")")
continue
# Update profile
success, message = update_profile_config(profile_name, best_fallback)
if success:
updated_profiles.append({
"profile": profile_name,
"old_provider": current_provider,
"new_provider": best_fallback
})
except Exception as e:
log("Error processing " + profile_name + ": " + str(e))
return updated_profiles
def load_state():
"""Load state from tmux-state.json."""
if STATE_FILE.exists():
try:
with open(STATE_FILE) as f:
return json.load(f)
except:
pass
return {}
def save_state(state):
"""Save state to tmux-state.json."""
LOG_DIR.mkdir(parents=True, exist_ok=True)
with open(STATE_FILE, "w") as f:
json.dump(state, f, indent=2)
def run_once():
"""Run provider health check once."""
log("=== Provider Health Check ===")
state = load_state()
# Build health map
health_map = build_health_map()
# Check profiles and update if needed
updated_profiles = check_profiles(health_map)
# Update state
state["provider_health"] = health_map
state["last_provider_check"] = datetime.now(timezone.utc).isoformat()
if updated_profiles:
state["last_profile_updates"] = updated_profiles
save_state(state)
# Summary
healthy_count = sum(1 for p in health_map.values() if p["healthy"])
total_count = len(health_map)
log("Health: " + str(healthy_count) + "/" + str(total_count) + " providers healthy")
if updated_profiles:
log("Updated " + str(len(updated_profiles)) + " profiles:")
for update in updated_profiles:
log(" " + update["profile"] + ": " + update["old_provider"] + " -> " + update["new_provider"])
def show_status():
"""Show provider health status."""
state = load_state()
health_map = state.get("provider_health", {})
if not health_map:
print("No provider health data available. Run without --status first.")
return
print("Provider Health (last updated: " + str(state.get("last_provider_check", "unknown")) + ")")
print("=" * 80)
for provider, info in sorted(health_map.items()):
status = "HEALTHY" if info["healthy"] else "UNHEALTHY"
message = info.get("message", "")
api_key = "yes" if info.get("api_key_present") else "no"
print(provider.ljust(20) + " " + status.ljust(10) + " API key: " + api_key + " - " + message)
# Show recent updates
updates = state.get("last_profile_updates", [])
if updates:
print()
print("Recent Profile Updates:")
for update in updates:
print(" " + update["profile"] + ": " + update["old_provider"] + " -> " + update["new_provider"])
def daemon_mode():
"""Run continuously."""
log("Starting provider health daemon (check every 300s)")
while True:
try:
run_once()
time.sleep(300) # Check every 5 minutes
except KeyboardInterrupt:
log("Daemon stopped by user")
break
except Exception as e:
log("Error: " + str(e))
time.sleep(60)
def main():
if "--status" in sys.argv:
show_status()
elif "--daemon" in sys.argv:
daemon_mode()
else:
run_once()
if __name__ == "__main__":
main()

View File

@@ -196,7 +196,37 @@
"paused_reason": null,
"skills": [],
"skill": null
},
{
"id": "tmux-supervisor-513",
"name": "Autonomous Cron Supervisor",
"prompt": "Load the tmux-supervisor skill and execute the monitoring protocol.\n\nCheck both `dev` and `timmy` tmux sessions for idle panes. Only send Telegram notifications on actionable events (idle, overflow, failure). Be silent when all agents are working.\n\nSteps:\n1. List all tmux sessions (skip 'Alexander')\n2. For each session, list windows and panes\n3. Capture each pane and classify state (idle vs active)\n4. For idle panes: read context, craft context-aware prompt\n5. Send /queue prompts to idle panes\n6. Verify prompts landed\n7. Only notify via Telegram if:\n - A pane was prompted (idle detected)\n - A pane shows context overflow (>80%)\n - A pane is stuck or crashed\n8. If all panes are active: respond with [SILENT]",
"schedule": {
"kind": "interval",
"minutes": 7,
"display": "every 7m"
},
"schedule_display": "every 7m",
"repeat": {
"times": null,
"completed": 0
},
"enabled": true,
"created_at": "2026-04-15T03:00:00.000000+00:00",
"next_run_at": null,
"last_run_at": null,
"last_status": null,
"last_error": null,
"deliver": "telegram",
"origin": null,
"state": "scheduled",
"paused_at": null,
"paused_reason": null,
"skills": [
"tmux-supervisor"
],
"skill": "tmux-supervisor"
}
],
"updated_at": "2026-04-13T02:00:00+00:00"
}
}

View File

@@ -0,0 +1,74 @@
# Visual Evidence — Gemma 4 Multimodal Scene Description Generator
## Test Image: Coffee Beans (Macro Photo)
### Gemma 4 Vision Analysis (via Ollama)
**Model:** gemma4:latest (8B, Q4_K_M)
**Input:** sample_photo.jpg (46KB JPEG)
**Structured Output (JSONL):**
```json
{
"mood": "dark",
"colors": ["dark brown", "espresso", "black"],
"composition": "close-up",
"camera": "static",
"lighting": "soft",
"description": "An extreme close-up shot captures a dense pile of roasted coffee beans. The beans are a uniform, deep dark brown and appear slightly oily, filling the entire frame. The focus emphasizes the rich texture and individual shapes of the beans."
}
```
### Hermes Vision Analysis (Cross-Validation)
**Scene ID:** COFFEE_MACRO_001
**Mood:** Warm, aromatic, and comforting
**Dominant Colors:** Deep umber, burnt sienna, espresso black, mahogany
**Composition:** Full-frame fill, centrally weighted
**Camera:** High-angle, close-up (Macro)
**Lighting:** Soft, diffused top-lighting
## Test Image: Abstract Geometric Composition
### Gemma 4 Vision Analysis
**Input:** scene1.jpg (10KB, PIL-generated)
**Structured Output (JSONL):**
```json
{
"mood": "energetic",
"colors": ["deep blue", "yellow", "coral"],
"composition": "wide-shot",
"camera": "static",
"lighting": "artificial",
"description": "This is an abstract graphic composition set against a solid, deep blue background. A bright yellow square is placed in the upper left quadrant, while a large, solid coral-colored circle occupies the lower right quadrant. The geometric shapes create a high-contrast, minimalist visual balance."
}
```
## Verification Summary
| Test | Status | Details |
|------|--------|---------|
| Model detection | ✅ PASS | `gemma4:latest` auto-detected |
| Image scanning | ✅ PASS | 2 images found recursively |
| Vision analysis | ✅ PASS | Both images described accurately |
| JSON parsing | ✅ PASS | Structured output with all fields |
| Training format | ✅ PASS | JSONL with source, model, timestamp |
| ShareGPT format | ⚠️ PARTIAL | Works but needs retry on rate limit |
## Running the Generator
```bash
# Check model availability
python scripts/generate_scene_descriptions.py --check-model
# Generate scene descriptions from assets
python scripts/generate_scene_descriptions.py --input ./assets --output training-data/scene-descriptions-auto.jsonl
# Limit to 10 files with specific model
python scripts/generate_scene_descriptions.py --input ./assets --model gemma4:latest --limit 10
# ShareGPT format for training pipeline
python scripts/generate_scene_descriptions.py --input ./assets --format sharegpt
```

View File

@@ -0,0 +1,409 @@
#!/usr/bin/env python3
"""
Auto-generate scene descriptions from image/video assets.
Scans a directory for media files, generates scene descriptions using
a local vision model (Ollama), and outputs training pairs in JSONL format.
Supports Gemma 4 multimodal vision via Ollama. Falls back gracefully when
models are unavailable.
Usage:
python scripts/generate_scene_descriptions.py --input ./assets --output training-data/scene-descriptions-auto.jsonl
python scripts/generate_scene_descriptions.py --input ./assets --model gemma4:latest --limit 50
python scripts/generate_scene_descriptions.py --input ./assets --format sharegpt
python scripts/generate_scene_descriptions.py --dry-run # List files without generating
python scripts/generate_scene_descriptions.py --input ./assets --check-model # Verify model availability
Ref: timmy-config#689
"""
import argparse
import base64
import json
import os
import re
import subprocess
import sys
import time
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
# Supported media extensions
IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"}
VIDEO_EXTS = {".mp4", ".webm", ".mov", ".avi", ".mkv"}
ALL_EXTS = IMAGE_EXTS | VIDEO_EXTS
# File size limit (50MB) — prevents unbounded memory usage on large images
MAX_FILE_SIZE = 50 * 1024 * 1024
# Vision models in preference order (best first)
VISION_MODELS = [
"gemma4:latest", # Gemma 4 — multimodal vision (8B, Q4_K_M)
"gemma3:12b", # Gemma 3 — fallback vision
"llava:latest", # LLaVA — generic vision
"llava-phi3:latest", # LLaVA-Phi3 — lightweight vision
]
# Vision model prompt template (structured JSON output)
SCENE_PROMPT = """Describe this image for a visual scene database. Output ONLY valid JSON (no markdown, no explanation):
{
"mood": "one of: calm, energetic, dark, warm, cool, chaotic, serene, tense, joyful, melancholic",
"colors": ["dominant color 1", "dominant color 2", "dominant color 3"],
"composition": "one of: close-up, wide-shot, medium-shot, low-angle, high-angle, bird-eye, profile, over-shoulder",
"camera": "one of: static, slow-pan, tracking, handheld, crane, dolly, steady, locked-off",
"lighting": "one of: natural, artificial, mixed, dramatic, soft, harsh, backlit",
"description": "2-3 sentence visual description of the scene"
}
Be specific. Describe what you see, not what you imagine."""
# ShareGPT format prompt (for training pipeline integration)
SHAREGPT_SCENE_PROMPT = """Analyze this image and describe the visual scene. Include mood, dominant colors, composition, camera angle, lighting, and a vivid 2-3 sentence description."""
def check_model_available(model: str, ollama_url: str = "http://localhost:11434") -> bool:
"""Check if a model is available in Ollama."""
try:
req = urllib.request.Request(f"{ollama_url}/api/tags")
resp = urllib.request.urlopen(req, timeout=10)
data = json.loads(resp.read())
available = [m["name"] for m in data.get("models", [])]
return model in available
except Exception:
return False
def auto_detect_model(ollama_url: str = "http://localhost:11434") -> Optional[str]:
"""Auto-detect the best available vision model."""
for model in VISION_MODELS:
if check_model_available(model, ollama_url):
print(f"Auto-detected vision model: {model}", file=sys.stderr)
return model
return None
def scan_media(input_dir: str) -> list[Path]:
"""Scan directory for media files recursively."""
media_files = []
input_path = Path(input_dir)
if not input_path.exists():
print(f"Error: {input_dir} does not exist", file=sys.stderr)
return media_files
for ext in sorted(ALL_EXTS):
media_files.extend(input_path.rglob(f"*{ext}"))
media_files.extend(input_path.rglob(f"*{ext.upper()}"))
return sorted(set(media_files))
def extract_video_frame(video_path: Path, output_path: Path) -> bool:
"""Extract a representative frame from a video using ffmpeg."""
try:
result = subprocess.run(
# FIX #3: Seek 2s in before grabbing frame — avoids black/title frames
["ffmpeg", "-ss", "2", "-i", str(video_path), "-vframes", "1",
"-q:v", "2", str(output_path), "-y"],
capture_output=True, timeout=30,
)
if result.returncode != 0 and result.stderr:
print(f" ffmpeg stderr: {result.stderr.decode(errors='replace')[:200]}", file=sys.stderr)
return output_path.exists() and output_path.stat().st_size > 0
except FileNotFoundError:
print(" ffmpeg not found — skipping video frame extraction", file=sys.stderr)
return False
except Exception as e:
print(f" ffmpeg error: {e}", file=sys.stderr)
return False
def describe_image(
image_path: Path,
model: str = "gemma4:latest",
ollama_url: str = "http://localhost:11434",
max_retries: int = 2,
) -> Optional[dict]:
"""Generate scene description using Ollama vision model with retry."""
# FIX #1: Check file size before reading into memory
if image_path.stat().st_size > MAX_FILE_SIZE:
print(f" Skipping {image_path.name}: exceeds {MAX_FILE_SIZE // (1024*1024)}MB limit", file=sys.stderr)
return None
for attempt in range(max_retries + 1):
try:
with open(image_path, "rb") as f:
image_b64 = base64.b64encode(f.read()).decode()
req = urllib.request.Request(
f"{ollama_url}/api/generate",
data=json.dumps({
"model": model,
"prompt": SCENE_PROMPT,
"images": [image_b64],
"stream": False,
"options": {"temperature": 0.3, "num_predict": 1024}
}).encode(),
headers={"Content-Type": "application/json"},
)
resp = urllib.request.urlopen(req, timeout=120)
data = json.loads(resp.read())
response_text = data.get("response", "")
# Parse JSON from response (handle both complete and truncated JSON)
json_match = re.search(r"\{[\s\S]*\}", response_text)
if not json_match:
# Try to find opening brace for truncated JSON
brace_match = re.search(r"\{", response_text)
if brace_match:
json_match = brace_match
if json_match:
raw_json = json_match.group() if hasattr(json_match, 'group') else response_text[json_match.start():]
# Try strict parse first
try:
parsed = json.loads(raw_json)
required = ["mood", "colors", "composition", "camera", "description"]
if all(k in parsed for k in required) and parsed.get("description"):
return parsed
except json.JSONDecodeError:
# Attempt repair: extract fields from truncated JSON
repaired = {}
for field in ["mood", "colors", "composition", "camera", "lighting", "description"]:
pat = rf'"\s*{field}"\s*:\s*"([^"]*)"'
m = re.search(pat, response_text)
if m:
repaired[field] = m.group(1)
elif field == "colors":
colors_match = re.search(r'"colors"\s*:\s*\[([^\]]*)\]', response_text)
if colors_match:
repaired[field] = [c.strip().strip('"') for c in colors_match.group(1).split(",") if c.strip()]
else:
repaired[field] = []
else:
repaired[field] = "unknown"
if repaired.get("description") or repaired.get("mood") != "unknown":
return repaired
# Final fallback: natural language response
clean = re.sub(r"[*_`#]", "", response_text).strip()
clean = re.sub(r"\n{3,}", "\n\n", clean)
return {
"description": clean[:500] if clean else response_text[:500],
"mood": "unknown",
"colors": [],
"composition": "unknown",
"camera": "unknown",
"lighting": "unknown"
}
except (urllib.error.URLError, TimeoutError) as e:
if attempt < max_retries:
wait = 2 ** attempt
print(f" Retry {attempt + 1}/{max_retries} after {wait}s: {e}", file=sys.stderr)
time.sleep(wait)
else:
print(f" Error describing {image_path.name}: {e}", file=sys.stderr)
return None
except Exception as e:
print(f" Error describing {image_path.name}: {e}", file=sys.stderr)
return None
def describe_image_sharegpt(
image_path: Path,
model: str = "gemma4:latest",
ollama_url: str = "http://localhost:11434",
max_retries: int = 2,
) -> Optional[str]:
"""Generate scene description in natural language for ShareGPT format."""
# FIX #1: Check file size before reading into memory
if image_path.stat().st_size > MAX_FILE_SIZE:
print(f" Skipping {image_path.name}: exceeds {MAX_FILE_SIZE // (1024*1024)}MB limit", file=sys.stderr)
return None
for attempt in range(max_retries + 1):
try:
with open(image_path, "rb") as f:
image_b64 = base64.b64encode(f.read()).decode()
req = urllib.request.Request(
f"{ollama_url}/api/generate",
data=json.dumps({
"model": model,
"prompt": SHAREGPT_SCENE_PROMPT,
"images": [image_b64],
"stream": False,
"options": {"temperature": 0.5, "num_predict": 256}
}).encode(),
headers={"Content-Type": "application/json"},
)
resp = urllib.request.urlopen(req, timeout=120)
data = json.loads(resp.read())
return data.get("response", "").strip()
except (urllib.error.URLError, TimeoutError) as e:
if attempt < max_retries:
time.sleep(2 ** attempt)
else:
return None
except Exception:
return None
def generate_training_pairs(
media_files: list[Path],
model: str,
ollama_url: str,
limit: int = 0,
dry_run: bool = False,
output_format: str = "jsonl",
) -> list[dict]:
"""Generate training pairs from media files."""
pairs = []
files = media_files[:limit] if limit > 0 else media_files
print(f"Processing {len(files)} files with model {model}...", file=sys.stderr)
for i, media_path in enumerate(files):
print(f" [{i + 1}/{len(files)}] {media_path.name}...", file=sys.stderr, end=" ", flush=True)
if dry_run:
print("(dry run)", file=sys.stderr)
pairs.append({"source": str(media_path), "status": "dry-run"})
continue
is_video = media_path.suffix.lower() in VIDEO_EXTS
work_path = media_path
if is_video:
frame_path = media_path.with_suffix(".frame.jpg")
if extract_video_frame(media_path, frame_path):
work_path = frame_path
else:
print("SKIP (frame extraction failed)", file=sys.stderr)
continue
try:
if output_format == "sharegpt":
# ShareGPT format for training pipeline
description = describe_image_sharegpt(work_path, model, ollama_url)
if description:
pair = {
"conversations": [
{"from": "human", "value": f"<image>\n{SHAREGPT_SCENE_PROMPT}"},
{"from": "gpt", "value": description}
],
"source": str(media_path),
"media_type": "video" if is_video else "image",
"model": model,
"generated_at": datetime.now(timezone.utc).isoformat(),
}
pairs.append(pair)
print("OK", file=sys.stderr)
else:
print("FAIL", file=sys.stderr)
else:
# Structured JSONL format
description = describe_image(work_path, model, ollama_url)
if description:
pair = {
"source": str(media_path),
"media_type": "video" if is_video else "image",
"description": description,
"model": model,
"generated_at": datetime.now(timezone.utc).isoformat(),
}
pairs.append(pair)
print("OK", file=sys.stderr)
else:
print("FAIL", file=sys.stderr)
finally:
# FIX #6: Cleanup temp frame in try/finally — survives crashes
if is_video and work_path != media_path:
try:
work_path.unlink()
except Exception:
pass
# Small delay between files (reduced from 0.5s — Ollama is local)
time.sleep(0.1)
return pairs
def main():
parser = argparse.ArgumentParser(
description="Auto-generate scene descriptions from media assets using vision AI"
)
parser.add_argument("--input", "-i", default="", help="Input directory with media files")
parser.add_argument("--output", "-o", default="training-data/scene-descriptions-auto.jsonl")
parser.add_argument("--model", "-m", default=None, help="Ollama model name (auto-detects best available if omitted)")
parser.add_argument("--ollama-url", default="http://localhost:11434")
parser.add_argument("--limit", "-l", type=int, default=0, help="Max files to process (0=all)")
parser.add_argument("--dry-run", action="store_true", help="List files without generating")
parser.add_argument("--check-model", action="store_true", help="Check model availability and exit")
parser.add_argument("--format", choices=["jsonl", "sharegpt"], default="jsonl",
help="Output format: jsonl (structured) or sharegpt (training pipeline)")
args = parser.parse_args()
# Model detection
if args.check_model:
if args.model:
available = check_model_available(args.model, args.ollama_url)
print(f"Model '{args.model}': {'✅ available' if available else '❌ not found'}")
else:
model = auto_detect_model(args.ollama_url)
if model:
print(f"✅ Best available: {model}")
else:
print("❌ No vision models found in Ollama — install one with: ollama pull gemma4:latest")
sys.exit(0)
# Auto-detect model if not specified
model = args.model
if not model:
model = auto_detect_model(args.ollama_url)
if not model:
# Fall back to best default even if not installed — let Ollama handle the error
model = "gemma4:latest"
print(f"Warning: No vision models detected. Falling back to {model}", file=sys.stderr)
# Validate input
if not args.input:
print("Error: --input is required (unless using --check-model)", file=sys.stderr)
sys.exit(1)
# Scan and process
media_files = scan_media(args.input)
print(f"Found {len(media_files)} media files", file=sys.stderr)
if not media_files:
print("No media files found.", file=sys.stderr)
sys.exit(1)
pairs = generate_training_pairs(
media_files, model, args.ollama_url,
args.limit, args.dry_run, args.format
)
# Write output
output_path = Path(args.output)
output_path.parent.mkdir(parents=True, exist_ok=True)
with open(output_path, "w") as f:
for pair in pairs:
f.write(json.dumps(pair, ensure_ascii=False) + "\n")
print(f"\nWrote {len(pairs)} pairs to {output_path}", file=sys.stderr)
# Summary
success = len([p for p in pairs if "description" in p or "conversations" in p])
failed = len(pairs) - success
if failed > 0:
print(f" ⚠️ {failed} files failed", file=sys.stderr)
if __name__ == "__main__":
main()

View File

@@ -19,9 +19,11 @@ from glitch_patterns import (
GlitchPattern,
GlitchSeverity,
MATRIX_GLITCH_PATTERNS,
THREEJS_CATEGORIES,
build_vision_prompt,
get_pattern_by_category,
get_patterns_by_severity,
get_threejs_patterns,
)
from matrix_glitch_detector import (
@@ -40,7 +42,7 @@ class TestGlitchPatterns(unittest.TestCase):
def test_pattern_count(self):
"""Verify we have a reasonable number of defined patterns."""
self.assertGreaterEqual(len(MATRIX_GLITCH_PATTERNS), 8)
self.assertGreaterEqual(len(MATRIX_GLITCH_PATTERNS), 14) # 10 generic + 6 Three.js
def test_all_patterns_have_required_fields(self):
"""Every pattern must have category, name, description, severity, prompts."""
@@ -88,6 +90,9 @@ class TestGlitchPatterns(unittest.TestCase):
self.assertIn("Floating Object", prompt)
self.assertIn("Z-Fighting", prompt)
self.assertIn("Missing", prompt)
# Three.js patterns should be included
self.assertIn("Shader Compilation Failure", prompt)
self.assertIn("Bloom Overflow", prompt)
def test_build_vision_prompt_subset(self):
"""Vision prompt with subset should only include specified patterns."""
@@ -248,7 +253,7 @@ class TestGlitchDetector(unittest.TestCase):
try:
report = run_demo(output_path)
self.assertEqual(len(report.glitches), 4)
self.assertEqual(len(report.glitches), 6) # 4 original + 2 Three.js
self.assertGreater(report.summary["total_glitches"], 0)
self.assertTrue(output_path.exists())
@@ -260,6 +265,93 @@ class TestGlitchDetector(unittest.TestCase):
output_path.unlink(missing_ok=True)
class TestThreeJsPatterns(unittest.TestCase):
"""Tests for Three.js-specific glitch patterns (timmy-config#543)."""
def test_get_threejs_patterns_returns_only_threejs(self):
"""get_threejs_patterns() should return only Three.js categories."""
patterns = get_threejs_patterns()
self.assertEqual(len(patterns), 6)
for p in patterns:
self.assertIn(p.category, THREEJS_CATEGORIES)
def test_threejs_patterns_have_required_fields(self):
"""All Three.js patterns must have valid fields."""
for p in get_threejs_patterns():
self.assertIsInstance(p.category, GlitchCategory)
self.assertTrue(p.name)
self.assertTrue(p.description)
self.assertIsInstance(p.severity, GlitchSeverity)
self.assertGreater(len(p.detection_prompts), 0)
self.assertGreater(len(p.visual_indicators), 0)
def test_shader_failure_is_critical(self):
"""Shader compilation failure should be CRITICAL severity."""
p = get_pattern_by_category(GlitchCategory.SHADER_FAILURE)
self.assertIsNotNone(p)
self.assertEqual(p.severity, GlitchSeverity.CRITICAL)
def test_texture_placeholder_is_critical(self):
"""Texture placeholder (1x1 white) should be CRITICAL severity."""
p = get_pattern_by_category(GlitchCategory.TEXTURE_PLACEHOLDER)
self.assertIsNotNone(p)
self.assertEqual(p.severity, GlitchSeverity.CRITICAL)
def test_infer_severity_shader_failure(self):
"""Shader failure should infer critical/high."""
self.assertEqual(_infer_severity("shader_failure", 0.8), "critical")
self.assertEqual(_infer_severity("shader_failure", 0.5), "high")
def test_infer_severity_texture_placeholder(self):
"""Texture placeholder should infer critical/high."""
self.assertEqual(_infer_severity("texture_placeholder", 0.8), "critical")
self.assertEqual(_infer_severity("texture_placeholder", 0.5), "high")
def test_infer_severity_uv_mapping(self):
"""UV mapping error should infer high/medium."""
self.assertEqual(_infer_severity("uv_mapping_error", 0.8), "high")
self.assertEqual(_infer_severity("uv_mapping_error", 0.5), "medium")
def test_infer_severity_frustum_culling(self):
"""Frustum culling should infer medium/low."""
self.assertEqual(_infer_severity("frustum_culling", 0.7), "medium")
self.assertEqual(_infer_severity("frustum_culling", 0.4), "low")
def test_infer_severity_shadow_map(self):
"""Shadow map artifact should infer medium/low."""
self.assertEqual(_infer_severity("shadow_map_artifact", 0.7), "medium")
self.assertEqual(_infer_severity("shadow_map_artifact", 0.4), "low")
def test_infer_severity_bloom_overflow(self):
"""Bloom overflow should infer medium/low (default path)."""
self.assertEqual(_infer_severity("bloom_overflow", 0.7), "medium")
self.assertEqual(_infer_severity("bloom_overflow", 0.4), "low")
def test_threejs_patterns_in_vision_prompt(self):
"""Three.js patterns should appear in the composite vision prompt."""
prompt = build_vision_prompt()
self.assertIn("shader_failure", prompt)
self.assertIn("texture_placeholder", prompt)
self.assertIn("uv_mapping_error", prompt)
self.assertIn("frustum_culling", prompt)
self.assertIn("shadow_map_artifact", prompt)
self.assertIn("bloom_overflow", prompt)
def test_threejs_subset_prompt(self):
"""Building prompt from Three.js-only patterns should work."""
threejs = get_threejs_patterns()
prompt = build_vision_prompt(threejs)
self.assertIn("Shader Compilation Failure", prompt)
self.assertNotIn("Floating Object", prompt) # generic, not Three.js
def test_report_metadata_version(self):
"""Report metadata should reference both issues."""
report = run_demo()
self.assertEqual(report.metadata["detector_version"], "0.2.0")
self.assertIn("543", report.metadata["reference"])
class TestIntegration(unittest.TestCase):
"""Integration-level tests."""
@@ -276,6 +368,13 @@ class TestIntegration(unittest.TestCase):
expected = {"floating_assets", "z_fighting", "missing_textures", "clipping", "broken_normals"}
self.assertTrue(expected.issubset(category_values))
def test_patterns_cover_threejs_themes(self):
"""Patterns should cover Three.js-specific glitch themes (#543)."""
category_values = {p.category.value for p in MATRIX_GLITCH_PATTERNS}
threejs_expected = {"shader_failure", "texture_placeholder", "uv_mapping_error",
"frustum_culling", "shadow_map_artifact", "bloom_overflow"}
self.assertTrue(threejs_expected.issubset(category_values))
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,333 @@
#!/usr/bin/env python3
"""
Tests for generate_scene_descriptions.py
Tests the scene description generation pipeline including:
- Media file scanning
- Model detection
- JSON parsing from vision responses
- Output format validation
Ref: timmy-config#689
"""
import json
import tempfile
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch
# Add scripts to path for import
import sys
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
from generate_scene_descriptions import (
IMAGE_EXTS,
VIDEO_EXTS,
ALL_EXTS,
VISION_MODELS,
auto_detect_model,
check_model_available,
scan_media,
extract_video_frame,
)
class TestMediaScanning(unittest.TestCase):
"""Test media file scanning."""
def test_scan_empty_directory(self):
with tempfile.TemporaryDirectory() as tmpdir:
result = scan_media(tmpdir)
self.assertEqual(result, [])
def test_scan_nonexistent_directory(self):
result = scan_media("/nonexistent/path/that/does/not/exist")
self.assertEqual(result, [])
def test_scan_with_images(self):
with tempfile.TemporaryDirectory() as tmpdir:
# Create test files
for ext in [".jpg", ".png", ".webp"]:
(Path(tmpdir) / f"test{ext}").touch()
result = scan_media(tmpdir)
self.assertEqual(len(result), 3)
def test_scan_recursive(self):
with tempfile.TemporaryDirectory() as tmpdir:
subdir = Path(tmpdir) / "sub" / "dir"
subdir.mkdir(parents=True)
(subdir / "deep.jpg").touch()
(Path(tmpdir) / "top.png").touch()
result = scan_media(tmpdir)
self.assertEqual(len(result), 2)
def test_scan_ignores_unsupported(self):
with tempfile.TemporaryDirectory() as tmpdir:
(Path(tmpdir) / "image.jpg").touch()
(Path(tmpdir) / "document.pdf").touch()
(Path(tmpdir) / "script.py").touch()
result = scan_media(tmpdir)
self.assertEqual(len(result), 1)
def test_scan_sorted_output(self):
with tempfile.TemporaryDirectory() as tmpdir:
for name in ["z.jpg", "a.png", "m.webp"]:
(Path(tmpdir) / name).touch()
result = scan_media(tmpdir)
names = [p.name for p in result]
self.assertEqual(names, sorted(names))
class TestModelDetection(unittest.TestCase):
"""Test model availability detection."""
@patch('generate_scene_descriptions.urllib.request.urlopen')
def test_check_model_available(self, mock_urlopen):
mock_resp = MagicMock()
mock_resp.read.return_value = json.dumps({
"models": [{"name": "gemma4:latest"}]
}).encode()
mock_urlopen.return_value.__enter__ = MagicMock(return_value=mock_resp)
mock_urlopen.return_value.__exit__ = MagicMock(return_value=False)
mock_urlopen.return_value = mock_resp
result = check_model_available("gemma4:latest")
self.assertTrue(result)
@patch('generate_scene_descriptions.urllib.request.urlopen')
def test_check_model_not_available(self, mock_urlopen):
mock_resp = MagicMock()
mock_resp.read.return_value = json.dumps({
"models": [{"name": "llama2:7b"}]
}).encode()
mock_urlopen.return_value = mock_resp
result = check_model_available("gemma4:latest")
self.assertFalse(result)
@patch('generate_scene_descriptions.check_model_available')
def test_auto_detect_prefers_gemma4(self, mock_check):
def side_effect(model, url):
return model == "gemma4:latest"
mock_check.side_effect = side_effect
result = auto_detect_model()
self.assertEqual(result, "gemma4:latest")
@patch('generate_scene_descriptions.check_model_available')
def test_auto_detect_falls_back(self, mock_check):
def side_effect(model, url):
return model == "llava:latest"
mock_check.side_effect = side_effect
result = auto_detect_model()
self.assertEqual(result, "llava:latest")
@patch('generate_scene_descriptions.check_model_available')
def test_auto_detect_returns_none_when_no_models(self, mock_check):
mock_check.return_value = False
result = auto_detect_model()
self.assertIsNone(result)
class TestConstants(unittest.TestCase):
"""Test constant definitions."""
def test_image_extensions(self):
self.assertIn(".jpg", IMAGE_EXTS)
self.assertIn(".png", IMAGE_EXTS)
self.assertIn(".webp", IMAGE_EXTS)
def test_video_extensions(self):
self.assertIn(".mp4", VIDEO_EXTS)
self.assertIn(".webm", VIDEO_EXTS)
def test_all_extensions_union(self):
self.assertEqual(ALL_EXTS, IMAGE_EXTS | VIDEO_EXTS)
def test_vision_models_ordered(self):
self.assertEqual(VISION_MODELS[0], "gemma4:latest")
self.assertIn("llava:latest", VISION_MODELS)
class TestVideoFrameExtraction(unittest.TestCase):
"""Test video frame extraction."""
def test_extract_nonexistent_video(self):
result = extract_video_frame(Path("/nonexistent.mp4"), Path("/tmp/frame.jpg"))
self.assertFalse(result)
class TestDescribeImage(unittest.TestCase):
"""Test describe_image() with mocked Ollama responses."""
def test_skips_oversized_file(self):
"""Files exceeding MAX_FILE_SIZE should be skipped without API call."""
import generate_scene_descriptions
with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as f:
f.write(b"\x00" * (51 * 1024 * 1024))
f.flush()
result = generate_scene_descriptions.describe_image(Path(f.name))
Path(f.name).unlink()
self.assertIsNone(result)
@patch('generate_scene_descriptions.urllib.request.urlopen')
def test_parses_valid_json_response(self, mock_urlopen):
"""Valid JSON response should be parsed and returned."""
import generate_scene_descriptions
resp_data = {
"response": '{"mood": "calm", "colors": ["blue", "white"], "composition": "wide-shot", "camera": "static", "lighting": "natural", "description": "A serene ocean scene."}'
}
mock_resp = MagicMock()
mock_resp.read.return_value = json.dumps(resp_data).encode()
mock_urlopen.return_value = mock_resp
with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as f:
f.write(b"\xff\xd8\xff\xe0" + b"\x00" * 1000)
f.flush()
result = generate_scene_descriptions.describe_image(Path(f.name))
Path(f.name).unlink()
self.assertIsNotNone(result)
self.assertEqual(result["mood"], "calm")
self.assertIn("lighting", result)
@patch('generate_scene_descriptions.urllib.request.urlopen')
def test_repair_truncated_json(self, mock_urlopen):
"""Truncated JSON should be repaired with regex extraction."""
import generate_scene_descriptions
resp_data = {
"response": '{"mood": "dark", "colors": ["red"], "composition": "close-up", "camera": "handheld", "lighting": "dramatic", "description": "A shadowy figure in a dimly lit alley'
}
mock_resp = MagicMock()
mock_resp.read.return_value = json.dumps(resp_data).encode()
mock_urlopen.return_value = mock_resp
with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as f:
f.write(b"\xff\xd8\xff\xe0" + b"\x00" * 1000)
f.flush()
result = generate_scene_descriptions.describe_image(Path(f.name))
Path(f.name).unlink()
self.assertIsNotNone(result)
self.assertEqual(result["mood"], "dark")
self.assertEqual(result["lighting"], "dramatic")
@patch('generate_scene_descriptions.urllib.request.urlopen')
def test_fallback_on_invalid_json(self, mock_urlopen):
"""Completely invalid JSON response should still return a fallback."""
import generate_scene_descriptions
resp_data = {"response": "This is just plain text describing a beautiful sunset over mountains."}
mock_resp = MagicMock()
mock_resp.read.return_value = json.dumps(resp_data).encode()
mock_urlopen.return_value = mock_resp
with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as f:
f.write(b"\xff\xd8\xff\xe0" + b"\x00" * 1000)
f.flush()
result = generate_scene_descriptions.describe_image(Path(f.name))
Path(f.name).unlink()
self.assertIsNotNone(result)
self.assertIn("description", result)
self.assertIn("lighting", result)
class TestDescribeImageSharegpt(unittest.TestCase):
"""Test describe_image_sharegpt() with mocked Ollama responses."""
def test_skips_oversized_file(self):
"""Files exceeding MAX_FILE_SIZE should be skipped."""
import generate_scene_descriptions
with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as f:
f.write(b"\x00" * (51 * 1024 * 1024))
f.flush()
result = generate_scene_descriptions.describe_image_sharegpt(Path(f.name))
Path(f.name).unlink()
self.assertIsNone(result)
@patch('generate_scene_descriptions.urllib.request.urlopen')
def test_returns_natural_language(self, mock_urlopen):
"""Should return the raw response text."""
import generate_scene_descriptions
resp_data = {"response": "A warm sunset over rolling hills with golden light."}
mock_resp = MagicMock()
mock_resp.read.return_value = json.dumps(resp_data).encode()
mock_urlopen.return_value = mock_resp
with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as f:
f.write(b"\xff\xd8\xff\xe0" + b"\x00" * 1000)
f.flush()
result = generate_scene_descriptions.describe_image_sharegpt(Path(f.name))
Path(f.name).unlink()
self.assertIsNotNone(result)
self.assertIn("sunset", result)
class TestGenerateTrainingPairs(unittest.TestCase):
"""Test generate_training_pairs() orchestration."""
@patch('generate_scene_descriptions.describe_image')
def test_jsonl_output_format(self, mock_describe):
"""JSONL format should produce structured description objects."""
import generate_scene_descriptions
mock_describe.return_value = {"mood": "calm", "description": "Test"}
with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as f:
f.write(b"\x00" * 1000)
f.flush()
pairs = generate_scene_descriptions.generate_training_pairs(
[Path(f.name)], "test-model", "http://localhost:11434",
output_format="jsonl"
)
Path(f.name).unlink()
self.assertEqual(len(pairs), 1)
self.assertIn("description", pairs[0])
self.assertIn("generated_at", pairs[0])
@patch('generate_scene_descriptions.describe_image_sharegpt')
def test_sharegpt_output_format(self, mock_describe):
"""ShareGPT format should produce conversation objects."""
import generate_scene_descriptions
mock_describe.return_value = "A description of the scene."
with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as f:
f.write(b"\x00" * 1000)
f.flush()
pairs = generate_scene_descriptions.generate_training_pairs(
[Path(f.name)], "test-model", "http://localhost:11434",
output_format="sharegpt"
)
Path(f.name).unlink()
self.assertEqual(len(pairs), 1)
self.assertIn("conversations", pairs[0])
self.assertEqual(len(pairs[0]["conversations"]), 2)
@patch('generate_scene_descriptions.describe_image')
def test_dry_run_skips_api_calls(self, mock_describe):
"""Dry run should not call describe_image."""
import generate_scene_descriptions
with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as f:
f.write(b"\x00" * 1000)
f.flush()
pairs = generate_scene_descriptions.generate_training_pairs(
[Path(f.name)], "test-model", "http://localhost:11434",
dry_run=True
)
Path(f.name).unlink()
mock_describe.assert_not_called()
self.assertEqual(len(pairs), 1)
self.assertEqual(pairs[0]["status"], "dry-run")
if __name__ == "__main__":
unittest.main()