native_apps.py

#!/usr/bin/env python3
"""Generate every place that links OctoSense's native apps (Python 3.9+).

Native apps (Rust crates the shell links: App Hub, Rinx, Terminal, and the
opt-in Sheets, Reference, AppCard) are declared once, in `native-apps.json`
at the repository root (ADR 0004 §1). This script writes what follows from
it, each between `# BEGIN native-apps: <block>` and `# END native-apps:
<block>` markers:

  Cargo.toml             the app crates' pins in [workspace.dependencies]
                         ("pins") and the [patch] entries that resolve a
                         git source to its `local` checkout ("patch <git>")
  crates/shell/Cargo.toml the optional dependencies ("deps"), the ones
                         native mobile targets link without flags ("mobile
                         deps") and the `app-<id>` / `mobile-apps` features
                         ("features")
  desktop/Cargo.toml,    the package's default features and the forwarded
  phone/Cargo.toml       `app-<id>` / `mobile-apps` features ("features")

the whole of `crates/shell/src/native_apps.rs`: what `linked_modules()`
pushes under which feature, each app's hosting per target (the default the
shell uses when the person has not switched an app) and its agent block
(its own tools, the kernel tools and other apps' tools its agent is
granted, its tool-call budget); and the whole of
`crates/ai-host/src/native_agents.rs`: the `octos.*` services each app's
agent is granted, which `Policy::shipped()` hands out.

It refuses `process` hosting on mobile or wasm targets, plain `process` on
Linux (non-Vulkan Linux runs every app in-process: `process-if-vulkan`),
`process` without a `bin`, and two revisions of one git repository in the
root pins. Apps released by tag (Rinx, hagency-org/Rinx#37) must be pinned by
a release tag (`vX.Y.Z` or `vX.Y.Z-rc.N`) together with the commit it names:
the root pin says `tag = …`, and --check also reads Cargo.lock and refuses a
lock whose commit for that tag is not the manifest's `rev` (a moved tag). It
also holds the hand-kept desktop catalog (`desktop/config/apps.json`) to the
manifest: a native app's row must name its `crate` and `bin` and the source
the manifest says (a process launch builds those, never the row's), and an
app that runs as a process on a desktop must have a row. After a write, Cargo.lock is brought in step by cargo itself
(`cargo update -p` for a pin that changed, then `cargo metadata`); --no-lock
skips that.

  python3 tools/native_apps.py            # regenerate after editing the manifest
  python3 tools/native_apps.py --check    # CI: exit 1 when anything drifted
"""
import argparse
import json
from pathlib import Path
import re
import subprocess
import sys

ROOT = Path(__file__).resolve().parents[1]
MANIFEST = "native-apps.json"
RUST_FILE = "crates/shell/src/native_apps.rs"
# Apps OctoSense takes only as tagged releases (hagency-org/Rinx#37): never a
# raw commit. main may take a release candidate; OctoSense's own releases take
# only a final release (checked by their release workflows, not here).
TAGGED_RELEASES = {"rinx"}
RELEASE_TAG = r"v\d+\.\d+\.\d+(-rc\.\d+)?"
AGENTS_FILE = "crates/ai-host/src/native_agents.rs"
PACKAGES = {"desktop": "desktop/Cargo.toml", "phone": "phone/Cargo.toml"}

TARGETS = ("macos", "windows", "linux", "android", "ios", "ohos")
# Targets without child processes (`host::processes_available()` is false).
MODULE_ONLY = ("android", "ios", "ohos", "wasm")
HOSTINGS = ("module", "process", "process-if-vulkan", "none")
# How a package links an app: in its default features, in its `mobile-apps`
# set, only when asked for, or not at all.
SHELL_LINKS = ("default", "mobile-apps", "opt-in", "off")
# Every standard shell build runs the octos kernel as a shell service, and
# the `wasm` service for apps' own WebAssembly functions (ADR 0011; the shell
# links it on macOS, Linux and Android only).
BASE_DEFAULT = ["octos-core", "wasm-functions"]
# Features on by default in one shell only: the phone offers app agents the
# system toolbox (ADR 0002 §6), reading pages in its own WebView.
SHELL_BASE_DEFAULT = {"phone": ["toolbox-peers"], "desktop": ["craft-engines"]}
APP_KEYS = {"id", "feature", "crate", "source", "module", "bin", "bin_features", "default_features", "crate_features",
            "implies", "hosting", "shells", "native_mobile", "sandbox", "storage", "agent", "kernel"}
REQUIRED_KEYS = APP_KEYS - {"feature", "bin_features", "kernel"}
# `kernel`: the port to the shell's octos kernel an app that is itself an
# octos client gets (ADR 0003, "An app that is an octos client"): `coding`,
# the coding scope the shell's kernel router enforces. Without it, a port the
# app opens is closed.
KERNEL_PORTS = ("coding",)
# The workspace member a process launch selects with the app's crate, so the
# build gets `bin_features` and the workspace lock (clients.rs `launch_plan`).
PROCESS_APPS = "crates/process-apps/Cargo.toml"
NOTE = "(generated by tools/native_apps.py from native-apps.json; do not edit)"


class ManifestError(Exception):
    pass


def note(message):
    print(f"native-apps: {message}", file=sys.stderr)


# --- The manifest ----------------------------------------------------------


def load(root):
    path = Path(root) / MANIFEST
    try:
        data = json.loads(path.read_text())
    except (OSError, ValueError) as error:
        raise ManifestError(f"{MANIFEST}: {error}")
    return validate(data)


