| """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} |
|
|
|
|
| |
| 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 |
| 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 |
|
|
| |
| @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") |
|
|
| |
| @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)) |
|
|
| |
| @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)) |
|
|
| |
| @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) |
|
|
| |
| @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") |
|
|
| |
| @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) |
|
|
| |
| @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) |
|
|
| |
| @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) |
|
|
| |
| @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} |
|
|
| |
| @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"}) |
|
|
| |
| @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 |
|
|
