Buckets:

arrmlet
/

tracecraft-website

Files

xet

arrmlet/tracecraft-website / landing-page /artifacts /implementation /app.py

arrmlet

4 months ago

download

raw

4.55 kB

	"""Tracecraft Dashboard — FastAPI backend that shells out to `tracecraft` CLI."""

	import re
	import subprocess
	from pathlib import Path

	from fastapi import FastAPI
	from fastapi.responses import FileResponse

	app = FastAPI(title="Tracecraft Dashboard")

	KNOWN_STEPS = ["design", "implementation", "review", "deploy"]


	def run(cmd: str) -> str:
	"""Run a shell command and return stdout."""
	result = subprocess.run(
	cmd, shell=True, capture_output=True, text=True, timeout=15
	)
	return result.stdout.strip()


	# ── API endpoints ────────────────────────────────────────────────────────


	@app.get("/api/agents")
	def get_agents():
	"""Parse `tracecraft agents` table output."""
	raw = run("tracecraft agents")
	agents = []
	for line in raw.splitlines():
	# Skip header/separator lines
	if not line.strip() or line.startswith("ID") or line.startswith("---"):
	continue
	parts = line.split()
	if len(parts) >= 4:
	agents.append({
	"name": parts[0],
	"status": parts[1],
	"step": parts[2] if parts[2] != "-" else "",
	"heartbeat": " ".join(parts[3:]),
	})
	return {"agents": agents}


	@app.get("/api/memory")
	def get_memory():
	"""List memory keys, then fetch each value."""
	raw = run("tracecraft memory list")
	entries = []
	for key in raw.splitlines():
	key = key.strip()
	if not key:
	continue
	value = run(f"tracecraft memory get {key}")
	entries.append({"key": key, "value": value})
	return {"memory": entries}


	@app.get("/api/messages")
	def get_messages():
	"""Parse `tracecraft inbox` output.

	Expected format:
	[<timestamp>] (<channel>) <sender>: <body>
	or: [<timestamp>] <sender> -> <recipient>: <body>
	"""
	raw = run("tracecraft inbox")
	messages = []
	for line in raw.splitlines():
	line = line.strip()
	if not line or line == "No messages.":
	continue
	# Format: [timestamp] (channel) sender: body
	m = re.match(
	r"\[([^\]]+)\]\s+\((\w+)\)\s+(\w+):\s+(.*)", line
	)
	if m:
	channel = m.group(2)
	sender = m.group(3)
	if channel == "broadcast":
	to = "_broadcast"
	else:
	# direct message — to is us (developer)
	to = "developer"
	messages.append({
	"timestamp": m.group(1),
	"from": sender,
	"to": to,
	"body": m.group(4),
	})
	continue
	# Fallback — treat entire line as body
	messages.append({"timestamp": "", "from": "", "to": "", "body": line})
	return {"messages": messages}


	@app.get("/api/steps")
	def get_steps():
	"""Query step-status for each known step."""
	steps = []
	for name in KNOWN_STEPS:
	raw = run(f"tracecraft step-status {name}")
	# Format: "design: complete (agent: developer)"
	m = re.match(r"(\w+):\s+(\w+)(?:\s+\(agent:\s+(\w+)\))?", raw)
	if m:
	steps.append({
	"name": m.group(1),
	"status": m.group(2),
	"agent": m.group(3) or "",
	})
	else:
	steps.append({"name": name, "status": "unknown", "agent": ""})
	return {"steps": steps}


	@app.get("/api/artifacts")
	def get_artifacts():
	"""Parse `tracecraft artifact list` — one path per line."""
	raw = run("tracecraft artifact list")
	artifacts = []
	for line in raw.splitlines():
	line = line.strip()
	if not line:
	continue
	parts = line.split("/")
	# artifacts/<step>/<filename>
	step = parts[1] if len(parts) > 1 else ""
	name = parts[-1]
	artifacts.append({"file": name, "step": step, "path": line})
	return {"artifacts": artifacts}


	# ── Serve the frontend ───────────────────────────────────────────────────


	@app.get("/messages.json")
	def serve_messages_json():
	p = Path(__file__).parent / "messages.json"
	if p.exists():
	return FileResponse(p, media_type="application/json")
	return {"messages": []}


	@app.get("/")
	def serve_index():
	return FileResponse(Path(__file__).parent / "index.html")

Xet Storage Details

Size:: 4.55 kB
Xet hash:: e0223eb0c00163a2f1257a8897625eab8f687ba7039167d7dff68d3613996515

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.