platform/harness/workspace.py

"""harness/workspace.py — the unified WORKSPACE object model (owner directive 2026-07-12:
"every single time we create a module or analytics from the AIOS, it should be modular — a
workspace").

Every artifact AIOS creates — a view, a dashboard, an alert, a scheduled report — is a MODULAR,
self-contained workspace object: it carries everything needed to re-run itself (the governed
query / metric reference), renders through the house primitives, composes into bigger objects
(views → dashboards → schedules), and exports as a TENANT-AGNOSTIC template (queries reference
topic/metric KEYS; scope lives in the model, so a template stamps onto any tenant whose model
has those keys — the same principle that made model/ itself the template library).

This module is the facade: one uniform envelope + one listing/manage/export surface over the
underlying stores (views.json, routines.json). It does not replace them — it makes them one
workspace.
"""
import json
import time
from pathlib import Path

import harness.views as V
import harness.routines as R
import harness.semantic as SEM

TEMPLATES_DIR = Path(__file__).resolve().parents[1] / "model" / "templates"

# query keys that are tenant/context-specific and must NOT travel in a template
_STRIP = ("team_id", "filters", "date_from", "date_to")


def items():
    """Every workspace object, uniform envelope: {kind, name, detail, created_at}."""
    out = []
    vs, ds = V.views(), V.dashboards()
    dash_members = {vn for d in ds.values() for vn in d.get("views", [])}
    for name, v in vs.items():
        spec = v.get("chart") or {}
        q = spec.get("query") or {}
        out.append({"kind": "view", "name": name,
                    "detail": f"{q.get('topic', '?')} · {', '.join(q.get('measures') or [])}"
                              + (" · in a dashboard" if name in dash_members else ""),
                    "created_at": v.get("saved_at", "")})
    for name, d in ds.items():
        out.append({"kind": "dashboard", "name": name,
                    "detail": f"{len(d.get('views', []))} view(s): "
                              + ", ".join(d.get("views", [])[:4]),
                    "created_at": d.get("created_at", "")})
    rt = R.load()
    for key, a in rt.get("alerts", {}).items():
        out.append({"kind": "alert", "name": key,
                    "detail": f"{a.get('metric')} · {a.get('mode')}"
                              + (f" · team {a['team_id']}" if a.get("team_id") else ""),
                    "created_at": a.get("created_at", "")})
    for key, r in rt.get("reports", {}).items():
        out.append({"kind": "report", "name": key,
                    "detail": f"{r.get('cadence')} · {r.get('channel')}",
                    "created_at": r.get("created_at", "")})
    for key, i in (rt.get("insights") or {}).items():
        out.append({"kind": "insight", "name": key,
                    "detail": f"{i.get('cadence')} · \"{(i.get('prompt') or '')[:60]}\"",
                    "created_at": i.get("created_at", "")})
    return sorted(out, key=lambda x: x.get("created_at") or "", reverse=True)


def delete(kind, name):
    if kind == "view":
        V.delete_view(name)
    elif kind == "dashboard":
        V.delete_dashboard(name)
    elif kind == "alert":
        R.delete_alert(name)
    elif kind == "report":
        R.delete_report(name)
    elif kind == "insight":
        R.delete_insight(name)
    else:
        raise SEM.ModelError(f"unknown workspace kind {kind!r}")


# ------------------------------------------------------------------ templates (tenant-agnostic)

def _templatize_view(name):
    v = V.views().get(name)
    if not v:
        raise SEM.ModelError(f"unknown view {name!r}")
    spec = dict(v.get("chart") or {})
    q = dict(spec.get("query") or {})
    if not q:
        raise SEM.ModelError(f"view {name!r} carries no query — cannot templatize")
    stripped = {k: q.pop(k) for k in _STRIP if k in q}
    stripped = {k: v for k, v in stripped.items() if v}      # record only real tenant baggage
    q["window"] = "ytd"                      # instantiation default; tenant picks at stamp time
    spec["query"] = q
    return {"name": name, "spec": spec,
            "stripped": sorted(stripped)}    # honest record of what was tenant-specific


def export_template(kind, name):
    """Export a view or dashboard as a tenant-agnostic template JSON in model/templates/.
    Dashboards embed their views' templates — one self-contained modular unit."""
    if kind == "view":
        tpl = {"template": "view", "exported_at": time.strftime("%Y-%m-%d %H:%M"),
               **_templatize_view(name)}
    elif kind == "dashboard":
        d = V.dashboards().get(name)
        if d is None:
            raise SEM.ModelError(f"unknown dashboard {name!r}")
        tpl = {"template": "dashboard", "name": name,
               "exported_at": time.strftime("%Y-%m-%d %H:%M"),
               "views": [_templatize_view(vn) for vn in d.get("views", [])]}
    else:
        raise SEM.ModelError("only views and dashboards templatize (alerts/reports are one-liners "
                             "— recreate from the metric key)")
    TEMPLATES_DIR.mkdir(parents=True, exist_ok=True)
    safe = "".join(c if c.isalnum() or c in "-_ " else "_" for c in name).strip().replace(" ", "-")
    path = TEMPLATES_DIR / f"{kind}-{safe}.json"
    path.write_text(json.dumps(tpl, indent=1), encoding="utf-8")
    return str(path)


