"""FastAPI application exposing the versioned section-14 APIs (/api/v1) and the dashboard (/). Run: uvicorn apps.api:create_app --factory --port 8000 """ from __future__ import annotations import threading from pathlib import Path from typing import Any, Optional from fastapi import Body, FastAPI, HTTPException, Query, Request from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, PlainTextResponse, Response from pydantic import BaseModel, Field from core.errors import ErrorCode, MatterOSError from core.representation import MatterStructure DASHBOARD = Path(__file__).resolve().parent / "dashboard" VIEWER_STATIC = Path(__file__).resolve().parent.parent / "visualization" / "static" STATUS = {ErrorCode.INVALID_INPUT: 400, ErrorCode.UNSUPPORTED_METHOD: 422, ErrorCode.RESOURCE_LIMIT: 429, ErrorCode.TIMEOUT: 504, ErrorCode.CHECKSUM_MISMATCH: 409, ErrorCode.CANCELLED: 409, ErrorCode.DATA_SOURCE_UNAVAILABLE: 503} # ---------------- request models ---------------- class ProjectIn(BaseModel): name: str description: str = "" goal: Optional[dict] = None constraints: Optional[dict] = None autonomy_mode: str = "bounded-autonomous" class MissionIn(BaseModel): project_id: str research_question: str spec: dict = Field(default_factory=dict) goal: Optional[dict] = None horizon: str = "short_experiment" budget: Optional[dict] = None autonomy_mode: Optional[str] = None class GenerateIn(BaseModel): project_id: str generator: str # random_cluster | known_family | symmetry_aware_crystal | graph_grammar | smiles | lj_random_cluster | motif params: dict = Field(default_factory=dict) mission_id: Optional[str] = None screen: bool = True constraints: dict = Field(default_factory=dict) class TransformIn(BaseModel): operation: str params: dict = Field(default_factory=dict) mission_id: Optional[str] = None class SubmitIn(BaseModel): candidate_id: str solver_id: str observable: str params: dict = Field(default_factory=dict) mission_id: Optional[str] = None priority: str = "normal" budget: Optional[dict] = None seed: int = 0 class MessageIn(BaseModel): sender: str recipient: str kind: str = "info" body: dict = Field(default_factory=dict) mission_id: Optional[str] = None refs: list[str] = Field(default_factory=list) class EventIn(BaseModel): subject: str predicate: str object: Any = None actor: str = "owner" evidence_refs: list[str] = Field(default_factory=list) def create_app(home: Optional[str] = None, db_path: Optional[str] = None) -> FastAPI: from matter_os.app import MatterOS mos = MatterOS(db_path=db_path, home=home) api = FastAPI(title="MATTER OS API", version="1.0.0", description="Versioned typed interfaces (spec section 14). Evidence first.") api.state.mos = mos threads: dict[str, threading.Thread] = {} V = "/api/v1" @api.exception_handler(MatterOSError) async def _err(request: Request, exc: MatterOSError): return JSONResponse(status_code=STATUS.get(exc.code, 500), content=exc.to_dict()) def dump(x): if hasattr(x, "model_dump"): return x.model_dump(mode="json") if isinstance(x, list): return [dump(i) for i in x] return x # ---------------- dashboard ---------------- @api.get("/", response_class=HTMLResponse) def dashboard(): return (DASHBOARD / "index.html").read_text() @api.get("/static/viewer.js") def viewer_js(): return FileResponse(VIEWER_STATIC / "viewer.js", media_type="application/javascript") # ---------------- Project API ---------------- @api.post(f"{V}/projects") def create_project(p: ProjectIn): return dump(mos.create_project(p.name, p.description, p.goal, p.constraints, p.autonomy_mode)) @api.get(f"{V}/projects") def list_projects(include_archived: bool = False): return dump(mos.repo.list_projects(include_archived)) @api.put(f"{V}/projects/{{project_id}}/goal") def update_goal(project_id: str, goal: dict = Body(...)): return dump(mos.update_goal(project_id, goal)) @api.put(f"{V}/projects/{{project_id}}/constraints") def set_constraints(project_id: str, constraints: dict = Body(...)): return dump(mos.set_constraints(project_id, constraints)) @api.post(f"{V}/projects/{{project_id}}/archive") def archive_project(project_id: str, confirm: bool = False): return dump(mos.archive_project(project_id, confirm)) # ---------------- Mission API ---------------- @api.post(f"{V}/missions") def create_mission(m: MissionIn): return dump(mos.missions.create_mission(m.project_id, m.research_question, goal=m.goal, spec=m.spec, horizon=m.horizon, budget=m.budget, autonomy_mode=m.autonomy_mode)) @api.get(f"{V}/missions") def list_missions(project_id: Optional[str] = None): return dump(mos.repo.list_missions(project_id)) @api.get(f"{V}/missions/{{mission_id}}") def get_mission(mission_id: str): return {**mos.missions.status(mission_id), "mission": dump(mos.repo.get_mission(mission_id)), "running_thread": mission_id in threads and threads[mission_id].is_alive()} @api.post(f"{V}/missions/{{mission_id}}/plan") def plan_tasks(mission_id: str): return dump(mos.missions.plan_tasks(mission_id)) @api.post(f"{V}/missions/{{mission_id}}/run") def run_mission(mission_id: str, background: bool = True): if background: if mission_id in threads and threads[mission_id].is_alive(): return {"status": "already_running"} t = threading.Thread(target=lambda: mos.missions.run(mission_id), daemon=True) threads[mission_id] = t; t.start() return {"status": "started", "mission_id": mission_id} return dump(mos.missions.run(mission_id)) for action in ("pause", "resume", "approve"): def _mk(action=action): def handler(mission_id: str): return dump(getattr(mos.missions, action)(mission_id)) handler.__name__ = f"mission_{action}" return handler api.post(f"{V}/missions/{{mission_id}}/{action}")(_mk()) @api.post(f"{V}/missions/{{mission_id}}/cancel") def cancel(mission_id: str, confirm: bool = False): return dump(mos.missions.cancel(mission_id, confirm)) @api.post(f"{V}/missions/{{mission_id}}/reject") def reject(mission_id: str, reason: str = ""): return dump(mos.missions.reject(mission_id, reason)) @api.post(f"{V}/missions/{{mission_id}}/branch") def branch(mission_id: str, body: dict = Body(default={})): return dump(mos.missions.branch(mission_id, body.get("research_question"), body.get("spec_overrides"))) @api.post(f"{V}/missions/{{mission_id}}/budget") def change_budget(mission_id: str, budget: dict = Body(...)): return dump(mos.missions.change_budget(mission_id, budget)) # ---------------- Candidate API ---------------- @api.post(f"{V}/candidates/generate") def generate(g: GenerateIn): from discovery import generators as gen from core.io import from_smiles fns = {"random_cluster": gen.random_cluster, "known_family": gen.known_family, "symmetry_aware_crystal": gen.symmetry_aware_crystal, "lj_random_cluster": gen.lj_random_cluster, "motif": gen.cluster_motif, "smiles": from_smiles, "graph_grammar": lambda **kw: gen.graph_grammar_molecule(**kw)[0]} if g.generator == "learned_model": return gen.LearnedProposalModel().propose(1, 0) if g.generator not in fns: raise MatterOSError(ErrorCode.INVALID_INPUT, f"unknown generator; allowed {sorted(fns) + ['learned_model (stub)']}") ms = fns[g.generator](**g.params) c = mos.candidates.register(ms, g.project_id, g.generator, mission_id=g.mission_id, constraints=g.constraints, screen=g.screen) return dump(c) @api.get(f"{V}/candidates") def list_candidates(project_id: Optional[str] = None, mission_id: Optional[str] = None, maturity: Optional[str] = None): from storage.indexes import search_candidates return search_candidates(mos, project_id=project_id, maturity=maturity) if not mission_id else \ [r for r in search_candidates(mos, project_id=project_id, maturity=maturity) if r["mission_id"] == mission_id] @api.get(f"{V}/candidates/compare") def compare(a: str, b: str): return mos.candidates.compare(a, b) @api.get(f"{V}/candidates/{{candidate_id}}") def get_candidate(candidate_id: str): c = mos.repo.get_candidate(candidate_id) return {"candidate": dump(c), "unified": MatterStructure.from_candidate(c).unified_view(), "versions": [{"version": v["version"], "checksum": v["checksum"]} for v in mos.repo.candidate_versions(candidate_id)]} @api.post(f"{V}/candidates/{{candidate_id}}/transform") def transform(candidate_id: str, t: TransformIn): return dump(mos.candidates.transform(candidate_id, t.operation, t.params, t.mission_id)) @api.post(f"{V}/candidates/{{candidate_id}}/screen") def screen(candidate_id: str, constraints: dict = Body(default={})): return dump(mos.candidates.screen(candidate_id, constraints or None)) @api.post(f"{V}/candidates/validate_schema") def validate_schema(payload: dict = Body(...)): return mos.candidates.validate_schema(payload) @api.post(f"{V}/candidates/import") def import_structure(project_id: str, fmt: str, text: str = Body(..., media_type="text/plain")): return dump(mos.candidates.import_structure(text, fmt, project_id)) @api.get(f"{V}/candidates/{{candidate_id}}/export", response_class=PlainTextResponse) def export_candidate(candidate_id: str, fmt: str = "xyz"): return mos.candidates.export(candidate_id, fmt) # ---------------- Simulation API ---------------- @api.get(f"{V}/simulation/solvers") def solvers(): return mos.registry.describe() @api.get(f"{V}/simulation/compatible") def list_compatible_solvers(candidate_id: str, observable: str): return mos.simulation.list_compatible_solvers(candidate_id, observable) @api.get(f"{V}/simulation/plan") def plan(candidate_id: str, observable: str): return mos.router.plan(MatterStructure.from_candidate(mos.repo.get_candidate(candidate_id)), observable) @api.get(f"{V}/simulation/estimate") def estimate_cost(candidate_id: str, solver_id: str, observable: str): return mos.simulation.estimate_cost(candidate_id, solver_id, observable) @api.post(f"{V}/simulation/runs") def submit_run(s: SubmitIn): return dump(mos.simulation.submit_run(s.candidate_id, s.solver_id, s.observable, s.params, mission_id=s.mission_id, priority=s.priority, budget=s.budget, seed=s.seed)) @api.get(f"{V}/simulation/runs") def list_runs(candidate_id: Optional[str] = None, mission_id: Optional[str] = None, limit: int = 200): runs = mos.repo.list_runs(candidate_id=candidate_id, mission_id=mission_id)[-limit:] return [{"run_id": r.run_id, "candidate_id": r.candidate_id, "method_id": r.method_id, "exit_status": r.exit_status, "convergence_status": r.convergence_status, "error_code": r.error_code, "finished_at": r.timestamps.get("finished_at"), "wall_s": r.timestamps.get("wall_s"), "router_rationale": r.router_rationale, "observables": {k: v for k, v in r.observables.items() if k not in ("forces", "frequencies", "electronic_descriptors")}} for r in runs] @api.get(f"{V}/simulation/runs/{{run_id}}") def get_run(run_id: str): return dump(mos.repo.get_run(run_id)) @api.get(f"{V}/simulation/jobs") def list_jobs(status: Optional[str] = None, mission_id: Optional[str] = None): return mos.jobs.list(status, mission_id) @api.get(f"{V}/simulation/jobs/{{job_id}}") def get_status(job_id: str): return mos.simulation.get_status(job_id) @api.post(f"{V}/simulation/jobs/{{job_id}}/cancel") def cancel_job(job_id: str): mos.jobs.cancel(job_id); return mos.jobs.get(job_id) @api.get(f"{V}/simulation/runs/{{run_id}}/artifacts") def fetch_artifacts(run_id: str): return mos.simulation.fetch_artifacts(run_id) @api.get(f"{V}/artifacts/{{artifact_id}}") def get_artifact(artifact_id: str): return Response(mos.artifacts.get(artifact_id), media_type="text/plain") # ---------------- Agent API ---------------- @api.get(f"{V}/agents") def roster(): return mos.agents.roster() @api.post(f"{V}/agents/assign_task") def assign_task(role: str, mission_id: str, stage: str): return mos.agents.assign_task(role, mission_id, stage) @api.post(f"{V}/agents/messages") def send_message(m: MessageIn): return dump(mos.agents.send_message(m.sender, m.recipient, m.kind, m.body, m.mission_id, m.refs)) @api.get(f"{V}/agents/messages") def messages(mission_id: str): return mos.agents.bus.thread(mission_id) @api.post(f"{V}/agents/publish_result") def publish_result(role: str, mission_id: str, body: dict = Body(...), refs: list[str] = Query(default=[])): return mos.agents.publish_result(role, mission_id, body, refs) @api.post(f"{V}/agents/request_review") def request_review(requester: str, reviewer: str, mission_id: str, question: str, refs: list[str] = Query(default=[])): return dump(mos.agents.request_review(requester, reviewer, mission_id, refs, question)) @api.get(f"{V}/agents/evaluations") def evaluations(role: Optional[str] = None): from agents.evaluation import history return history(mos.db, role) @api.post(f"{V}/agents/benchmarks") def run_role_benchmarks(): from agents.evaluation import role_benchmarks return role_benchmarks(mos) @api.get(f"{V}/agents/improvements") def improvements(): return [dict(r, doc=mos.db.loads(r["doc"])) for r in mos.db.query("SELECT * FROM improvements ORDER BY created_at")] @api.post(f"{V}/agents/improvements/{{iid}}/deploy") def deploy(iid: str, approved_by_owner: bool = False): from agents.meta import deploy as _d return _d(mos, iid, approved_by_owner) @api.post(f"{V}/agents/improvements/{{iid}}/rollback") def rollback(iid: str): from agents.meta import rollback as _r return _r(mos, iid) # ---------------- Evidence API ---------------- @api.post(f"{V}/evidence/{{candidate_id}}/maturity") def compute_maturity(candidate_id: str): return mos.evidence.compute_maturity(candidate_id) @api.post(f"{V}/evidence/{{candidate_id}}/claims") def attach_claim(candidate_id: str, claim_text: str = Body(..., embed=True), laboratory_evidence: Optional[dict] = Body(default=None, embed=True)): from core.schemas import LaboratoryEvidence lab = LaboratoryEvidence(**laboratory_evidence) if laboratory_evidence else None return dump(mos.evidence.attach_claim(candidate_id, claim_text, lab)) @api.get(f"{V}/evidence/claims") def claims(candidate_id: Optional[str] = None): return dump(mos.repo.list_claims(candidate_id)) @api.post(f"{V}/evidence/{{candidate_id}}/replication") def request_replication(candidate_id: str, run_id: str, seed: int = 1): return mos.evidence.request_replication(candidate_id, run_id, seed=seed) @api.post(f"{V}/evidence/compare_sources") def compare_sources(values: list[dict] = Body(...)): return mos.evidence.compare_sources(values) @api.post(f"{V}/evidence/benchmarks") def benchmarks(project_id: str, ids: list[str] = Query(default=[])): from simulation.benchmarks import run_all return run_all(mos, project_id, ids or None) # ---------------- Knowledge API ---------------- @api.post(f"{V}/knowledge/events") def propose_event(e: EventIn): return mos.knowledge.propose_event(e.subject, e.predicate, e.object, e.actor, e.evidence_refs) @api.post(f"{V}/knowledge/events/{{fact_id}}/verify") def verify_event(fact_id: str, actor: str = "data_steward", evidence_refs: list[str] = Query(default=[])): return mos.knowledge.verify_event(fact_id, actor, evidence_refs) @api.get(f"{V}/knowledge/events/{{fact_id}}") def fact_history(fact_id: str): return mos.knowledge.kg.history(fact_id) @api.get(f"{V}/knowledge/search") def search_graph(text: str = "", status: Optional[str] = None, etype: Optional[str] = None): return mos.knowledge.search_graph(text, status, etype) @api.get(f"{V}/knowledge/lineage/{{candidate_id}}") def get_lineage(candidate_id: str): return mos.knowledge.get_lineage(candidate_id) @api.get(f"{V}/knowledge/status_counts") def kg_counts(): return mos.knowledge.kg.status_counts() @api.get(f"{V}/knowledge/sources") def sources(): from knowledge.provenance import list_sources return list_sources(mos.db) # ---------------- Resource API ---------------- @api.get(f"{V}/resources/capacity") def get_capacity(): return mos.resources.get_capacity() @api.post(f"{V}/resources/allocate") def allocate(job_id: str, priority: str = "normal", request: dict = Body(default={})): a = mos.resources.allocate(job_id, priority, request) return {"allocation": a, "granted": a is not None} @api.post(f"{V}/resources/release") def release(job_id: str): mos.resources.release(job_id); return {"released": job_id} @api.post(f"{V}/resources/checkpoint/{{target_id}}") def checkpoint(target_id: str): from orchestration.checkpoints import checkpoint as ck return ck(mos, target_id) @api.post(f"{V}/resources/enforce_quota") def enforce_quota(request: dict = Body(...)): return mos.resources.enforce_quota(request) @api.post(f"{V}/resources/snapshot") def snapshot(): from storage.archive import snapshot as snap return {"snapshot": snap(mos).name} @api.get(f"{V}/resources/integrity") def integrity(): from storage.archive import verify_integrity return verify_integrity(mos) @api.get(f"{V}/resources/queue") def queue(): from orchestration.queues import queue_snapshot return queue_snapshot(mos) @api.post(f"{V}/control/pause_all") def pause_all(paused: bool = True): mos.paused_global = paused mos.db.audit("owner", "pause_all" if paused else "resume_all", "*", {}) return {"paused_global": paused} # ---------------- Export API ---------------- @api.get(f"{V}/export/structure/{{candidate_id}}", response_class=PlainTextResponse) def export_structure(candidate_id: str, fmt: str = "xyz"): return mos.candidates.export(candidate_id, fmt) @api.get(f"{V}/export/run/{{run_id}}") def export_run_bundle(run_id: str): from storage.archive import export_run_bundle as e return Response(e(mos, run_id), media_type="application/gzip", headers={"Content-Disposition": f"attachment; filename={run_id}.tar.gz"}) @api.get(f"{V}/export/report/{{mission_id}}") def export_report(mission_id: str, fmt: str = "md"): from storage.archive import export_report as e return Response(e(mos, mission_id, fmt), media_type="text/markdown" if fmt == "md" else "application/json") @api.get(f"{V}/export/project/{{project_id}}") def export_project(project_id: str): from storage.archive import export_project as e return Response(e(mos, project_id), media_type="application/gzip", headers={"Content-Disposition": f"attachment; filename={project_id}.tar.gz"}) # ---------------- Observability / dashboard data ---------------- @api.get(f"{V}/observability/audit") def audit(limit: int = 200, actor: Optional[str] = None): if actor: rows = mos.db.query("SELECT * FROM audit_log WHERE actor=? ORDER BY id DESC LIMIT ?", (actor, limit)) else: rows = mos.db.query("SELECT * FROM audit_log ORDER BY id DESC LIMIT ?", (limit,)) return [dict(r, details=mos.db.loads(r["details"])) for r in rows] @api.get(f"{V}/observability/metrics") def metrics(limit: int = 500): return mos.db.query("SELECT * FROM metrics ORDER BY id DESC LIMIT ?", (limit,)) @api.get(f"{V}/observability/failures") def failures(mission_id: Optional[str] = None): rows = mos.db.query("SELECT * FROM failures" + (" WHERE mission_id=?" if mission_id else "") + " ORDER BY created_at DESC LIMIT 200", (mission_id,) if mission_id else ()) return [dict(r, doc=mos.db.loads(r["doc"])) for r in rows] @api.get(f"{V}/viewer/{{candidate_id}}") def viewer_data(candidate_id: str, run_id: Optional[str] = None): from visualization.viewer import structure_payload return structure_payload(mos, candidate_id, run_id) @api.get(f"{V}/dashboard/summary") def summary(): from visualization.dashboards import summary as s return s(mos) return api