416 lines
15 KiB
Python
416 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""File-based TAC Product Factory MVP."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
CATALOG = ROOT / "catalog.json"
|
|
EVENT_DIR = ROOT / ".events"
|
|
PRODUCTS_DIR = ROOT / "generated"
|
|
DASHBOARD = ROOT / "DASHBOARD.md"
|
|
PLANS_DIR = ROOT.parent / "plans"
|
|
RECEIPTS_DIR = ROOT / "receipts"
|
|
RSI_EVIDENCE_RECEIPT = "packages/pi-real-engineering/docs/receipts/rsi-proof-20260619T093724Z.md"
|
|
LOCAL_RSI_EVIDENCE_RECEIPT = "products/receipts/rsi-proof-20260619T093724Z.md"
|
|
RSI_DECISION = "healthy_but_autopatch_unproven"
|
|
RSI_CLAIM = (
|
|
"Factory RSI is healthy at 65/65 with cron, skill_health, factory_watcher, "
|
|
"and deploy_route present; auto_patch_proven=false until a degraded-skill "
|
|
"recovery receipt exists; rsi_canary evidence is tracked separately"
|
|
)
|
|
FACTORY_STATE = {
|
|
"factory_phase": 4,
|
|
"agent_rsi_phases": "A-D operational",
|
|
"containers": 25,
|
|
"agents": 6,
|
|
"infra": 14,
|
|
"monitoring_deploy": 5,
|
|
"skills": 63,
|
|
"latest_score": "65/65",
|
|
"factory_watcher": True,
|
|
"deploy_route_present": True,
|
|
"root_commit": "55bdf3f",
|
|
"hermes_agent_commit": "fd3053b",
|
|
}
|
|
LOOP_ENGINEERING = {
|
|
"agent_loop": "ADW phases run tools until product task completion",
|
|
"verification_loop": "plan validation, pytest checks, verifier_result, and receipts gate readiness",
|
|
"event_driven_loop": "GitHub webhooks, cron skill_health, event JSONL, and dashboard refresh trigger work",
|
|
"hill_climbing_loop": "skill_health/SkillOpt can propose signed action patches; guarded by receipts and rollback",
|
|
}
|
|
|
|
|
|
def slugify(text: str) -> str:
|
|
slug = re.sub(r"[^a-zA-Z0-9]+", "-", text.lower()).strip("-")
|
|
return slug[:60] or "product"
|
|
|
|
|
|
def load_catalog() -> list[dict]:
|
|
return json.loads(CATALOG.read_text(encoding="utf-8")) if CATALOG.exists() else []
|
|
|
|
|
|
def log_event(kind: str, **data) -> Path:
|
|
EVENT_DIR.mkdir(parents=True, exist_ok=True)
|
|
path = EVENT_DIR / f"{datetime.now(timezone.utc).date().isoformat()}.jsonl"
|
|
event = {"ts": datetime.now(timezone.utc).isoformat(), "kind": kind, **data}
|
|
with path.open("a", encoding="utf-8") as f:
|
|
f.write(json.dumps(event, ensure_ascii=False) + "\n")
|
|
return path
|
|
|
|
|
|
def living_plan_metadata(product_id: str) -> dict:
|
|
return {
|
|
"plan_id": product_id,
|
|
"plan_ref": f"plans/{product_id}/plan.md",
|
|
"format": ["markdown", "html-first", "append-only-amendments"],
|
|
"required_sections": ["purpose", "problem", "solution", "files", "phases", "validation", "references", "notes"],
|
|
"deploy_endpoint": "git-proxy:8099/deploy",
|
|
"legacy_internal": ["deploy-webhook:8098"],
|
|
"rsi_decision": RSI_DECISION,
|
|
"auto_patch_proven": False,
|
|
"rsi_canary_recovery_evidence": True,
|
|
"rsi_evidence_receipt": RSI_EVIDENCE_RECEIPT,
|
|
"local_rsi_evidence_receipt": LOCAL_RSI_EVIDENCE_RECEIPT,
|
|
"rsi_claim": RSI_CLAIM,
|
|
"factory_state": FACTORY_STATE,
|
|
"loop_engineering": LOOP_ENGINEERING,
|
|
}
|
|
|
|
|
|
def render_plan_md(product_id: str, purpose: str) -> str:
|
|
return f"""---
|
|
product_id: {product_id}
|
|
purpose: {purpose}
|
|
references:
|
|
- products/AGENT_PRODUCTS.md
|
|
---
|
|
|
|
## Purpose
|
|
{purpose}
|
|
|
|
## Problem
|
|
Define the product problem before build.
|
|
|
|
## Solution
|
|
Build the smallest agent-run product that satisfies the plan.
|
|
|
|
## Files
|
|
- products/generated/{product_id}/spec.json
|
|
- products/generated/{product_id}/README.md
|
|
|
|
## Phases
|
|
- [ ] Plan
|
|
- [ ] Build
|
|
- [ ] Test
|
|
- [ ] Verify
|
|
- [ ] Signed deploy
|
|
|
|
## Validation
|
|
- uv run --with pytest pytest -q
|
|
|
|
## References
|
|
- products/AGENT_PRODUCTS.md
|
|
|
|
## Notes
|
|
Plan F3 living artifact. Amendments append below.
|
|
"""
|
|
|
|
|
|
def render_plan_html(product_id: str, purpose: str) -> str:
|
|
return f"""<section data-plan=\"{product_id}\">
|
|
<h1>{product_id}</h1>
|
|
<h2>Purpose</h2><p>{purpose}</p>
|
|
<h2>Problem</h2><p>Define the product problem before build.</p>
|
|
<h2>Solution</h2><p>Build the smallest agent-run product that satisfies the plan.</p>
|
|
<h2>Files</h2><p>See plan.md.</p>
|
|
<h2>Phases</h2><ul><li>Plan</li><li>Build</li><li>Test</li><li>Verify</li><li>Signed deploy</li></ul>
|
|
<h2>Validation</h2><p>uv run --with pytest pytest -q</p>
|
|
<h2>References</h2><p>products/AGENT_PRODUCTS.md</p>
|
|
<h2>Notes</h2><p>Plan F3 living artifact.</p>
|
|
</section>
|
|
"""
|
|
|
|
|
|
def create_plan(product_id: str, purpose: str, overwrite: bool = False) -> Path:
|
|
path = PLANS_DIR / product_id
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
md = path / "plan.md"
|
|
html = path / "plan.html"
|
|
if md.exists() and not overwrite:
|
|
raise FileExistsError(f"plan already exists: {md}")
|
|
md.write_text(render_plan_md(product_id, purpose), encoding="utf-8")
|
|
html.write_text(render_plan_html(product_id, purpose), encoding="utf-8")
|
|
return md
|
|
|
|
|
|
def validate_product_plan(product_id: str) -> dict:
|
|
from products.plan_validator import validate_plan
|
|
return validate_plan(PLANS_DIR / product_id / "plan.md", strict=True)
|
|
|
|
|
|
def run_adw_pipeline(product_id: str, zte: bool = False) -> dict:
|
|
try:
|
|
from adw_modules.adw_pipeline import run_pipeline
|
|
return run_pipeline(product_id, zte=zte)
|
|
except Exception as exc:
|
|
return {"ok": False, "error": str(exc), "phases": []}
|
|
|
|
|
|
def latest_event(kind: str | None = None) -> dict | None:
|
|
if not EVENT_DIR.exists():
|
|
return None
|
|
for path in sorted(EVENT_DIR.glob("*.jsonl"), reverse=True):
|
|
for line in reversed(path.read_text(encoding="utf-8").splitlines()):
|
|
event = json.loads(line)
|
|
if kind is None or event.get("kind") == kind:
|
|
return event
|
|
return None
|
|
|
|
|
|
def write_receipt(product_id: str, status: str, checks_ok: bool, plan_ok: bool, deploy_route: str) -> Path:
|
|
RECEIPTS_DIR.mkdir(parents=True, exist_ok=True)
|
|
ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
|
path = RECEIPTS_DIR / f"{ts}-{product_id}.json"
|
|
receipt = {
|
|
"product_id": product_id,
|
|
"status": status,
|
|
"checks_ok": checks_ok,
|
|
"plan_ok": plan_ok,
|
|
"deploy_route": deploy_route,
|
|
"rsi_decision": RSI_DECISION,
|
|
"auto_patch_proven": False,
|
|
"rsi_canary_recovery_evidence": True,
|
|
"rsi_evidence_receipt": RSI_EVIDENCE_RECEIPT,
|
|
"local_rsi_evidence_receipt": LOCAL_RSI_EVIDENCE_RECEIPT,
|
|
"rsi_claim": RSI_CLAIM,
|
|
"factory_state": FACTORY_STATE,
|
|
"loop_engineering": LOOP_ENGINEERING,
|
|
}
|
|
path.write_text(json.dumps(receipt, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def create_product(idea: str, overwrite: bool = False) -> dict:
|
|
product_id = slugify(idea)
|
|
path = PRODUCTS_DIR / product_id
|
|
if path.exists() and not overwrite:
|
|
raise FileExistsError(f"product already exists: {product_id}")
|
|
|
|
event_path = log_event("idea_received", product_id=product_id, idea=idea)
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
plan_path = create_plan(product_id, idea, overwrite=overwrite)
|
|
plan_check = validate_product_plan(product_id)
|
|
spec = {
|
|
"id": product_id,
|
|
"idea": idea,
|
|
"spine": ["plan", "build", "test", "verify", "signed-deploy", "observe", "improve"],
|
|
"skills": [
|
|
"agent-chain",
|
|
"agent-team",
|
|
"verifier-builder",
|
|
"confidence-ladder",
|
|
"damage-control",
|
|
"tool-call-tracer",
|
|
"experiment-loop",
|
|
"tilldone",
|
|
],
|
|
"agents": ["planner", "builder", "reviewer", "executor", "tracker"],
|
|
"deploy_actions": ["patch_skill_from_pr"],
|
|
"status": "specified",
|
|
"living_plan": living_plan_metadata(product_id),
|
|
"plan_valid": plan_check["ok"],
|
|
"singularity": {
|
|
"class": 3,
|
|
"grade": 3,
|
|
"layer": "orchestrator-adw",
|
|
"principle": "build the system that builds the system",
|
|
},
|
|
}
|
|
(path / "spec.json").write_text(json.dumps(spec, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
(path / "README.md").write_text(render_readme(spec), encoding="utf-8")
|
|
log_event("product_spec_created", product_id=product_id, idea=idea, spec=str(path / "spec.json"))
|
|
log_event(
|
|
"plan_written",
|
|
product_id=product_id,
|
|
plan=str(plan_path),
|
|
plan_id=product_id,
|
|
plan_ref=living_plan_metadata(product_id)["plan_ref"],
|
|
plan_ok=plan_check["ok"],
|
|
)
|
|
log_event("adw_pipeline_started", product_id=product_id)
|
|
pipeline_result = run_adw_pipeline(product_id)
|
|
checks = ["plan", "build", "test", "review", "document", "ship"]
|
|
checks_ok = bool(pipeline_result.get("ok")) and plan_check["ok"]
|
|
log_event(
|
|
"checks_run",
|
|
product_id=product_id,
|
|
checks=checks,
|
|
checks_ok=checks_ok,
|
|
plan_ok=plan_check["ok"],
|
|
plan_id=product_id,
|
|
plan_ref=living_plan_metadata(product_id)["plan_ref"],
|
|
)
|
|
log_event("verifier_result", product_id=product_id, ok=checks_ok)
|
|
deploy_route = "git-proxy:8099/deploy"
|
|
log_event("deploy_action_requested", product_id=product_id, action="patch_skill_from_pr", route=deploy_route)
|
|
status = "ready" if checks_ok else "needs_attention"
|
|
receipt_path = write_receipt(product_id, status, checks_ok, plan_check["ok"], deploy_route)
|
|
log_event("outcome", product_id=product_id, status=status, receipt=str(receipt_path))
|
|
render_dashboard()
|
|
return {"product_id": product_id, "path": str(path), "plan": str(plan_path), "receipt": str(receipt_path), "event_log": str(event_path)}
|
|
|
|
|
|
def render_readme(spec: dict) -> str:
|
|
return f"""# {spec['id']}
|
|
|
|
Idea: {spec['idea']}
|
|
|
|
## Agent spine
|
|
|
|
`{' -> '.join(spec['spine'])}`
|
|
|
|
## Skills
|
|
|
|
{chr(10).join(f"- `{s}`" for s in spec['skills'])}
|
|
|
|
## Deploy
|
|
|
|
Only signed factory actions are allowed:
|
|
|
|
{chr(10).join(f"- `{a}`" for a in spec['deploy_actions'])}
|
|
|
|
## Living plan
|
|
|
|
Plan ref: `{spec['living_plan']['plan_ref']}`
|
|
|
|
Deploy endpoint: `{spec['living_plan']['deploy_endpoint']}`
|
|
|
|
RSI claim: {spec['living_plan']['rsi_claim']}.
|
|
|
|
RSI evidence receipt: `{spec['living_plan']['rsi_evidence_receipt']}`.
|
|
Local evidence mirror: `{spec['living_plan']['local_rsi_evidence_receipt']}`.
|
|
|
|
## Factory state
|
|
|
|
{chr(10).join(f"- `{k}`: {v}" for k, v in spec['living_plan']['factory_state'].items())}
|
|
|
|
## Loop engineering
|
|
|
|
{chr(10).join(f"- `{k}`: {v}" for k, v in spec['living_plan']['loop_engineering'].items())}
|
|
|
|
## Singularity ladder
|
|
|
|
Class {spec['singularity']['class']}, Grade {spec['singularity']['grade']}: `{spec['singularity']['layer']}`
|
|
|
|
Principle: {spec['singularity']['principle']}.
|
|
"""
|
|
|
|
|
|
def render_ladder() -> str:
|
|
rows = ["| Product | Class | Grade | Agentic layer |", "|---|---:|---:|---|"]
|
|
for item in load_catalog():
|
|
rows.append(
|
|
f"| {item['name']} | {item.get('singularity_class', '')} | "
|
|
f"{item.get('singularity_grade', '')} | {item.get('agentic_layer', '')} |"
|
|
)
|
|
return "\n".join(rows)
|
|
|
|
|
|
def render_dashboard() -> str:
|
|
rows = [
|
|
"| Product | Price | Status | Class | Grade | Agentic layer | Skills | Deploy actions |",
|
|
"|---|---:|---|---:|---:|---|---:|---|",
|
|
]
|
|
for item in load_catalog():
|
|
rows.append(
|
|
f"| {item['name']} | ${item.get('price', 0)} | {item.get('status', '')} | "
|
|
f"{item.get('singularity_class', '')} | {item.get('singularity_grade', '')} | "
|
|
f"{item.get('agentic_layer', '')} | {len(item.get('skills', []))} | "
|
|
f"{', '.join(item.get('deploy_actions', [])) or 'none'} |"
|
|
)
|
|
generated = sorted((PRODUCTS_DIR).glob("*/spec.json")) if PRODUCTS_DIR.exists() else []
|
|
last_check = latest_event("checks_run")
|
|
outcome = latest_event("outcome")
|
|
last_check_text = (
|
|
f"- `{last_check['product_id']}`: checks_ok=`{last_check.get('checks_ok')}` plan_ok=`{last_check.get('plan_ok')}` checks={', '.join(last_check.get('checks', []))}"
|
|
if last_check else "- No checks logged yet"
|
|
)
|
|
readiness_text = (
|
|
f"- `{outcome['product_id']}`: {outcome.get('status', 'unknown')} via signed `git-proxy:8099/deploy`; receipt={outcome.get('receipt', 'missing')}"
|
|
if outcome else "- No outcome logged yet"
|
|
)
|
|
text = f"""# TAC Products Dashboard
|
|
|
|
{chr(10).join(rows)}
|
|
|
|
## Generated products
|
|
|
|
{chr(10).join(f'- `{p.parent.name}`' for p in generated) or '- none yet'}
|
|
|
|
## Last check
|
|
|
|
{last_check_text}
|
|
|
|
## Deploy readiness
|
|
|
|
{readiness_text}
|
|
|
|
## Plan status
|
|
|
|
- Plans live in `plans/<product_id>/plan.md` and must validate before build/deploy.
|
|
- Amendments are append-only via `## Amendment <UTC timestamp>`.
|
|
|
|
## Open risks
|
|
|
|
- Do not overclaim arbitrary self-improvement; only verified canary/receipts belong here.
|
|
- Legacy direct `:8098` deploy-webhook is internal/stale for guarded deploy; use `:8099/deploy`.
|
|
|
|
## Current factory truth
|
|
|
|
- RSI status: live factory reports {FACTORY_STATE['latest_score']} healthy, cron + skill_health present, factory_watcher active, deploy_route present, and `auto_patch_proven=false`; `rsi_canary` recovery evidence is linked at `{RSI_EVIDENCE_RECEIPT}` and mirrored at `{LOCAL_RSI_EVIDENCE_RECEIPT}`.
|
|
- Engine factory memory: {FACTORY_STATE['containers']} containers, {FACTORY_STATE['agents']} agents, {FACTORY_STATE['infra']} infra, {FACTORY_STATE['monitoring_deploy']} monitoring/deploy, {FACTORY_STATE['skills']} skills.
|
|
- Commit evidence: root `{FACTORY_STATE['root_commit']}`, hermes-agent `{FACTORY_STATE['hermes_agent_commit']}`.
|
|
- Deploy path must use signed actions through git-proxy `:8099/deploy`, not raw shell or direct legacy `:8098` deploy.
|
|
|
|
## Loop engineering
|
|
|
|
- Loop 1 agent: ADW planner/builder/tester/reviewer/deployer phases run tools until done.
|
|
- Loop 2 verification: plan validation, pytest, verifier events, and receipts gate readiness.
|
|
- Loop 3 event-driven: GitHub webhooks, cron `skill_health`, JSONL events, and dashboard refresh trigger work.
|
|
- Loop 4 hill-climbing: SkillOpt/skill_health can request signed patches, but broader auto-patch autonomy stays unproven without degraded-skill recovery receipts.
|
|
|
|
## Next action
|
|
|
|
```bash
|
|
python products/product_factory.py "agent-powered customer support triage product"
|
|
python products/product_factory.py ladder
|
|
uv run --with pytest pytest -q
|
|
```
|
|
"""
|
|
DASHBOARD.write_text(text, encoding="utf-8")
|
|
return text
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
argv = argv or sys.argv[1:]
|
|
if argv == ["dashboard"]:
|
|
print(render_dashboard())
|
|
return 0
|
|
if argv == ["ladder"]:
|
|
print(render_ladder())
|
|
return 0
|
|
if not argv:
|
|
print("usage: python products/product_factory.py '<product idea>' | dashboard | ladder", file=sys.stderr)
|
|
return 2
|
|
print(json.dumps(create_product(" ".join(argv)), indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|