PROPOSALS_DIR = Path(__file__).resolve().parents[1] / "model" / "proposals"


def promote(kind, name, note=""):
    """Nominate a workspace object for PROMOTION into the governed model (principle 3: model
    just-in-time, promote what's proven). v0: writes a reviewable proposal file under
    model/proposals/ — the next Claude session is the Modeler: it reviews the proposal, lands
    the actual model diff (a metric, a skill recipe, or a curated template), and deletes the
    proposal. Git is the branch mode; nothing merges without that review."""
    if kind == "view":
        v = V.views().get(name)
        if not v:
            raise SEM.ModelError(f"unknown view {name!r}")
        spec = v.get("chart") or {}
        payload = {"kind": "view", "name": name, "spec": spec}
        q = spec.get("query") or {}
        hint = (f"candidate: a named metric or skill recipe on topic {q.get('topic')!r} "
                f"(measures {q.get('measures')}, group_by {q.get('group_by')})")
    elif kind == "dashboard":
        d = V.dashboards().get(name)
        if d is None:
            raise SEM.ModelError(f"unknown dashboard {name!r}")
        payload = {"kind": "dashboard", "name": name, "views": d.get("views", []),
                   "view_specs": {vn: (V.views().get(vn) or {}).get("chart")
                                  for vn in d.get("views", [])}}
        hint = "candidate: a curated template in model/templates/ or a skill recipe"
    else:
        raise SEM.ModelError("only views and dashboards can be nominated")
    try:
        import harness.telemetry as TEL
        usage = TEL.summary()
    except Exception:
        usage = {}
    PROPOSALS_DIR.mkdir(parents=True, exist_ok=True)
    safe = "".join(c if c.isalnum() or c in "-_ " else "_" for c in name).strip().replace(" ", "-")
    path = PROPOSALS_DIR / f"{time.strftime('%Y%m%d-%H%M')}-{kind}-{safe}.json"
    path.write_text(json.dumps({
        "proposed_at": time.strftime("%Y-%m-%d %H:%M"),
        "note": note, "suggested": hint, "usage_at_nomination": usage, **payload,
        "reviewer_instructions": (
            "MODELER REVIEW (next Claude session): decide metric / skill-recipe / template / "
            "reject; land the model diff in model/*.yml with a validate contract where "
            "applicable; re-run the eval gate; DELETE this proposal file; log the decision."),
    }, indent=1), encoding="utf-8")
    return str(path)


def proposals():
    if not PROPOSALS_DIR.exists():
        return []
    return sorted(p.name for p in PROPOSALS_DIR.glob("*.json"))


def templates():
    if not TEMPLATES_DIR.exists():
        return []
    return sorted(p.name for p in TEMPLATES_DIR.glob("*.json"))


def _ytd():
    from datetime import date
    t = date.today()
    return t.replace(month=1, day=1).isoformat(), t.isoformat()


def instantiate_template(filename, new_name=None):
    """Stamp a template into THIS tenant's workspace: window resolves to YTD now; tenant scope
    comes from the model (topics/tenant.yml), so no per-tenant edits are needed."""
    path = TEMPLATES_DIR / filename
    if not path.exists():
        raise SEM.ModelError(f"unknown template {filename!r}")
    tpl = json.loads(path.read_text(encoding="utf-8"))
    df, dt_ = _ytd()

    def _stamp_view(vtpl, vname):
        spec = dict(vtpl["spec"])
        q = dict(spec.get("query") or {})
        q.pop("window", None)
        q["date_from"], q["date_to"] = df, dt_
        spec["query"] = q
        spec["title"] = vname
        import harness.tools as T
        T.save_view(vname, {"chart": spec})
        return vname

    def _fresh(name):
        """Never clobber an existing object — suffix until free."""
        vs, ds = V.views(), V.dashboards()
        cand, i = name, 2
        while cand in vs or cand in ds:
            cand = f"{name} ({i})"
            i += 1
        return cand

    if tpl.get("template") == "view":
        vname = _fresh(new_name or tpl["name"])
        return {"created": "view", "name": _stamp_view(tpl, vname)}
    if tpl.get("template") == "dashboard":
        dname = _fresh(new_name or tpl["name"])
        # namespace member views under the new dashboard so instantiation NEVER overwrites
        # existing views (modularity means copies, not aliasing)
        vnames = [_stamp_view(vt, _fresh(f"{dname} · {vt['name']}"))
                  for vt in tpl.get("views", [])]
        import harness.tools as T
        T.compose_dashboard(dname, vnames)
        return {"created": "dashboard", "name": dname, "views": vnames}
    raise SEM.ModelError(f"bad template file {filename!r}")