def validate(data):
    """The apps, checked; every problem is reported at once."""
    problems = []
    if not isinstance(data, dict) or data.get("schema") != 1 or not isinstance(data.get("apps"), list):
        raise ManifestError(f'{MANIFEST}: expected {{"schema": 1, "apps": [...]}}')
    seen, features, crates = set(), set(), set()
    for index, app in enumerate(data["apps"]):
        where = f"apps[{index}]"
        if not isinstance(app, dict):
            problems.append(f"{where}: not an object")
            continue
        where = f"{app.get('id', where)}"
        missing = sorted(REQUIRED_KEYS - set(app))
        unknown = sorted(set(app) - APP_KEYS)
        if missing:
            problems.append(f"{where}: missing {', '.join(missing)}")
        if unknown:
            problems.append(f"{where}: unknown {', '.join(unknown)}")
        if missing:
            continue
        ident = app["id"]
        if not isinstance(ident, str) or not re.fullmatch(r"[a-z][a-z0-9-]*", ident):
            problems.append(f"{where}: id must be lowercase letters, digits and '-'")
        if ident in seen:
            problems.append(f"{where}: duplicate id")
        seen.add(ident)
        feature = feature_of(app)
        if feature in features:
            problems.append(f"{where}: duplicate feature {feature}")
        features.add(feature)
        if app["crate"] in crates:
            problems.append(f"{where}: duplicate crate {app['crate']}")
        crates.add(app["crate"])
        source = app["source"]
        if not isinstance(source, dict):
            problems.append(f"{where}: source must be an object")
        elif "path" in source:
            if set(source) != {"path"}:
                problems.append(f"{where}: a path source has only 'path'")
        elif set(source) not in ({"git", "rev", "local"}, {"git", "tag", "rev", "local"}):
            problems.append(f"{where}: source is {{git, rev, local}}, {{git, tag, rev, local}} or {{path}}")
        elif not re.fullmatch(r"[0-9a-f]{40}", str(source["rev"])):
            problems.append(f"{where}: source.rev must be a full commit id (for a tag: the commit it names)")
        elif "tag" in source and not re.fullmatch(RELEASE_TAG, str(source["tag"])):
            problems.append(f"{where}: source.tag must be a release tag vX.Y.Z or vX.Y.Z-rc.N")
        if isinstance(source, dict) and ident in TAGGED_RELEASES and "tag" not in source:
            problems.append(f"{where}: is taken only as a tagged release (hagency-org/Rinx#37): pin source.tag, not a raw rev")
        process_only = app["module"] is None
        if not process_only and (not modules_of(app) or not all(re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*(::[A-Za-z_][A-Za-z0-9_]*)+", m) for m in modules_of(app))):
            problems.append(f"{where}: module must be a Rust path (or a list of them), or null for a process-only app")
        if process_only:
            # No module: the app runs only as its own desktop process (ADR
            # 0004 §2: a browser whose engine owns its process, a task
            # manager). Nothing links it, so no shell names it.
            if not app["bin"]:
                problems.append(f"{where}: a process-only app (module null) needs a bin")
            if app["shells"] != {"desktop": "off", "phone": "off"} if isinstance(app["shells"], dict) else True:
                problems.append(f"{where}: a process-only app is linked by no shell: shells must be off")
        if app["bin"] is not None and not (isinstance(app["bin"], str) and app["bin"]):
            problems.append(f"{where}: bin must be a binary name or null")
        bin_features = app.get("bin_features", [])
        if not isinstance(bin_features, list) or not all(isinstance(x, str) for x in bin_features):
            problems.append(f"{where}: bin_features must be a list of strings")
        elif bin_features and not app["bin"]:
            problems.append(f"{where}: bin_features without a bin")
        if not isinstance(app["default_features"], bool):
            problems.append(f"{where}: default_features must be true or false")
        for key in ("crate_features", "implies"):
            if not isinstance(app[key], list) or not all(isinstance(x, str) for x in app[key]):
                problems.append(f"{where}: {key} must be a list of strings")
        hosting = app["hosting"]
        if not isinstance(hosting, dict):
            problems.append(f"{where}: hosting must be an object")
        else:
            for target in TARGETS:
                if target not in hosting:
                    problems.append(f"{where}: hosting.{target} is missing")
            for target, value in hosting.items():
                if target not in TARGETS + ("wasm",):
                    problems.append(f"{where}: hosting.{target}: unknown target")
                elif value not in HOSTINGS:
                    problems.append(f"{where}: hosting.{target}: {value!r} is not one of {', '.join(HOSTINGS)}")
                elif value == "none" and app["module"] is not None:
                    problems.append(f"{where}: hosting.{target}: 'none' is for a process-only app (module null)")
                elif value == "module" and app["module"] is None:
                    problems.append(f"{where}: hosting.{target}: a process-only app (module null) has no module")
                elif value not in ("module", "none") and target in MODULE_ONLY:
                    problems.append(f"{where}: hosting.{target}: {target} has no processes; only 'module' (or 'none' for a process-only app)")
                elif value == "process" and target == "linux":
                    problems.append(f"{where}: hosting.linux: 'process' would run without Vulkan+Wayland; "
                                    "use 'process-if-vulkan' (ADR 0004 §2)")
                elif value in ("process", "process-if-vulkan") and not app["bin"]:
                    problems.append(f"{where}: hosting.{target}: '{value}' needs a bin")
        shells = app["shells"]
        if not isinstance(shells, dict) or set(shells) != set(PACKAGES):
            problems.append(f"{where}: shells must name {', '.join(PACKAGES)}")
        else:
            for shell, value in shells.items():
                if value not in SHELL_LINKS:
                    problems.append(f"{where}: shells.{shell}: {value!r} is not one of {', '.join(SHELL_LINKS)}")
        if app["native_mobile"] not in ("always", "feature"):
            problems.append(f"{where}: native_mobile must be 'always' or 'feature'")
        for key in ("sandbox", "storage", "agent"):
            if not isinstance(app[key], dict):
                problems.append(f"{where}: {key} must be an object")
        if isinstance(app["sandbox"], dict):
            problems += [f"{where}: {p}" for p in sandbox_problems(app["sandbox"])]
        if isinstance(app["storage"], dict):
            problems += [f"{where}: {p}" for p in storage_problems(app["storage"])]
        if "kernel" in app and app["kernel"] not in KERNEL_PORTS:
            problems.append(f"{where}: kernel must be one of {', '.join(KERNEL_PORTS)}")
        if isinstance(app["agent"], dict) and not isinstance(app["agent"].get("octos"), list):
            problems.append(f"{where}: agent.octos must be a list")
        if isinstance(app["agent"], dict):
            problems += [f"{where}: {p}" for p in tool_policy_problems(app["agent"].get("tool_policy", {}))]
            problems += [f"{where}: {p}" for p in agent_problems(ident, app["agent"])]
    problems += grant_problems([a for a in data["apps"] if isinstance(a, dict) and isinstance(a.get("agent"), dict)])
    if problems:
        raise ManifestError(f"{MANIFEST}:\n  " + "\n  ".join(problems))
    return data["apps"]


NETWORKS = ("none", "any")
EXTERNAL_ROOTS = ("home", "documents", "downloads", "desktop", "pictures", "music", "movies", "tmp")


def sandbox_problems(sandbox):
    """`sandbox`: {"network": "none"|"any", "processes": bool} (ADR 0004 §3):
    what a process-hosted instance's OS sandbox allows beyond its files."""
    problems = []
    if set(sandbox) != {"network", "processes"}:
        problems.append("sandbox needs exactly network and processes")
    if sandbox.get("network") not in NETWORKS:
        problems.append(f"sandbox.network must be one of {', '.join(NETWORKS)}")
    if not isinstance(sandbox.get("processes"), bool):
        problems.append("sandbox.processes must be true or false")
    return problems


STORAGE_KEYS = ("max_bytes", "accounts", "agent_workspace", "cache_max_bytes", "external")


def storage_problems(storage):
    """`storage` (ADR 0004 §11), the block the shell parses with
    `app_storage::StorageSpec` at startup: `accounts` a bool, `agent_workspace`
    "account" or "none", the optional byte ceilings positive whole numbers,
    `external` a list of `<root>[/<path>]:ro|rw` grants outside the jail
    (part of the sandbox, never of an agent's workspace). Unknown fields are
    refused, as the shell refuses them."""
    problems = []
    unknown = sorted(set(storage) - set(STORAGE_KEYS))
    if unknown:
        problems.append(f"storage.{unknown[0]} is not a storage field (one of {', '.join(STORAGE_KEYS)})")
    if not isinstance(storage.get("accounts"), bool):
        problems.append("storage.accounts must be true or false")
    if storage.get("agent_workspace", "account") not in ("account", "none"):
        problems.append("storage.agent_workspace must be 'account' or 'none'")
    for key in ("max_bytes", "cache_max_bytes"):
        value = storage.get(key)
        if value is not None and (isinstance(value, bool) or not isinstance(value, int) or value <= 0):
            problems.append(f"storage.{key} must be a positive whole number of bytes")
    external = storage.get("external")
    if not isinstance(external, list):
        return problems + ["storage.external must be a list"]
    for grant in external:
        match = isinstance(grant, str) and re.fullmatch(r"([a-z]+)(/[A-Za-z0-9._/-]+)?:(ro|rw)", grant)
        if not match or match.group(1) not in EXTERNAL_ROOTS or ".." in grant:
            problems.append(f"storage.external: {grant!r} is not <root>[/<path>]:ro|rw with root one of {', '.join(EXTERNAL_ROOTS)}")
    return problems


def tool_policy_problems(policy):
    """`agent.tool_policy`: {tool: {"confirm": "host"|"app", "auto_approvable": bool}}
    (ADR 0004 §8). A tool the shell confirms live and no standing rule may
    approve is `confirm: host`, `auto_approvable: false`."""
    if not isinstance(policy, dict):
        return ["agent.tool_policy must be an object"]
    problems = []
    for tool, rule in policy.items():
        if not re.fullmatch(r"[a-z][a-z0-9_]*", tool):
            problems.append(f"agent.tool_policy: {tool!r} is not a tool name")
        elif not isinstance(rule, dict) or set(rule) != {"confirm", "auto_approvable"}:
            problems.append(f"agent.tool_policy.{tool}: needs exactly confirm and auto_approvable")
        elif rule["confirm"] not in ("host", "app"):
            problems.append(f"agent.tool_policy.{tool}.confirm must be 'host' or 'app'")
        elif not isinstance(rule["auto_approvable"], bool):
            problems.append(f"agent.tool_policy.{tool}.auto_approvable must be true or false")
    return problems


AGENT_KEYS = {"octos", "tools", "generic_tools", "grants", "tool_policy", "budget", "system_tools", "own_tools"}
# The fields a `tools.json` entry may carry (octos `ToolDecl`, UPCR-2026-035);
# `app` is the shell's to set.
DECL_FIELDS = {"name", "description", "input_schema", "output_schema", "risk", "background", "outward", "confirm", "shareable"}
RISKS = ("read", "act", "destructive")
# octos's own shell: never granted to an app agent (ADR 0004 §12); command
# execution is a host tool (`terminal.run`) with a live approval.
OCTOS_SHELL = ("shell", "bash", "exec_command", "write_stdin", "group:runtime")


def agent_problems(ident, agent):
    """The agent block (ADR 0004 §1, §7, §12): `octos` (its `octos.*`
    services), `tools` (null, or its own `tools.json` entries, named
    `<id>.<tool>`), `generic_tools` (the exact octos kernel tools its agent
    gets; never octos's shell), `grants` (other apps' shareable tools,
    `{"app", "tool"}`), `budget` (`calls_per_turn`, `calls_per_day`),
    `system_tools` (its own tools the system agent may call while the app
    runs: each one of its `tools`, a shareable read tool, by full name),
    `own_tools` (the tools its own agent may call, by full name: each one
    of its `tools`; without it, all of them)."""
    problems = []
    unknown = sorted(set(agent) - AGENT_KEYS)
    if unknown:
        problems.append(f"agent: unknown {', '.join(unknown)}")
    tools = agent.get("tools")
    names = []
    if tools is not None:
        if not isinstance(tools, list) or not tools:
            problems.append("agent.tools must be null or a non-empty list of tools.json entries")
            tools = []
        for i, tool in enumerate(tools):
            where = f"agent.tools[{i}]"
            if not isinstance(tool, dict):
                problems.append(f"{where}: not an object")
                continue
            extra = sorted(set(tool) - DECL_FIELDS)
            if extra:
                problems.append(f"{where}: unknown {', '.join(extra)} (the kernel refuses them)")
            name = tool.get("name")
            if not isinstance(name, str) or not re.fullmatch(rf"{re.escape(ident)}(\.[a-z][a-z0-9_]*)+", name):
                problems.append(f"{where}: name must be {ident}.<tool>")
            elif name in names:
                problems.append(f"{where}: {name} is declared twice")
            else:
                names.append(name)
            if not isinstance(tool.get("description"), str) or not tool["description"].strip():
                problems.append(f"{where}: description is required")
            schema = tool.get("input_schema")
            if not isinstance(schema, dict) or schema.get("type") != "object":
                problems.append(f"{where}: input_schema must describe an object")
            if "output_schema" in tool and not isinstance(tool["output_schema"], dict):
                problems.append(f"{where}: output_schema must be a schema object")
            if tool.get("risk") not in RISKS:
                problems.append(f"{where}: risk must be one of {', '.join(RISKS)}")
            if tool.get("confirm", "host") not in ("host", "app"):
                problems.append(f"{where}: confirm must be 'host' or 'app'")
            for flag in ("shareable", "background", "outward"):
                if flag in tool and not isinstance(tool[flag], bool):
                    problems.append(f"{where}: {flag} must be true or false")
    policy = agent.get("tool_policy", {})
    if isinstance(policy, dict) and tools is not None:
        short = {n.split(".", 1)[1] for n in names}
        for rule in policy:
            if rule not in short:
                problems.append(f"agent.tool_policy.{rule}: {ident} declares no tool {ident}.{rule}")
    generic = agent.get("generic_tools", [])
    if not isinstance(generic, list):
        problems.append("agent.generic_tools must be a list of octos tool names")
        generic = []
    for g in generic:
        if g in OCTOS_SHELL:
            problems.append(f"agent.generic_tools: {g} is octos's own shell, which no app agent gets "
                            "(ADR 0004 §12: command execution is a host tool)")
        elif not (isinstance(g, str) and re.fullmatch(r"[A-Za-z0-9_-]{1,64}", g)):
            problems.append(f"agent.generic_tools: {g!r} is not an octos tool name")
    if len(set(map(str, generic))) != len(generic):
        problems.append("agent.generic_tools names a tool twice")
    system = agent.get("system_tools", [])
    declared = {t.get("name"): t for t in (tools or []) if isinstance(t, dict)}
    if not isinstance(system, list) or not all(isinstance(t, str) for t in system):
        problems.append("agent.system_tools must be a list of its own tool names")
        system = []
    for name in system:
        tool = declared.get(name)
        if tool is None:
            problems.append(f"agent.system_tools: {name} is not one of {ident}'s agent.tools")
        elif tool.get("risk") != "read" or tool.get("shareable") is not True:
            problems.append(f"agent.system_tools: {name} must be a shareable read tool (the system agent gets read tools only)")
    if len(set(system)) != len(system):
        problems.append("agent.system_tools names a tool twice")
    own = agent.get("own_tools")
    if own is not None:
        if not isinstance(own, list) or not all(isinstance(t, str) for t in own):
            problems.append("agent.own_tools must be a list of its own tool names")
            own = []
        for name in own:
            if name not in declared:
                problems.append(f"agent.own_tools: {name} is not one of {ident}'s agent.tools")
        if len(set(own)) != len(own):
            problems.append("agent.own_tools names a tool twice")
    budget = agent.get("budget")
    if budget is not None:
        if not isinstance(budget, dict) or not set(budget) <= {"calls_per_turn", "calls_per_day"}:
            problems.append("agent.budget is {calls_per_turn?, calls_per_day?}")
        else:
            for key, value in budget.items():
                if not isinstance(value, int) or isinstance(value, bool) or value < 1:
                    problems.append(f"agent.budget.{key} must be a positive integer")
    grants = agent.get("grants", [])
    if not isinstance(grants, list) or not all(isinstance(g, dict) and set(g) == {"app", "tool"}
                                               and all(isinstance(g[k], str) for k in g) for g in grants):
        problems.append('agent.grants must be a list of {"app": <owning app>, "tool": <its tool>}')
    return problems


def grant_problems(apps):
    """Each cross-app grant names another app's shareable tool: a native
    owner's tool must be in its `agent.tools` and `shareable`; a script
    owner (`os.*`) is checked when App Hub loads its `tools.json`."""
    problems = []
    declared = {a.get("id"): {t.get("name"): t for t in (a["agent"].get("tools") or []) if isinstance(t, dict)} for a in apps}
    for app in apps:
        grants = app["agent"].get("grants", [])
        if not isinstance(grants, list):
            continue
        for grant in grants:
            if not isinstance(grant, dict) or set(grant) != {"app", "tool"}:
                continue
            owner, tool = grant["app"], grant["tool"]
            where = f"{app.get('id')}: agent.grants {owner}/{tool}"
            if owner == app.get("id"):
                problems.append(f"{where}: an app's own tools need no grant")
            elif owner in declared:
                entry = declared[owner].get(tool)
                if entry is None:
                    problems.append(f"{where}: {owner} declares no {tool}")
                elif entry.get("shareable") is not True:
                    problems.append(f"{where}: {tool} is not shareable")
            elif not (isinstance(owner, str) and owner.startswith("os.")):
                problems.append(f"{where}: no native app {owner}, and not a system app (os.*)")
    return problems


def feature_of(app):
    return app.get("feature") or f"app-{app['id']}"


def modules_of(app):
    """The app's linked modules; none for a process-only app."""
    module = app["module"]
    if isinstance(module, str):
        return [module]
    if isinstance(module, list) and all(isinstance(m, str) for m in module):
        return module
    return []


def same_git(a, b):
    def norm(url):
        url = url.strip().rstrip("/")
        return (url[:-4] if url.endswith(".git") else url).lower()
    return norm(a) == norm(b)


# --- Rendering -------------------------------------------------------------


def s(value):
    """A TOML basic string (JSON's escaping is a subset of TOML's)."""
    return json.dumps(value)


def toml_list(values):
    return "[" + ", ".join(s(v) for v in values) + "]"


def pin_line(app):
    source = app["source"]
    if "path" in source:
        fields = [f"path = {s(source['path'])}"]
    else:
        pin = f"tag = {s(source['tag'])}" if "tag" in source else f"rev = {s(source['rev'])}"
        fields = [f"git = {s(source['git'])}", pin]
    if not app["default_features"]:
        fields.append("default-features = false")
    if app["crate_features"]:
        fields.append(f"features = {toml_list(app['crate_features'])}")
    return f"{app['crate']} = {{ {', '.join(fields)} }}"


def patch_urls(apps):
    """The git sources some app resolves to a local checkout, in order."""
    urls = []
    for app in apps:
        source = app["source"]
        if source.get("local") and not any(same_git(u, source["git"]) for u in urls):
            urls.append(source["git"])
    return urls


def root_blocks(apps):
    blocks = {"pins": [pin_line(app) for app in apps]}
    for url in patch_urls(apps):
        blocks[f"patch {url}"] = [
            f"{app['crate']} = {{ path = {s(app['source']['local'])} }}"
            for app in apps
            if app["source"].get("local") and same_git(app["source"]["git"], url)
        ]
    return blocks


def in_mobile_apps(app):
    return app["shells"]["phone"] in ("default", "mobile-apps")


def shell_blocks(apps):
    def dep(app, optional):
        fields = ["workspace = true"]
        if app["crate_features"]:
            fields.append(f"features = {toml_list(app['crate_features'])}")
        if optional:
            fields.append("optional = true")
        return f"{app['crate']} = {{ {', '.join(fields)} }}"

    # A process-only app is no dependency of the shell's: nothing links it.
    apps = [app for app in apps if modules_of(app)]
    mobile_apps = [feature_of(app) for app in apps if in_mobile_apps(app)] + BASE_DEFAULT
    features = [f"mobile-apps = {toml_list(mobile_apps)}"]
    features += [f"{feature_of(app)} = {toml_list(['dep:' + app['crate']] + app['implies'])}" for app in apps]
    return {
        "deps": [dep(app, True) for app in apps],
        "mobile deps": [dep(app, False) for app in apps if app["native_mobile"] == "always"],
        "features": features,
    }


def process_blocks(apps):
    """`crates/process-apps/Cargo.toml`: every app with a `bin`, with the
    features its binary needs, as a process launch builds it."""
    deps = []
    for app in apps:
        if not app["bin"]:
            continue
        features = app["crate_features"] + [f for f in app.get("bin_features", []) if f not in app["crate_features"]]
        fields = ["workspace = true"] + ([f"features = {toml_list(features)}"] if features else [])
        deps.append(f"{app['crate']} = {{ {', '.join(fields)} }}")
    return {"process deps": deps}


def optional_deps(cargo_toml):
    """The optional dependencies a package's own [dependencies] declares."""
    names, table = set(), None
    for line in cargo_toml.splitlines():
        stripped = line.strip()
        if stripped.startswith("["):
            table = stripped
            continue
        if table == "[dependencies]":
            match = re.match(r"([A-Za-z0-9_-]+)\s*=\s*\{(.*)\}", stripped)
            if match and re.search(r"\boptional\s*=\s*true\b", match.group(2)):
                names.add(match.group(1))
    return names


def package_block(apps, shell, cargo_toml):
    """A package's default and forwarded features. An app whose crate the
    package also depends on itself (optionally) turns that dependency on."""
    own = optional_deps(cargo_toml)
    linked = [app for app in apps if app["shells"][shell] != "off"]
    default = BASE_DEFAULT + SHELL_BASE_DEFAULT.get(shell, []) + [feature_of(app) for app in linked if app["shells"][shell] == "default"]
    mobile = [feature_of(app) for app in linked if in_mobile_apps(app) and app["crate"] in own]
    lines = [f"default = {toml_list(default)}",
             f"mobile-apps = {toml_list(mobile + ['octosense-shell/mobile-apps'])}"]
    for app in linked:
        deps = [f"dep:{app['crate']}"] if app["crate"] in own else []
        lines.append(f"{feature_of(app)} = {toml_list(deps + ['octosense-shell/' + feature_of(app)])}")
    return {"features": lines}


def rust_raw(text):
    """A Rust raw string literal holding `text`."""
    if '"#' in text:
        raise ManifestError(f"cannot embed {text!r} in a raw string")
    return f'r#"{text}"#'


def rust_hosting(value):
    return {"module": "Hosting::Module", "process": "Hosting::Process",
            "process-if-vulkan": "Hosting::ProcessIfVulkan", "none": "Hosting::None"}[value]


def render_rust(apps):
    out = [
        f"//! @generated by tools/native_apps.py from native-apps.json: do not edit;",
        "//! change the manifest and run the script (ADR 0004 §1).",
        "//!",
        "//! The native apps this shell can link, in link order: the modules",
        "//! `apps::linked_modules()` pushes under each app's feature, and the",
        "//! hosting each app gets by default on each target.",
        "",
        "use makepad_app_module::AppModule;",
        "",
        "/// An app's hosting on one target, as the manifest declares it.",
        "#[derive(Clone, Copy, Debug, PartialEq, Eq)]",
        "pub enum Hosting {",
        "    /// In the shell's process: a splash isolate per instance.",
        "    Module,",
        "    /// Its own process (`bin`), connected to the shell's hub.",
        "    Process,",
        "    /// Its own process on a Vulkan build in a Wayland session; in the",
        "    /// shell's process otherwise (a process-only app: not there).",
        "    ProcessIfVulkan,",
        "    /// Not on this target: a process-only app (no module) where",
        "    /// there are no processes.",
        "    None,",
        "}",
        "",
        "/// Who shows the person a tool's live confirmation (ADR 0004 §8).",
        "#[derive(Clone, Copy, Debug, PartialEq, Eq)]",
        "pub enum Confirm {",
        "    /// The shell's own sheet (the assistant's confirm card).",
        "    Host,",
        "    /// The owning app's own sheet.",
        "    App,",
        "}",
        "",
        "/// How the shell treats one of an app's assistant tools.",
        "#[derive(Debug)]",
        "pub struct ToolPolicy {",
        "    pub tool: &'static str,",
        "    pub confirm: Confirm,",
        "    /// Whether a standing rule may answer its confirmation; `false`:",
        "    /// only the person, live, every time (outside developer mode).",
        "    pub auto_approvable: bool,",
        "}",
        "",
        "/// What a process-hosted instance may reach on the network (ADR 0004 §3).",
        "#[derive(Clone, Copy, Debug, PartialEq, Eq)]",
        "pub enum Network {",
        "    /// Nothing but the shell's hub on loopback.",
        "    None,",
        "    Any,",
        "}",
        "",
        "/// The port to the shell's octos kernel an app that is itself an octos",
        "/// client gets (`kernel`; ADR 0003).",
        "#[derive(Clone, Copy, Debug, PartialEq, Eq)]",
        "pub enum KernelPort {",
        "    /// None: a port the app opens is closed.",
        "    None,",
        "    /// The coding scope the kernel router enforces: the app's own",
        "    /// sessions, in workspaces the person picked.",
        "    Coding,",
        "}",
        "",
        "/// One `native-apps.json` entry, as far as the shell reads it.",
        "#[derive(Debug)]",
        "pub struct NativeApp {",
        "    pub id: &'static str,",
        "    /// The shell's Cargo feature that links it.",
        "    pub feature: &'static str,",
        "    /// The binary a process-hosted instance runs.",
        "    pub bin: Option<&'static str>,",
        "    pub macos: Hosting,",
        "    pub windows: Hosting,",
        "    pub linux: Hosting,",
        "    pub android: Hosting,",
        "    pub ios: Hosting,",
        "    pub ohos: Hosting,",
        "    pub wasm: Hosting,",
        "    /// The `octos.*` services its reviewed entry grants it.",
        "    pub octos: &'static [&'static str],",
        "    /// Its assistant tools' confirmation rules (`agent.tool_policy`).",
        "    pub tools: &'static [ToolPolicy],",
        "    /// `sandbox.network`: what its OS sandbox lets it reach.",
        "    pub network: Network,",
        "    /// `sandbox.processes`: whether it may start child processes.",
        "    pub processes: bool,",
        "    /// `storage.accounts`: one folder (and agent) per account.",
        "    pub accounts: bool,",
        "    /// `storage.external`: `<root>[/<path>]:ro|rw` outside its jail.",
        "    pub external: &'static [&'static str],",
        "    /// The whole `storage` block as JSON, which the shell parses with",
        "    /// `app_storage::StorageSpec` at startup (ADR 0004 §11).",
        "    pub storage: &'static str,",
        "    /// `agent.tools`: its own tools, the `tools.json` entries (a JSON",
        "    /// array; empty when it declares none).",
        "    pub tools_json: &'static str,",
        "    /// `agent.generic_tools`: exactly the octos kernel tools its agent",
        "    /// gets (none when empty; never octos's shell).",
        "    pub generic_tools: &'static [&'static str],",
        "    /// `agent.grants`: other apps' shareable tools its agent may call,",
        "    /// as (owning app, tool).",
        "    pub grants: &'static [(&'static str, &'static str)],",
        "    /// `agent.system_tools`: its own read tools the system agent may",
        "    /// call while the app runs (full names).",
        "    pub system_tools: &'static [&'static str],",
        "    /// `agent.own_tools`: the tools its own agent may call (full",
        "    /// names; every one of `tools` unless the entry narrows them).",
        "    pub own_tools: &'static [&'static str],",
        "    /// `agent.budget`: its agent's tool calls per turn and per day",
        "    /// (`None`: the shell's defaults).",
        "    pub calls_per_turn: Option<u32>,",
        "    pub calls_per_day: Option<u32>,",
        "    /// `kernel`: its port to the shell's kernel.",
        "    pub kernel: KernelPort,",
        "}",
        "",
        "pub const APPS: &[NativeApp] = &[",
    ]
    for app in apps:
        hosting = dict(app["hosting"])
        hosting.setdefault("wasm", "module" if modules_of(app) else "none")
        bin_value = f"Some({s(app['bin'])})" if app["bin"] else "None"
        octos = ", ".join(s(x) for x in app["agent"]["octos"])
        out.append("    NativeApp {")
        out.append(f"        id: {s(app['id'])},")
        out.append(f"        feature: {s(feature_of(app))},")
        out.append(f"        bin: {bin_value},")
        for target in TARGETS + ("wasm",):
            out.append(f"        {target}: {rust_hosting(hosting[target])},")
        out.append(f"        octos: &[{octos}],")
        policy = app["agent"].get("tool_policy", {})
        if policy:
            out.append("        tools: &[")
            for tool, rule in policy.items():
                confirm = "Confirm::Host" if rule["confirm"] == "host" else "Confirm::App"
                auto = "true" if rule["auto_approvable"] else "false"
                out.append(f"            ToolPolicy {{ tool: {s(tool)}, confirm: {confirm}, auto_approvable: {auto} }},")
            out.append("        ],")
        else:
            out.append("        tools: &[],")
        network = "Network::Any" if app["sandbox"]["network"] == "any" else "Network::None"
        out.append(f"        network: {network},")
        out.append(f"        processes: {'true' if app['sandbox']['processes'] else 'false'},")
        out.append(f"        accounts: {'true' if app['storage']['accounts'] else 'false'},")
        external = ", ".join(s(x) for x in app["storage"]["external"])
        out.append(f"        external: &[{external}],")
        out.append(f"        storage: {rust_raw(json.dumps(app['storage'], separators=(', ', ': ')))},")
        agent = app["agent"]
        tools = json.dumps(agent.get("tools") or [], separators=(",", ":"), ensure_ascii=False)
        out.append(f"        tools_json: r##\"{tools}\"##,")
        generic = ", ".join(s(x) for x in agent.get("generic_tools", []))
        out.append(f"        generic_tools: &[{generic}],")
        grants = ", ".join(f"({s(g['app'])}, {s(g['tool'])})" for g in agent.get("grants", []))
        out.append(f"        grants: &[{grants}],")
        system = ", ".join(s(x) for x in agent.get("system_tools", []))
        out.append(f"        system_tools: &[{system}],")
        own = agent.get("own_tools")
        if own is None:
            own = [tool["name"] for tool in agent.get("tools") or []]
        out.append(f"        own_tools: &[{', '.join(s(x) for x in own)}],")
        budget = agent.get("budget") or {}
        for key in ("calls_per_turn", "calls_per_day"):
            value = f"Some({budget[key]})" if key in budget else "None"
            out.append(f"        {key}: {value},")
        out.append(f"        kernel: KernelPort::{app.get('kernel', 'none').capitalize()},")
        out.append("    },")
    out += [
        "];",
        "",
        "impl NativeApp {",
        "    /// Its hosting on the target this shell is built for.",
        "    pub const fn on_this_target(&self) -> Hosting {",
        "        if cfg!(target_arch = \"wasm32\") {",
        "            self.wasm",
        "        } else if cfg!(target_os = \"android\") {",
        "            self.android",
        "        } else if cfg!(target_os = \"ios\") {",
        "            self.ios",
        "        } else if cfg!(target_env = \"ohos\") {",
        "            self.ohos",
        "        } else if cfg!(target_os = \"macos\") {",
        "            self.macos",
        "        } else if cfg!(target_os = \"windows\") {",
        "            self.windows",
        "        } else {",
        "            self.linux",
        "        }",
        "    }",
        "}",
        "",
        "impl NativeApp {",
        "    /// The rule for one of its tools, if the manifest sets one.",
        "    pub fn tool(&self, name: &str) -> Option<&'static ToolPolicy> {",
        "        self.tools.iter().find(|rule| rule.tool == name)",
        "    }",
        "}",
        "",
        "/// The manifest entry for an app id.",
        "pub fn find(id: &str) -> Option<&'static NativeApp> {",
        "    APPS.iter().find(|app| app.id == id)",
        "}",
        "",
        "/// The Cargo package (`crate`) of an app with a `bin`: what a process",
        "/// launch builds from the workspace, `cargo build --locked -p",
        "/// octosense-process-apps -p <package>` (crates/process-apps).",
        "pub fn package_of(id: &str) -> Option<&'static str> {",
        "    match id {",
    ]
    out += [f"        {s(app['id'])} => Some({s(app['crate'])})," for app in apps if app["bin"]]
    out += [
        "        _ => None,",
        "    }",
        "}",
        "",
        "/// Push the native modules this build links, in manifest order.",
        "#[allow(unused_variables)]",
        "pub fn link(out: &mut Vec<&'static dyn AppModule>) {",
    ]
    for app in apps:
        feature = f"feature = {s(feature_of(app))}"
        cfg = f"any({feature}, native_mobile)" if app["native_mobile"] == "always" else feature
        modules = modules_of(app)
        if not modules:
            continue
        out.append(f"    #[cfg({cfg})]")
        if len(modules) == 1:
            out.append(f"    out.push(&{modules[0]});")
        else:
            out.append("    {")
            out += [f"        out.push(&{m});" for m in modules]
            out.append("    }")
    out += ["}", ""]
    return "\n".join(out)


def render_agents(apps):
    """`crates/ai-host/src/native_agents.rs`: the agent grants
    `Policy::shipped()` hands out, one line per app granted `octos.*`."""
    out = [
        "//! @generated by tools/native_apps.py from native-apps.json: do not edit;",
        "//! change the manifest and run the script (ADR 0004 §1, §4).",
        "//!",
        "//! The native apps whose reviewed entry grants their agent `octos.*`",
        "//! services (`agent.octos`): what [`crate::Policy::shipped`] hands out.",
        "",
        "/// (app id, the `octos.*` services its agent is granted).",
        "pub const NATIVE_AGENTS: &[(&str, &[&str])] = &[",
    ]
    for app in apps:
        octos = app["agent"]["octos"]
        if octos:
            out.append(f"    ({s(app['id'])}, &[{', '.join(s(x) for x in octos)}]),")
    out += ["];", ""]
    return "\n".join(out)


# --- Marked blocks ---------------------------------------------------------


def begin_marker(name):
    return f"# BEGIN native-apps: {name} {NOTE}"


def end_marker(name):
    return f"# END native-apps: {name}"


def replace_blocks(text, blocks, label):
    """`text` with each named block's lines between its markers replaced."""
    lines = text.split("\n")
    for name, body in blocks.items():
        begin, end = begin_marker(name), end_marker(name)
        starts = [i for i, line in enumerate(lines) if line.strip() == begin]
        ends = [i for i, line in enumerate(lines) if line.strip() == end]
        if len(starts) != 1 or len(ends) != 1 or ends[0] < starts[0]:
            raise ManifestError(f"{label}: needs exactly one '{begin}' ... '{end}' block")
        lines[starts[0] + 1:ends[0]] = body
    marked = re.findall(r"^\s*# BEGIN native-apps: (.*?) \(generated", text, re.M)
    for name in marked:
        if name not in blocks:
            raise ManifestError(f"{label}: block '{name}' has no source in {MANIFEST} (remove its markers)")
    return "\n".join(lines)


def workspace_pins(text):
    """(crate, git, rev) of each git pin in [workspace.dependencies]."""
    pins, table = [], None
    for line in text.splitlines():
        stripped = line.strip()
        if stripped.startswith("["):
            table = stripped
            continue
        if table != "[workspace.dependencies]":
            continue
        match = re.match(r"([A-Za-z0-9_-]+)\s*=\s*\{(.*)\}", stripped)
        if not match:
            continue
        git = re.search(r'\bgit\s*=\s*"([^"]+)"', match.group(2))
        rev = re.search(r'\brev\s*=\s*"([^"]+)"', match.group(2))
        if git and rev:
            pins.append((match.group(1), git.group(1), rev.group(1)))
    return pins


def check_one_revision(apps, text):
    """Each git repository an app comes from is pinned at one revision."""
    problems = []
    for app in apps:
        source = app["source"]
        if "git" not in source or "tag" in source:
            continue
        for crate, git, rev in workspace_pins(text):
            if same_git(git, source["git"]) and rev != source["rev"]:
                problems.append(f"{crate} pins {git} at {rev[:12]}, {app['id']} at {source['rev'][:12]}")
    if problems:
        raise ManifestError("Cargo.toml: one revision per repository:\n  " + "\n  ".join(sorted(set(problems))))


def generate(root):
    """{relative path: (current text or None, generated text)}."""
    root = Path(root)
    apps = load(root)

    def read(rel):
        path = root / rel
        return path.read_text() if path.is_file() else None

    out = {}
    text = read("Cargo.toml")
    if text is None:
        raise ManifestError("Cargo.toml is missing")
    new = replace_blocks(text, root_blocks(apps), "Cargo.toml")
    check_one_revision(apps, new)
    out["Cargo.toml"] = (text, new)
    rel = "crates/shell/Cargo.toml"
    text = read(rel)
    if text is None:
        raise ManifestError(f"{rel} is missing")
    out[rel] = (text, replace_blocks(text, shell_blocks(apps), rel))
    text = read(PROCESS_APPS)
    if text is None:
        raise ManifestError(f"{PROCESS_APPS} is missing")
    out[PROCESS_APPS] = (text, replace_blocks(text, process_blocks(apps), PROCESS_APPS))
    for shell, rel in PACKAGES.items():
        text = read(rel)
        if text is None:
            raise ManifestError(f"{rel} is missing")
        out[rel] = (text, replace_blocks(text, package_block(apps, shell, text), rel))
    out[RUST_FILE] = (read(RUST_FILE), render_rust(apps))
    out[AGENTS_FILE] = (read(AGENTS_FILE), render_agents(apps))
    return out


def check_tagged_locks(root, apps):
    """A tag-pinned app's Cargo.lock entry names the manifest's commit.

    Cargo records `git+<url>?tag=<tag>#<commit>`; a tag moved after review
    shows up here as a different commit."""
    lock = Path(root) / "Cargo.lock"
    if not lock.is_file():
        return []
    text, problems = lock.read_text(), []
    for app in apps:
        source = app["source"]
        if "tag" not in source:
            continue
        pattern = re.compile(r'name = "%s"\nversion = "[^"]*"\nsource = "git\+([^"?#]+)\?tag=([^"#]+)#([0-9a-f]{40})"' % re.escape(app["crate"]))
        found = [m for m in pattern.finditer(text) if same_git(m.group(1), source["git"])]
        if not found:
            problems.append(f"Cargo.lock has no {app['crate']} from {source['git']} at tag {source['tag']}; run python3 tools/native_apps.py")
        for m in found:
            if m.group(2) != source["tag"] or m.group(3) != source["rev"]:
                problems.append(f"Cargo.lock: {app['crate']} {m.group(2)} is {m.group(3)[:12]}, native-apps.json says {source['tag']} is {source['rev'][:12]} (a moved tag?)")
    return problems


DESKTOP_CATALOG = "desktop/config/apps.json"
DESKTOP_TARGETS = ("macos", "windows", "linux")


def catalog_problems(root, apps):
    """The desktop catalog's rows for native apps agree with the manifest.

    A process launch builds `crate` and runs `bin` from native-apps.json
    (`clients::launch_plan`); the catalog row only names and places the app.
    So a row that says another package, binary or source would only
    mislead, and an app that runs as a process on a desktop needs a row."""
    path = Path(root) / DESKTOP_CATALOG
    if not path.is_file():
        return [f"{DESKTOP_CATALOG} is missing"]
    try:
        rows = json.loads(path.read_text())
    except json.JSONDecodeError as error:
        return [f"{DESKTOP_CATALOG}: {error}"]
    by_id = {row.get("id"): row for row in rows if isinstance(row, dict)}
    problems = []
    for app in apps:
        ident = app["id"]
        row = by_id.get(ident)
        process = any(app["hosting"][target] != "module" for target in DESKTOP_TARGETS)
        if row is None:
            if process:
                problems.append(f"{DESKTOP_CATALOG} has no row for {ident}, which runs as a process on a desktop")
            continue
        if not app["bin"]:
            problems.append(f"{DESKTOP_CATALOG}: {ident} has a row but no bin in {MANIFEST}: it has no process form")
            continue
        if row.get("package") != app["crate"]:
            problems.append(f"{DESKTOP_CATALOG}: {ident}'s package is {row.get('package')!r}, {MANIFEST} says {app['crate']!r}")
        if row.get("bin") != app["bin"]:
            problems.append(f"{DESKTOP_CATALOG}: {ident}'s bin is {row.get('bin')!r}, {MANIFEST} says {app['bin']!r}")
        source = app["source"]
        if "path" in source:
            want = (Path(root) / source["path"] / "Cargo.toml").resolve()
            got = row.get("manifest")
            if got is None or (path.parent / got).resolve() != want:
                problems.append(f"{DESKTOP_CATALOG}: {ident}'s manifest should be {source['path']}/Cargo.toml, as {MANIFEST} says")
        elif str(source.get("local") or "").startswith(".sources/makepad/"):
            if row.get("source") != "makepad":
                problems.append(f"{DESKTOP_CATALOG}: {ident} comes from Makepad in {MANIFEST}; its row should say \"source\": \"makepad\"")
        else:
            problems.append(f"{DESKTOP_CATALOG}: {ident}'s source in {MANIFEST} has no catalog form")
    return problems


def changed_pins(old, new):
    """Crates whose pin in [workspace.dependencies] changed (not added)."""
    def pins(text):
        found, table = {}, None
        for line in text.splitlines():
            stripped = line.strip()
            if stripped.startswith("["):
                table = stripped
                continue
            match = re.match(r"([A-Za-z0-9_-]+)\s*=", stripped)
            if table == "[workspace.dependencies]" and match:
                found[match.group(1)] = stripped
        return found
    before, after = pins(old), pins(new)
    return sorted(c for c in before if c in after and before[c] != after[c])


def update_lock(root, crates, cargo="cargo"):
    for crate in crates:
        note(f"cargo update -p {crate}")
        subprocess.run([cargo, "update", "-p", crate], cwd=root, check=True)
    note("cargo metadata (settles Cargo.lock)")
    subprocess.run([cargo, "metadata", "--format-version", "1"], cwd=root, check=True, stdout=subprocess.DEVNULL)


def main(argv=None, root=ROOT):
    parser = argparse.ArgumentParser(description=__doc__.split("\n")[0])
    parser.add_argument("--check", action="store_true", help="change nothing; exit 1 when a generated place drifted")
    parser.add_argument("--no-lock", action="store_true", help="do not run cargo to update Cargo.lock")
    args = parser.parse_args(argv)
    try:
        results = generate(root)
    except ManifestError as error:
        print(f"native-apps: {error}", file=sys.stderr)
        return 2
    drifted = [rel for rel, (old, new) in results.items() if old != new]
    if args.check:
        for rel in drifted:
            print(f"::error::{rel} is not what native-apps.json generates; run python3 tools/native_apps.py")
        lock_problems = check_tagged_locks(root, load(root)) + catalog_problems(root, load(root))
        for problem in lock_problems:
            print(f"::error::{problem}")
        if not drifted and not lock_problems:
            note("every generated place matches native-apps.json")
        return 1 if drifted or lock_problems else 0
    catalog = catalog_problems(root, load(root))
    for problem in catalog:
        print(f"::error::{problem}")
    for rel in drifted:
        (Path(root) / rel).write_text(results[rel][1])
        note(f"wrote {rel}")
    if not drifted:
        note("nothing to do")
    toml_changed = any(rel.endswith("Cargo.toml") for rel in drifted)
    if toml_changed and not args.no_lock:
        old, new = results["Cargo.toml"]
        update_lock(root, changed_pins(old, new))
    return 1 if catalog else 0


if __name__ == "__main__":
    sys.exit(main())