diff --git a/CHANGELOG.md b/CHANGELOG.md
index aaa72f9..c4bd759 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,56 @@ All notable changes to Code Context Control (C3) are documented here.
The format is loosely based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+## [2.157.0] - 2026-10-14
+
+### Added — the fleet views on the mobile gateway (API version 7)
+
+C3 Desk reads its Locks, Permissions, Rules, Built-in guards, Sessions and
+Now views from the Hub on loopback, which a phone cannot reach. The Oracle's
+mobile gateway now serves the same read models, so the companion app can show
+everything Desk shows. What the phone may change stays narrower than Desk.
+
+- `services/fleet_overview.py` builds the cross-project payloads (locks,
+ Access Guard rules, tool discipline, override policy, built-in guards, and a
+ Now snapshot). The Hub's `/api/hub/locks/overview`,
+ `/api/hub/access/overview`, `/api/hub/enforcement/overview`,
+ `/api/hub/overrides/policy/overview` and `GET /api/hub/access/builtin` now
+ call it too, so Desk and the phone read the same rows. Their responses are
+ unchanged.
+- New gateway routes (`oracle/services/mobile_fleet.py`):
+ - `GET /api/mobile/locks/overview`. Read only: breaking a lease stays on the
+ Hub.
+ - `GET /api/mobile/access/overview` and `GET /api/mobile/enforcement/overview`,
+ each with a `writable` block naming the scopes this gateway will edit.
+ - `GET /api/mobile/overrides/policy/overview` (global scope read-only).
+ - `POST /api/mobile/enforcement/clear` and
+ `POST /api/mobile/overrides/policy/clear`, project scope only. Clearing
+ asks for `confirm: "off"` or `confirm: "widen"` when what the project
+ would inherit is looser.
+ - `GET /api/mobile/access/builtin`, `POST /api/mobile/access/builtin/mode`
+ and `POST /api/mobile/access/builtin/lease/revoke`. The mode route only
+ tightens: a looser mode is a 403 (`needs_desk`, with the CLI command) for
+ every token kind, Desk's included. Leases are never minted here. Global
+ scope needs `mobile_access_global_scope`, even to tighten.
+ - `GET /api/mobile/sessions/overview`, `GET /api/mobile/sessions`,
+ `GET /api/mobile/sessions/detail` and `POST /api/mobile/sessions/mark`.
+ First and last prompts, the turn preview and titles copied from a first
+ prompt are withheld unless `mobile_sessions_transcripts` is on (default
+ off, per docs/sessions.md). Resume is not offered; the
+ `claude.ai/code` remote link is passed through.
+ - `GET /api/mobile/now`: live sessions from the MCP heartbeats, their newest
+ token totals and last C3 call, running jobs, last CI run and file-map
+ freshness, plus whether the Hub answers.
+- Capabilities `locks_overview`, `access_overview`, `enforcement_overview`,
+ `enforcement_clear`, `override_policy_overview`, `override_policy_clear`,
+ `builtin`, `builtin_tighten`, `sessions`, `sessions_write`, `now`. New
+ switches `mobile_builtin_enabled`, `mobile_builtin_tighten`,
+ `mobile_sessions_enabled`, `mobile_sessions_write` (default on) and
+ `mobile_sessions_transcripts` (default off).
+- Gateway invariant 2 is restated: no route can loosen or disable a built-in
+ guard. `tests/test_mobile_security_api.py` now pins the exact set of builtin
+ routes instead of banning the word.
+
## [2.156.0] - 2026-10-13
### Added — Scheduled tasks
diff --git a/README.md b/README.md
index d8959a7..bf31c96 100644
--- a/README.md
+++ b/README.md
@@ -324,7 +324,7 @@ C3 also records real per-tool usage to `.c3/tool_telemetry.jsonl`, so estimates
- **All web servers bind `127.0.0.1` by default** and are guarded against browser-based attacks even on loopback — a Host-header allowlist (defeats DNS rebinding) plus an Origin/Referer check on every request (defeats cross-origin CSRF), with scoped, non-wildcard CORS. There is still **no user authentication**, so do not expose these servers to an untrusted network without auth/TLS in front. Binding to a non-loopback interface is opt-in and warned at startup. *(Hardening added in v2.33.0.)*
- **Mobile gateway (`/api/mobile/*`)** — the Oracle exposes a companion-app surface for the C3 mobile app: merged cross-project feed, project overview, PM board read/write, digest, notification ack, plus the security surface (credential vault, Access Guard rules and masks, tool discipline). Unlike the legacy Oracle GETs it requires a **Bearer token on every method, GETs included** — the Discovery token, or (v2.125.0+) a **per-client token**: one credential per paired device, stored hashed in `~/.c3/oracle/clients.json`, revocable one at a time, and carrying a kind (`mobile` or `desk`) that override decisions are audited under. It validates every project path against the registry, and audits every write to the project's activity log and edit ledger. Pair the app from Oracle Settings → *Mobile app* (the QR carries the URL and a freshly minted per-device token; a legacy Discovery-token code is one button away). A same-machine client such as C3 Desk pairs itself with `POST /api/mobile/clients` and the owner-only `~/.c3/oracle/bootstrap.key`. Remote reachability is the same recipe as the Discovery API: set `bind_host` to the Tailscale/LAN interface IP (avoid `0.0.0.0`) and list that address in `allowed_hosts`, both in `~/.c3/oracle/config.json`; with such a bind the Oracle also listens on `127.0.0.1` (`loopback_listener`, default on) so local clients are not refused. Disable the whole surface with `"mobile_api_enabled": false`, or any subsystem with `mobile_credentials_enabled` / `mobile_access_enabled` / their `_write` twins.
-- **The mobile gateway never returns a credential value.** It has no reveal route by design: entries cross the wire through an allowlist serializer, and a value can only be identified by its length and an on-demand fingerprint. Two further levers ship **off** — `mobile_creds_agent_readable_raise` (raising `agent_readable` from the phone) and `mobile_access_global_scope` (machine-wide access rules) — because each exceeds what a leaked Bearer token could otherwise reach. Deliberately absent from the surface: bulk `.env` import, denial-counter clearing, `set_builtin_disabled`, and the raw-content access preview.
+- **The mobile gateway never returns a credential value.** It has no reveal route by design: entries cross the wire through an allowlist serializer, and a value can only be identified by its length and an on-demand fingerprint. Two further levers ship **off** — `mobile_creds_agent_readable_raise` (raising `agent_readable` from the phone) and `mobile_access_global_scope` (machine-wide access rules) — because each exceeds what a leaked Bearer token could otherwise reach. Deliberately absent from the surface: bulk `.env` import, denial-counter clearing, any way to loosen or disable a built-in guard (v2.157.0 adds read, tighten and end-a-lease-early only), lock force-release, and the raw-content access preview. Since v2.157.0 (API version 7) the gateway also serves the cross-project views C3 Desk reads from the Hub (locks, Access Guard rules, tool discipline, override policy, built-in guards, the session catalog without transcript text, and a Now snapshot), so the companion app can show what Desk shows.
- **No telemetry by default.** The OSS package collects nothing. Opt-in Sentry crash reporting requires the `[telemetry]` extra plus both `SENTRY_DSN` and `C3_TELEMETRY_OPT_IN=1`; even then request bodies, local variables, and prompts are stripped.
- **LLM memory distillation is local-first.** Cloud distillation (v2.51.0) is off by default and opt-in per project.
- **API keys** for third-party providers are read from the environment and never persisted by C3.
diff --git a/cli/c3.py b/cli/c3.py
index a40fb22..69f0a60 100644
--- a/cli/c3.py
+++ b/cli/c3.py
@@ -91,7 +91,7 @@
# Config
CONFIG_DIR = ".c3"
CONFIG_FILE = ".c3/config.json"
-__version__ = "2.156.0"
+__version__ = "2.157.0"
# The PreToolUse matcher for native subagent calls (installer and hub migration).
AGENT_MATCHER = "Agent|Task"
diff --git a/cli/guide/tools.html b/cli/guide/tools.html
index 9a76dd7..296e6d8 100644
--- a/cli/guide/tools.html
+++ b/cli/guide/tools.html
@@ -1610,7 +1610,7 @@ Examples
c3_schedule
execute
- v2.156.0
+ v2.157.0
Scheduled tasks — a skill, prompt or command the Oracle fires on a cadence, autonomously or after a human approves each run
diff --git a/cli/hub_server.py b/cli/hub_server.py
index 7f744d0..8ce9f56 100644
--- a/cli/hub_server.py
+++ b/cli/hub_server.py
@@ -31,7 +31,7 @@
from cli._hook_utils import hook_command_arg
from core.ide import PROFILES, detect_ide, get_profile, load_ide_config, normalize_ide_name
-from services import access_guard
+from services import access_guard, fleet_overview
from services.activity_log import ActivityLog
from services.atomic_json import write_json_atomic
from services.project_manager import ProjectManager
@@ -2710,38 +2710,11 @@ def api_hub_locks_overview():
Per-row isolation like /api/hub/credentials/overview: one unreadable
project must not blank the page. A project we cannot read reports
``error``/``initialized`` rather than an empty lease list — "0 leases"
- would claim the repo is clear when we simply do not know.
+ would claim the repo is clear when we simply do not know. Built by
+ services.fleet_overview, which the Oracle gateway's /locks/overview
+ shares, so Desk and the phone read the same rows.
"""
- from services import agent_locks as al
-
- rows, total = [], 0
- for p in _pm().list_registered():
- ppath = str(p.get("path") or "")
- row = {"name": p.get("name") or "", "path": ppath, "initialized": True,
- "error": None, "enabled": True, "mode": "advisory",
- "locks": [], "count": 0, "fencing": 0}
- try:
- if not (Path(ppath) / ".c3").is_dir():
- row["initialized"] = False
- else:
- cfg = al.config(ppath)
- row["enabled"] = cfg["enabled"]
- row["mode"] = cfg["mode"]
- snap = al.store_for(ppath).snapshot()
- row["locks"] = snap["locks"]
- row["count"] = snap["count"]
- row["fencing"] = snap["fencing"]
- total += snap["count"]
- except Exception as e:
- row["error"] = str(e)
- rows.append(row)
- return jsonify({"projects": rows, "total": total,
- # Surfaced so the UI can state its own limits instead of
- # implying a lease covers everything (spec §9).
- "coverage_note": (
- "Leases gate C3 tool surfaces only. A raw shell "
- "redirect, a non-Claude agent, or a human in an editor "
- "is not covered.")})
+ return jsonify(fleet_overview.locks(_pm().list_registered()))
@app.route("/api/projects/locks/force-release", methods=["POST"])
@@ -2798,84 +2771,10 @@ def api_hub_enforcement_overview():
Denial counts come from services.access_telemetry, split by layer so the
UI can point at the right lever: discipline blocks are cleared by changing
- the mode here, path denials by editing Access Guard rules.
+ the mode here, path denials by editing Access Guard rules. Shared with
+ the gateway's /enforcement/overview through services.fleet_overview.
"""
- from services import access_telemetry as at
- from services import enforcement_policy as ep
-
- rows = []
- totals = {"discipline": 0, "access": 0}
- for p in _pm().list_registered():
- ppath = str(p.get("path") or "")
- row = {"name": p.get("name") or "", "path": ppath, "initialized": True,
- "error": None, "mode": None, "scope": "", "set_by": "",
- "signal_ttl_s": ep.DEFAULT_SIGNAL_TTL_S, "warnings": [],
- "tier": "", "tier_implies": "", "denials": {}, "denial_total": 0}
- try:
- if not (Path(ppath) / ".c3").is_dir():
- row["initialized"] = False
- else:
- policy = ep.resolve(ppath)
- row["mode"] = policy.mode
- row["scope"] = policy.scope
- row["set_by"] = policy.set_by
- row["signal_ttl_s"] = policy.signal_ttl_s
- row["warnings"] = list(policy.warnings)
- row["blocked_tools"] = sorted(policy.blocked_tools)
-
- cfg_path = Path(ppath) / ".c3" / "config.json"
- try:
- cfg = json.loads(cfg_path.read_text(encoding="utf-8"))
- tier = str(cfg.get("permission_tier") or "")
- except Exception:
- tier = ""
- row["tier"] = tier
- row["tier_implies"] = ep.derive_from_tier(tier) if tier else ""
-
- agg = at.aggregate(ppath)
- row["denials"] = agg["by_layer"]
- row["denial_total"] = agg["total"]
- row["top_denials"] = [
- {**r, "fix": at.suggest(r)} for r in agg["rows"][:5]
- ]
- for layer, count in agg["by_layer"].items():
- totals[layer] = totals.get(layer, 0) + count
- except Exception as e:
- row["error"] = str(e)
- rows.append(row)
-
- # The global (~/.c3) section, shown as its own card. A corrupt global
- # config must not blank the tab — report the error inside the object.
- try:
- g = ep.resolve_global()
- global_policy = {
- "configured": g.scope == "global",
- "mode": g.mode or None,
- "set_by": g.set_by,
- "signal_ttl_s": g.signal_ttl_s,
- "blocked_tools": sorted(g.blocked_tools),
- "warnings": list(g.warnings),
- }
- except Exception as e:
- global_policy = {"configured": False, "mode": None, "error": str(e)}
-
- return jsonify({
- "projects": rows,
- "modes": [{"id": m, "help": ep.MODE_HELP[m]} for m in ep.MODES],
- "default_mode": ep.DEFAULT_MODE,
- "tier_map": ep.TIER_TO_MODE,
- "totals": totals,
- "global_policy": global_policy,
- # Stated so the tab can never imply that turning discipline down also
- # turns the security boundaries down. Mirrors docs/enforcement.md.
- "coverage_note": (
- "Tool discipline only governs whether native Edit/Write are "
- "pushed through c3_edit. At every mode — including off — Access "
- "Guard path rules, the credential-vault write guard, and agent "
- "locks still enforce. The edit ledger records native writes "
- "either way; strict additionally gets c3_edit's pre-edit snapshot."
- ),
- })
+ return jsonify(fleet_overview.enforcement(_pm().list_registered()))
@app.route("/api/projects/enforcement", methods=["GET"])
@@ -3526,12 +3425,9 @@ def api_hub_override_costs():
#: Verbatim from the spec, and the one thing a toggle screen must not let the
-#: reader forget: policy is a floor, not a promise.
-_OVERRIDE_POLICY_NOTE = (
- "Project and global policy merge by tightening only: a layer is "
- "escalatable iff both scopes allow it, and the numeric limits take "
- "the smaller value."
-)
+#: reader forget: policy is a floor, not a promise. One copy, shared with the
+#: overview and the Oracle gateway.
+_OVERRIDE_POLICY_NOTE = fleet_overview.OVERRIDE_POLICY_NOTE
def _hub_policy_audit(project_path: str, action: str, detail_extra: dict) -> None:
@@ -3650,51 +3546,10 @@ def api_hub_override_policy_overview():
section is reported as corrupt rather than as the disabled policy it
resolves to. ``project_keys`` names what the project pins itself, so a
screen can tell an inherited value from a chosen one. The global scope is
- served READ-ONLY: ``write_section`` is project-scope by design.
+ served READ-ONLY: ``write_section`` is project-scope by design. Shared
+ with the gateway's /overrides/policy/overview (services.fleet_overview).
"""
- from services import override_policy as opol
-
- rows = []
- for p in _pm().list_registered():
- ppath = str(p.get("path") or "")
- row = {"name": p.get("name") or Path(ppath).name, "path": ppath,
- "initialized": True, "error": None, "policy": None,
- "configured": False, "corrupt": False, "project_keys": []}
- try:
- if not (Path(ppath) / ".c3").is_dir():
- row["initialized"] = False
- else:
- keys, present, corrupt = opol.scope_keys(ppath)
- row["project_keys"] = keys
- row["configured"] = present
- row["corrupt"] = corrupt
- row["policy"] = opol.resolve(ppath).as_dict()
- except Exception as e:
- row["error"] = str(e)
- rows.append(row)
-
- try:
- keys, present, corrupt = opol.global_scope_keys()
- global_policy = {"configured": present, "corrupt": corrupt,
- "keys": keys, "error": None,
- "policy": opol.resolve_global().as_dict()}
- except Exception as e:
- global_policy = {"configured": False, "corrupt": False, "keys": [],
- "error": str(e), "policy": None}
-
- return jsonify({
- "projects": rows,
- "global_policy": global_policy,
- "defaults": {k: v for k, v in opol.DEFAULTS.items()
- if k != opol.WAKE_KEY},
- "layers": list(opol.LAYER_KEYS),
- "typed_confirm_layers": sorted(opol.TYPED_CONFIRM_LAYERS),
- "hard_max_ttl_s": opol.HARD_MAX_TTL_S,
- "coverage_note": _OVERRIDE_POLICY_NOTE,
- # What this Hub can do beyond the per-project GET/POST, so a client
- # built against a newer C3 can hide what an older one cannot serve.
- "features": ["clear"],
- })
+ return jsonify(fleet_overview.override_policy(_pm().list_registered()))
@app.route("/api/hub/overrides/policy/clear", methods=["POST"])
@@ -3962,45 +3817,10 @@ def api_hub_access_overview():
scope is reported as corrupt: it evaluates deny-all until repaired by
hand, which is the opposite of what an empty list would suggest.
Builtin rules are read once, from the global realm; a project-scope
- builtin mode (set with the CLI) is not reflected per row.
+ builtin mode (set with the CLI) is not reflected per row. Shared with the
+ gateway's /access/overview through services.fleet_overview.
"""
- empty = {"deny": [], "read_only": [], "confirm": [], "mask": 0,
- "corrupt": False}
- rows = []
- for p in _pm().list_registered():
- ppath = str(p.get("path") or "")
- row = {"name": p.get("name") or Path(ppath).name, "path": ppath,
- "initialized": True, "error": None, "rules": dict(empty)}
- try:
- if not (Path(ppath) / ".c3").is_dir():
- row["initialized"] = False
- else:
- row["rules"] = access_guard.scope_rules("project", ppath)
- except Exception as e:
- row["error"] = str(e)
- rows.append(row)
-
- try:
- global_rules = {**access_guard.scope_rules("global"), "error": None}
- except Exception as e:
- global_rules = {**empty, "error": str(e)}
- try:
- home = access_guard._global_base()
- b = access_guard.list_rules(str(home) if home else ".")["builtin"]
- builtin = {"deny": list(b.get("deny") or []),
- "read_only": list(b.get("read_only") or []),
- "confirm": list(b.get("confirm") or []), "error": None}
- except Exception as e:
- builtin = {"deny": [], "read_only": [], "confirm": [], "error": str(e)}
-
- return jsonify({
- "projects": rows,
- "global": global_rules,
- "builtin": builtin,
- # Strictest last. "allow" is the absence of a rule, not a kind.
- "kinds": ["confirm", "read_only", "deny"],
- "coverage_note": access_guard.COVERAGE_MATRIX,
- })
+ return jsonify(fleet_overview.access(_pm().list_registered()))
@app.route("/api/hub/access/rule", methods=["POST"])
@@ -4089,38 +3909,13 @@ def api_hub_access_builtin():
so a mode listed here is one evaluation honours. ``projects`` holds only
projects that set a mode of their own, or could not be read.
``strictness`` ranks each mode 0 (allow) to 3 (full deny), so a client
- can tell loosening from tightening without knowing the tiers.
+ can tell loosening from tightening without knowing the tiers. Shared
+ with the gateway's /access/builtin through services.fleet_overview.
"""
try:
- global_modes = access_guard._scope_modes("global")
+ return jsonify(fleet_overview.builtin(_pm().list_registered()))
except Exception as e:
return jsonify({"error": str(e)}), 500
- home = access_guard._global_base()
- guards = []
- for glob in access_guard.DISABLEABLE_BUILTINS:
- canon = access_guard._norm_builtin(glob)
- guards.append({
- "glob": canon,
- "global": global_modes.get(canon, "default"),
- "strictness": {m: access_guard.builtin_strictness(canon, m)
- for m in (*access_guard.BUILTIN_MODES, "default")},
- })
- projects = []
- for p in _pm().list_registered():
- ppath = str(p.get("path") or "")
- row = {"name": p.get("name") or Path(ppath).name, "path": ppath,
- "modes": {}, "error": None}
- try:
- base = Path(ppath).resolve()
- if (base / ".c3").is_dir() and base != home:
- row["modes"] = access_guard._scope_modes("project", ppath)
- except Exception as e:
- row["error"] = str(e)
- if row["modes"] or row["error"]:
- projects.append(row)
- from services import builtin_leases
- return jsonify({"guards": guards, "projects": projects,
- "leases": builtin_leases.active()})
@app.route("/api/hub/access/builtin/mode", methods=["POST"])
diff --git a/docs/access-guard.md b/docs/access-guard.md
index 4fbbf71..1fbcd0f 100644
--- a/docs/access-guard.md
+++ b/docs/access-guard.md
@@ -239,6 +239,15 @@ Hub, loopback only:
- A project without `.c3/` gets a 409, never a fresh config. Every real
write is audited on the target (a global one in `~/.c3`), `via: hub`.
+Since v2.157.0 the phone reads the same cross-project table from
+`GET /api/mobile/access/overview` (built by `services/fleet_overview.py`,
+which the Hub route above also calls). Its `writable` block says which scopes
+the gateway will edit; edits still go through `/api/mobile/access/rule` with
+its limits. Built-in guards on the gateway (`GET /api/mobile/access/builtin`,
+`POST .../builtin/mode`, `POST .../builtin/lease/revoke`) only ever tighten:
+a looser mode is a 403 for every token kind, and leases are minted from C3
+Desk alone.
+
C3 Desk's Locks → Permissions view is built on these three routes.
## 6. Residual risks (named, documented)
diff --git a/docs/override-requests.md b/docs/override-requests.md
index 15fd960..ae7bc3a 100644
--- a/docs/override-requests.md
+++ b/docs/override-requests.md
@@ -646,6 +646,8 @@ to see.
| `GET /api/mobile/overrides/policy` | The effective `override` section for a project (read-only view of §3.1). |
| `POST /api/mobile/overrides/policy` | Edit it, gated on `override_write` + typed confirm for any widening. |
| `GET /api/mobile/overrides/costs` | Per-(project, rule) counts over a trailing window (v2.127.0, §11 T5). Params `project` (optional — omit for every registered project), `days` (1..30, default 7). `{days, rules, count, project}`. |
+| `GET /api/mobile/overrides/policy/overview` | Every registered project's effective policy plus the read-only global scope (v2.157.0, capability `override_policy_overview`). Same payload as the Hub's overview, from `services/fleet_overview.py`. |
+| `POST /api/mobile/overrides/policy/clear` | `{project, confirm?}`: drop the project's own section so it inherits global; `confirm: "widen"` when that loosens (v2.157.0, `override_policy_clear`). |
Capabilities added to `mobile_api.CAPABILITIES`: `override`, `override_write`,
mapped to `mobile_override_enabled` / `mobile_override_write`; `override_costs`
diff --git a/docs/sessions.md b/docs/sessions.md
index 44e9bb8..a6ef072 100644
--- a/docs/sessions.md
+++ b/docs/sessions.md
@@ -122,8 +122,20 @@ has not been confirmed by Anthropic; an id in any other shape gets no link.
| `POST /api/hub/sessions/mark {path, id, op, reason?, successor?, summary?, next_steps?}` | recorded as `by: user` |
| `POST /api/hub/sessions/resume {path, id}` | 200 `{launched, command, cwd}`, 404, 409, or 500 with the command to run by hand |
-These are deliberately **not** on the Oracle gateway. Oracle can bind to the
-tailnet, and transcript previews should stay on this machine. The first
+Resume and the turn preview are deliberately **not** on the Oracle gateway.
+Oracle can bind to the tailnet, and transcript text should stay on this
+machine. Since 2.157.0 the gateway serves the catalog to the phone without it:
+
+| route | what |
+| --- | --- |
+| `GET /api/mobile/sessions/overview` | as the Hub's, over scanner-registered projects |
+| `GET /api/mobile/sessions?project=&stale=&q=&limit=&before=` | rows with `first_prompt` / `last_prompt` blanked and a title copied from the first prompt replaced by `(untitled)` (`title_source: withheld`); `resume` is only `{remote_url, can_launch: false, why_not}` |
+| `GET /api/mobile/sessions/detail?project=&id=` | row + decisions, snapshot and tasks; `preview` is empty |
+| `POST /api/mobile/sessions/mark {project, id, op, ...}` | recorded as `by: user` |
+
+`mobile_sessions_transcripts: true` in `~/.c3/oracle/config.json` lets the
+prompts, titles and preview through; every response says which with
+`transcripts`. The first
all-projects listing on a machine reads every transcript head and tail once.
Around 1,000 transcripts took about 20 seconds; later loads are served from
the cache.
diff --git a/oracle/config.py b/oracle/config.py
index 88fb664..1b8585c 100644
--- a/oracle/config.py
+++ b/oracle/config.py
@@ -83,6 +83,17 @@
"scheduler_default_max_budget_usd": 0,
"mobile_schedules_enabled": True, # read: schedules / runs / approvals
"mobile_schedules_write": True, # create / edit / run now / decide
+ # ── API v7 fleet views (oracle/services/mobile_fleet.py) ──
+ # Built-in guards: read modes and leases; tighten a mode and end a lease
+ # early. Loosening is never served here, whatever these say.
+ "mobile_builtin_enabled": True,
+ "mobile_builtin_tighten": True,
+ # Session catalog: rows, counts and marks. Transcript text (first/last
+ # prompt, the turn preview, a title copied from the first prompt) stays on
+ # this machine unless mobile_sessions_transcripts is on (docs/sessions.md).
+ "mobile_sessions_enabled": True,
+ "mobile_sessions_write": True,
+ "mobile_sessions_transcripts": False,
"auto_open_browser": True,
"theme": "dark",
"ui_last_tab": "chat", # persisted UI preference (hub parity)
diff --git a/oracle/services/mobile_api.py b/oracle/services/mobile_api.py
index d178f63..b5545f4 100644
--- a/oracle/services/mobile_api.py
+++ b/oracle/services/mobile_api.py
@@ -27,9 +27,13 @@
through ``credential_store.public_entry`` — an allowlist that cannot emit
one — and this module never imports ``get_value``/``expand_templates``.
``credential_store.is_resolvable`` exists so it never needs to.
-2. **No ``set_builtin_disabled`` route.** Disabling a builtin guard needs a
- keyring attestation and a typed confirmation; it stays a local, human
- operation on a machine someone is physically at.
+2. **No route can loosen or disable a builtin guard.** Disabling one, setting
+ a looser mode or minting a temporary lease needs a keyring attestation and
+ a person at the machine (C3 Desk's client token on the Hub); it stays a
+ local, human operation. Since API v7 the gateway READS builtin modes and
+ leases, TIGHTENS a mode and ENDS a lease early (``mobile_fleet``) — each
+ only ever makes the guard stricter. ``set_builtin_disabled`` is never
+ referenced here.
3. **No ``/access/preview`` equivalent.** That route (``cli/server.py``)
returns RAW file content and is explicitly human-only.
@@ -76,7 +80,14 @@
# trailing window, the "this rule is costing you" nudge (D3a). The
# /overrides/ detail also carries `allow_session_grants`,
# `max_ttl_s` and `session_confirm` at the top level.
-API_VERSION = 6
+# 6: scheduled tasks (`schedules`, `schedules_write`).
+# 7: the fleet views C3 Desk reads from the Hub (oracle/services/
+# mobile_fleet.py): cross-project /locks/overview, /access/overview,
+# /enforcement/overview, /overrides/policy/overview; /enforcement/clear
+# and /overrides/policy/clear; built-in guards read + tighten + lease
+# revoke (/access/builtin*); the session catalog (/sessions*, no
+# transcript text by default); and /now.
+API_VERSION = 7
CAPABILITIES = [
"feed", "feed_wait", "projects", "health", "pm", "pm_events", "digest",
@@ -99,6 +110,12 @@
# /api/mobile/chat/*. Advertised here because /info is the single place a
# client probes; the routes themselves live in that blueprint.
"chat",
+ # API v7 fleet views (oracle/services/mobile_fleet.py).
+ "locks_overview", "access_overview", "enforcement_overview",
+ "enforcement_clear", "override_policy_overview", "override_policy_clear",
+ "builtin", "builtin_tighten",
+ "sessions", "sessions_write",
+ "now",
]
# capability -> config key that gates it. Absent from this map = always on.
@@ -122,6 +139,17 @@
"suggestions_write": "mobile_suggestions_write",
"schedules": "mobile_schedules_enabled",
"schedules_write": "mobile_schedules_write",
+ # v7: each overview goes dark with the subsystem it reads; locks_overview
+ # and now are ops reads like /locks and /status, so always on.
+ "access_overview": "mobile_access_enabled",
+ "enforcement_overview": "mobile_access_enabled",
+ "enforcement_clear": "mobile_access_write",
+ "override_policy_overview": "mobile_override_enabled",
+ "override_policy_clear": "mobile_override_write",
+ "builtin": "mobile_builtin_enabled",
+ "builtin_tighten": "mobile_builtin_tighten",
+ "sessions": "mobile_sessions_enabled",
+ "sessions_write": "mobile_sessions_write",
}
FEED_TYPES = ("notification", "activity", "edit", "session_stat")
@@ -3529,3 +3557,10 @@ def mobile_review():
"interval_seconds": status.get("interval_seconds"),
"digest_enabled": bool(status.get("digest_enabled")),
})
+
+
+# ── API v7 fleet views ────────────────────────────────────
+# Registered on this blueprint for their side effect; kept in their own module
+# because they answer a different question (every project at once) from the
+# per-project routes above. Must run before oracle_server registers `bp`.
+from oracle.services import mobile_fleet as _mobile_fleet # noqa: E402,F401
diff --git a/oracle/services/mobile_fleet.py b/oracle/services/mobile_fleet.py
new file mode 100644
index 0000000..f925a36
--- /dev/null
+++ b/oracle/services/mobile_fleet.py
@@ -0,0 +1,584 @@
+"""Gateway API v7: the fleet views C3 Desk reads from the Hub, for the phone.
+
+C3 Desk answers "what is going on across every project" by asking the Hub on
+loopback. A phone cannot reach the Hub, so these routes serve the same read
+models over the authenticated gateway. The payloads come from
+``services.fleet_overview``, the module the Hub's own overview routes call,
+so Desk and the phone cannot drift apart field by field.
+
+What a phone may CHANGE here is narrower than what Desk may, on purpose:
+
+- Locks: read only. Breaking another agent's lease stays at the machine
+ holding the work (the Hub's force-release).
+- Access Guard and discipline: the overviews are read-only; per-project rule
+ and mode edits stay on the existing ``/access/rule`` and ``/enforcement``
+ routes with their limits. ``/enforcement/clear`` and
+ ``/overrides/policy/clear`` exist because "inherit again" is not a value
+ those routes can write; both ask for the typed word when the result loosens.
+- Built-in guards: read, TIGHTEN, and end a lease early. Loosening a
+ built-in guard, or minting a lease, is refused here for every token kind —
+ it stays with C3 Desk's Hub route, which proves a person with the Desk
+ client token. Nothing on this surface can disable a guard.
+- Sessions: catalog rows and marks, without transcript text. First and last
+ prompts, the turn preview and titles copied from a first prompt stay on
+ the machine (docs/sessions.md) unless ``mobile_sessions_transcripts`` is
+ switched on. Resuming opens a terminal on the PC, so it is not offered;
+ a ``claude.ai/code`` remote link is passed through when the session has
+ one.
+- Now: live sessions, running jobs, last CI and map freshness. Read only;
+ the restart and launch buttons Desk shows act on the PC and are absent.
+
+Imported for its side effect (route registration) at the bottom of
+``mobile_api``, so the routes exist before the blueprint is registered.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from flask import jsonify, request
+
+from oracle.services import mobile_api as m
+
+bp = m.bp
+
+#: Row fields that are verbatim transcript text.
+_TRANSCRIPT_FIELDS = ("first_prompt", "last_prompt")
+_STALE_FILTERS = {"hide", "only", "likely", "all"}
+_RESUME_ELSEWHERE = ("resume opens a terminal on the PC - use C3 Desk or the "
+ "Hub there")
+
+
+def _discovered() -> list[dict]:
+ """``[{name, path}]`` for every project the scanner knows."""
+ if m._scanner is None:
+ return []
+ try:
+ entries = m._scanner.discover()
+ except Exception:
+ return []
+ return [{"name": str(p.get("name") or Path(str(p.get("path") or "")).name),
+ "path": str(p.get("path") or "")}
+ for p in entries if p.get("path")]
+
+
+def _registered() -> list[dict]:
+ """``[{name, path}]`` for scanner projects with a ``.c3`` dir."""
+ if m._scanner is None:
+ return []
+ try:
+ entries = m._scanner.discover()
+ except Exception:
+ return []
+ return [{"name": str(p.get("name") or Path(str(p.get("path") or "")).name),
+ "path": str(p.get("path") or "")}
+ for p in entries if p.get("path") and p.get("has_c3")]
+
+
+def _name_for(resolved: Path) -> str:
+ want = str(resolved)
+ for p in _registered():
+ try:
+ if Path(p["path"]).resolve() == resolved:
+ return p["name"]
+ except OSError:
+ continue
+ return Path(want).name
+
+
+def _cap(name: str) -> bool:
+ return name in m._capabilities()
+
+
+# ── Locks ─────────────────────────────────────────────────
+
+@bp.route("/locks/overview", methods=["GET"])
+def mobile_locks_overview():
+ """Every project's agent leases in one call. Read-only by design."""
+ gate = m._feature_or_404("locks_overview")
+ if gate:
+ return gate
+ from services import fleet_overview
+ return jsonify(fleet_overview.locks(_discovered()))
+
+
+# ── Access Guard ──────────────────────────────────────────
+
+@bp.route("/access/overview", methods=["GET"])
+def mobile_access_overview():
+ """Each project's own path rules, plus global and builtin, for one glob
+ table. ``writable`` says which scopes this gateway will edit, so a client
+ shows the rest as read-only instead of collecting 403s."""
+ gate = m._feature_or_404("access_overview")
+ if gate:
+ return gate
+ from services import fleet_overview
+ body = fleet_overview.access(_discovered())
+ body["writable"] = {
+ "project": _cap("access_write"),
+ "global": _cap("access_write")
+ and bool(m._cfg().get("mobile_access_global_scope", False)),
+ }
+ return jsonify(body)
+
+
+# ── Tool discipline ───────────────────────────────────────
+
+@bp.route("/enforcement/overview", methods=["GET"])
+def mobile_enforcement_overview():
+ """Every project's discipline mode, its source and its denials."""
+ gate = m._feature_or_404("enforcement_overview")
+ if gate:
+ return gate
+ from services import fleet_overview
+ body = fleet_overview.enforcement(_discovered())
+ # Global-scope discipline is machine-wide and stays a desktop edit.
+ body["writable"] = {"project": _cap("enforcement_write"), "global": False}
+ return jsonify(body)
+
+
+@bp.route("/enforcement/clear", methods=["POST"])
+def mobile_enforcement_clear():
+ """Drop a project's own discipline mode so it inherits global again.
+
+ Body: ``{project, confirm?}``. Project scope only. When what it would
+ inherit is ``off`` (and it is not off already), this is the loosening
+ direction and needs ``confirm: "off"`` like setting off directly.
+ """
+ gate = (m._feature_or_404("enforcement_clear")
+ or m._feature_or_404("enforcement_write"))
+ if gate:
+ return gate
+ limited = m._security_gate()
+ if limited:
+ return limited
+ from services import enforcement_policy as ep
+ data = request.get_json(silent=True) or {}
+ resolved, err = m._project_or_404(data.get("project") or "")
+ if err:
+ return err
+ try:
+ current = ep.resolve(str(resolved)).mode or ep.DEFAULT_MODE
+ inherited = ep.resolve_global().mode or ep.DEFAULT_MODE
+ except (ValueError, TypeError, OSError) as exc:
+ return m._svc_error(exc)
+ if (inherited == ep.MODE_OFF and current != ep.MODE_OFF
+ and not m._confirmed(data, "off")):
+ return jsonify({
+ "error": "this project would inherit discipline off",
+ "needs_confirmation": True, "confirm_with": "off",
+ }), 400
+ try:
+ result = ep.clear(str(resolved), scope="project")
+ except (ValueError, TypeError) as exc:
+ return m._svc_error(exc)
+ if result.get("cleared"):
+ detail = {"kind": "enforcement", "action": "clear",
+ "previous": result.get("previous", ""),
+ "mode": result.get("mode", ""),
+ "scope": result.get("scope", ""), "via": "oracle-mobile",
+ "caller": m._caller()}
+ try:
+ from services.activity_log import ActivityLog
+ ActivityLog(str(resolved)).log("access_action", dict(detail))
+ except Exception:
+ pass
+ try:
+ from services.edit_ledger import EditLedger
+ EditLedger(str(resolved)).log_edit(
+ file="enforcement://project",
+ change_type="enforcement_clear",
+ summary=(f"tool discipline {result.get('previous') or 'default'}"
+ f" -> inherit ({result.get('mode')} from "
+ f"{result.get('scope')}) via mobile"),
+ tags=["enforcement", "access"], detail=dict(detail))
+ except Exception:
+ pass
+ m._gw_audit("mobile.enforcement.clear", {"mode": result.get("mode")})
+ return jsonify(result)
+
+
+# ── Override policy ───────────────────────────────────────
+
+@bp.route("/overrides/policy/overview", methods=["GET"])
+def mobile_override_policy_overview():
+ """Every project's effective override policy, plus the global scope
+ (read-only on every surface)."""
+ gate = m._feature_or_404("override_policy_overview")
+ if gate:
+ return gate
+ from services import fleet_overview
+ body = fleet_overview.override_policy(_discovered())
+ body["writable"] = {"project": _cap("override_write"), "global": False}
+ return jsonify(body)
+
+
+@bp.route("/overrides/policy/clear", methods=["POST"])
+def mobile_override_policy_clear():
+ """Drop a project's own ``override`` opinions so it inherits global.
+
+ Body: ``{project, confirm?}``. Clearing can LOOSEN (the project may have
+ been tighter than global), so it needs ``confirm: "widen"`` when it does.
+ ``wake`` survives the clear. A corrupt section is a 409.
+ """
+ gate = (m._feature_or_404("override_policy_clear")
+ or m._override_write_gate())
+ if gate:
+ return gate
+ from services import override_policy as opol
+ data = request.get_json(silent=True) or {}
+ resolved, err = m._project_or_404(data.get("project") or "")
+ if err:
+ return err
+ try:
+ kept, widens = opol.clear_section(
+ str(resolved), confirmed=m._confirmed(data, "widen"))
+ except opol.PolicyEditError as exc:
+ return jsonify(exc.payload), exc.status
+ except (OSError, ValueError) as exc:
+ return m._svc_error(exc)
+ cleared = kept is not None
+ if cleared:
+ m._override_audit("policy_clear",
+ {"id": "", "project_path": str(resolved), "rule": ""},
+ confirmed=bool(widens),
+ detail_extra={"kept": sorted(kept),
+ "widened": list(widens)})
+ m._gw_audit("mobile.override.policy_clear", {"widens": list(widens)})
+ return jsonify({"path": str(resolved), "cleared": cleared,
+ "policy": opol.resolve(str(resolved)).as_dict(),
+ "widened": widens})
+
+
+# ── Built-in guards ───────────────────────────────────────
+
+@bp.route("/access/builtin", methods=["GET"])
+def mobile_builtin_overview():
+ """Every Tier-1 builtin, its live global mode, per-project modes, leases.
+
+ ``loosen`` is always false here: a looser mode or a lease is set from
+ C3 Desk, never from this gateway.
+ """
+ gate = m._feature_or_404("builtin")
+ if gate:
+ return gate
+ from services import fleet_overview
+ try:
+ body = fleet_overview.builtin(_discovered())
+ except Exception as exc:
+ return m._svc_error(exc)
+ body["writable"] = {
+ "tighten": _cap("builtin_tighten"),
+ "tighten_global": _cap("builtin_tighten")
+ and bool(m._cfg().get("mobile_access_global_scope", False)),
+ "revoke": _cap("builtin_tighten"),
+ "loosen": False,
+ }
+ return jsonify(body)
+
+
+def _builtin_target(data: dict):
+ """``(scope, project_or_None, error_response_or_None)``."""
+ scope = str(data.get("scope") or "project").strip().lower()
+ if scope not in ("project", "global"):
+ return scope, None, (jsonify(
+ {"error": "scope must be 'project' or 'global'"}), 400)
+ if scope == "global":
+ if not m._cfg().get("mobile_access_global_scope", False):
+ return scope, None, (jsonify({
+ "error": "global-scope guard changes are disabled on this gateway",
+ "detail": "a global mode applies to every project on this "
+ "machine; enable mobile_access_global_scope, or use "
+ "C3 Desk.",
+ }), 403)
+ return scope, None, None
+ resolved, err = m._project_or_404(data.get("project") or "")
+ if err:
+ return scope, None, err
+ return scope, str(resolved), None
+
+
+@bp.route("/access/builtin/mode", methods=["POST"])
+def mobile_builtin_mode():
+ """Set a built-in guard's mode, in the TIGHTENING direction only.
+
+ Body: ``{scope: project|global, project?, glob, mode}``. A mode at least
+ as strict as the one agents see today is applied. A looser one is a 403
+ for every token kind, with the Desk route and CLI command that can do it.
+ """
+ gate = m._feature_or_404("builtin") or m._feature_or_404("builtin_tighten")
+ if gate:
+ return gate
+ limited = m._security_gate()
+ if limited:
+ return limited
+ from services import access_guard
+ data = request.get_json(silent=True) or {}
+ scope, project, err = _builtin_target(data)
+ if err:
+ return err
+ mode = str(data.get("mode") or "").strip()
+ canon = access_guard._norm_builtin(str(data.get("glob") or ""))
+ try:
+ new = access_guard.builtin_strictness(canon, mode)
+ live = (access_guard._scope_modes("global") if scope == "global"
+ else access_guard.effective_builtin_modes(project))
+ old = access_guard.builtin_strictness(canon, live.get(canon, "default"))
+ except ValueError as exc:
+ return jsonify({"error": str(exc)}), 400
+ if new < old:
+ return jsonify({
+ "error": "loosening a built-in guard is done from C3 Desk, "
+ "not the phone",
+ "needs_desk": True,
+ "command": f'c3 access builtin mode "{canon}" {mode}'
+ + (f' --project --path "{project}"'
+ if scope == "project" else ""),
+ }), 403
+ try:
+ result = access_guard.set_builtin_mode(canon, mode, scope,
+ project or ".")
+ except ValueError as exc:
+ return jsonify({"error": str(exc)}), 400
+ if result.get("changed"):
+ m._access_audit(f"builtin_mode_{result.get('mode')}",
+ result.get("glob") or canon, "builtin", scope,
+ project)
+ m._gw_audit("mobile.access.builtin_mode",
+ {"glob": canon, "mode": mode, "scope": scope})
+ return jsonify({**result, "loosened": False})
+
+
+@bp.route("/access/builtin/lease/revoke", methods=["POST"])
+def mobile_builtin_lease_revoke():
+ """End a temporary loosening now. Tightening, so no confirmation.
+
+ Body: ``{id}``. A lease on an unregistered project is a 404 here, the
+ same answer as an unknown one.
+ """
+ gate = m._feature_or_404("builtin") or m._feature_or_404("builtin_tighten")
+ if gate:
+ return gate
+ limited = m._security_gate()
+ if limited:
+ return limited
+ from services import builtin_leases
+ lease_id = str((request.get_json(silent=True) or {}).get("id") or "")
+ match = next((row for row in builtin_leases.active()
+ if str(row.get("id") or "") == lease_id), None) \
+ if lease_id else None
+ if match is None:
+ return jsonify({"error": "unknown or already-expired lease"}), 404
+ project = str(match.get("project_path") or "")
+ if project and m._resolve_registered(project) is None:
+ return jsonify({"error": "unknown or already-expired lease"}), 404
+ gone = builtin_leases.revoke(lease_id)
+ if gone is None:
+ return jsonify({"error": "unknown or already-expired lease"}), 404
+ m._access_audit("builtin_lease_revoke", gone.get("glob") or "", "builtin",
+ gone.get("scope") or "", gone.get("project_path") or None)
+ m._gw_audit("mobile.access.builtin_lease_revoke", {"id": lease_id})
+ return jsonify({"revoked": lease_id})
+
+
+# ── Sessions (history catalog) ────────────────────────────
+
+def _transcripts_allowed() -> bool:
+ return bool(m._cfg().get("mobile_sessions_transcripts", False))
+
+
+def _session_wire(row: dict, transcripts: bool) -> dict:
+ """One catalog row for the network.
+
+ Without ``mobile_sessions_transcripts`` the prompt text is blanked and a
+ title that IS the first prompt is withheld; resume is reduced to the
+ remote link, because launching opens a terminal on the PC.
+ """
+ out = dict(row)
+ if not transcripts:
+ for key in _TRANSCRIPT_FIELDS:
+ out[key] = ""
+ if out.get("title_source") == "first_prompt":
+ out["title"] = "(untitled)"
+ out["title_source"] = "withheld"
+ resume = row.get("resume") or {}
+ out["resume"] = {"remote_url": resume.get("remote_url") or None,
+ "can_launch": False, "why_not": _RESUME_ELSEWHERE}
+ return out
+
+
+def _sessions_project():
+ """``({name, path}, None)`` for ``?project=``, else an error response."""
+ resolved, err = m._project_or_404(str(request.args.get("project") or ""))
+ if err:
+ return None, err
+ return {"name": _name_for(resolved), "path": str(resolved)}, None
+
+
+@bp.route("/sessions/overview", methods=["GET"])
+def mobile_sessions_overview():
+ """Per-project session counts for the project rail. Stats files only."""
+ gate = m._feature_or_404("sessions")
+ if gate:
+ return gate
+ from concurrent.futures import ThreadPoolExecutor
+
+ from services import session_catalog as sc
+ transcripts = _transcripts_allowed()
+
+ def one(p):
+ row = {"name": p["name"], "path": p["path"], "initialized": True,
+ "error": None,
+ "counts": {"total": 0, "live": 0, "stale": 0, "idle": 0},
+ "last_active": "", "newest": None, "transcripts": False}
+ try:
+ row.update(sc.overview(p["path"], name=p["name"]))
+ except Exception as e:
+ row["error"] = str(e)
+ newest = row.get("newest")
+ if (not transcripts and isinstance(newest, dict)
+ and newest.get("title_source") == "first_prompt"):
+ row["newest"] = {**newest, "title": "(untitled)",
+ "title_source": "withheld"}
+ return row
+
+ with ThreadPoolExecutor(max_workers=8) as pool:
+ rows = list(pool.map(one, _registered()))
+ rows.sort(key=lambda r: r.get("last_active") or "", reverse=True)
+ return jsonify({"projects": rows, "transcripts": transcripts,
+ "features": ["list", "detail", "mark", "note", "remote"]})
+
+
+@bp.route("/sessions", methods=["GET"])
+def mobile_sessions_list():
+ """Rows for one project (``?project=``) or every project (none).
+
+ Query: ``stale`` hide|only|likely|all (default hide), ``q``, ``limit``
+ (default 50, max 200), ``before`` (ISO cursor from ``next_before``).
+ """
+ gate = m._feature_or_404("sessions")
+ if gate:
+ return gate
+ from services import session_catalog as sc
+ stale = (request.args.get("stale") or "hide").strip().lower()
+ if stale not in _STALE_FILTERS:
+ return jsonify({"error": f"stale must be one of {sorted(_STALE_FILTERS)}"}), 400
+ limit = m._limit_arg()
+ q = (request.args.get("q") or "").strip()[:200]
+ before = (request.args.get("before") or "").strip()
+ transcripts = _transcripts_allowed()
+ try:
+ if (request.args.get("project") or "").strip():
+ proj, err = _sessions_project()
+ if err:
+ return err
+ res = sc.list_sessions(proj["path"], name=proj["name"], stale=stale,
+ q=q, limit=limit, before=before)
+ res["errors"] = []
+ else:
+ res = sc.list_many(_registered(), stale=stale, q=q, limit=limit,
+ before=before)
+ except Exception as exc:
+ return m._svc_error(exc)
+ res["sessions"] = [_session_wire(r, transcripts)
+ for r in res.get("sessions") or []]
+ res["transcripts"] = transcripts
+ return jsonify(res)
+
+
+@bp.route("/sessions/detail", methods=["GET"])
+def mobile_sessions_detail():
+ """One session: the row plus decisions, the last snapshot and tasks.
+
+ Query: ``project``, ``id`` (full id or a prefix of 8+ characters). The
+ turn-by-turn preview is included only with ``mobile_sessions_transcripts``.
+ """
+ gate = m._feature_or_404("sessions")
+ if gate:
+ return gate
+ from services import session_catalog as sc
+ proj, err = _sessions_project()
+ if err:
+ return err
+ try:
+ res = sc.get_session(proj["path"], request.args.get("id") or "",
+ name=proj["name"])
+ except Exception as exc:
+ return m._svc_error(exc)
+ if "error" in res:
+ return jsonify({"error": res["error"]}), 404
+ transcripts = _transcripts_allowed()
+ out = _session_wire(res, transcripts)
+ if not transcripts:
+ out["preview"] = []
+ out["transcripts"] = transcripts
+ return jsonify(out)
+
+
+@bp.route("/sessions/mark", methods=["POST"])
+def mobile_sessions_mark():
+ """Mark a session stale / unstale, or leave a note on it.
+
+ Body: ``{project, id, op: stale|unstale|note, reason?, successor?,
+ summary?, next_steps?}``. Recorded as ``by: user`` — a person tapped it.
+ """
+ gate = m._feature_or_404("sessions") or m._feature_or_404("sessions_write")
+ if gate:
+ return gate
+ from services import session_catalog as sc
+ data = request.get_json(silent=True) or {}
+ resolved, err = m._project_or_404(str(data.get("project") or ""))
+ if err:
+ return err
+ op = str(data.get("op") or "")
+ res = sc.mark(str(resolved), str(data.get("id") or ""), op,
+ reason=str(data.get("reason") or "")[:400],
+ successor=str(data.get("successor") or ""),
+ summary=str(data.get("summary") or "")[:2000],
+ next_steps=str(data.get("next_steps") or "")[:2000],
+ by="user")
+ if "error" in res:
+ return jsonify({"error": res["error"]}), int(res.get("status") or 400)
+ m._gw_audit("mobile.sessions.mark", {"op": op})
+ row = res.get("row")
+ if isinstance(row, dict):
+ res = {**res, "row": _session_wire(row, _transcripts_allowed())}
+ return jsonify(res)
+
+
+# ── Now ───────────────────────────────────────────────────
+
+@bp.route("/now", methods=["GET"])
+def mobile_now():
+ """Live sessions, running jobs, last CI and map state across projects,
+ plus whether the Hub answers. The Desk's Now view, minus its buttons."""
+ gate = m._feature_or_404("now")
+ if gate:
+ return gate
+ from services import fleet_overview
+ jobs: list[dict] = []
+ try:
+ from services.shell_jobs import JobStore
+ jobs = [r for r in JobStore().list_all(project_path=None,
+ limit=m._SOURCE_CAP)
+ if r.get("status") in ("queued", "running")
+ and m._resolve_registered(str(r.get("project_path") or ""))
+ is not None]
+ except Exception:
+ jobs = []
+ body = fleet_overview.now(_registered(), job_rows=jobs)
+ hub_available = False
+ try:
+ import json
+ import urllib.request
+ hub_url = str(m._cfg().get("hub_url", "http://localhost:3330")).rstrip("/")
+ with urllib.request.urlopen(f"{hub_url}/api/health", timeout=2) as r:
+ hub_available = json.loads(r.read()).get("service") == "c3-hub"
+ except Exception:
+ hub_available = False
+ body["services"] = {
+ "oracle": {"up": True, "version": m._c3_version(),
+ "api_version": m.API_VERSION},
+ "hub": {"available": hub_available},
+ }
+ return jsonify(body)
diff --git a/pyproject.toml b/pyproject.toml
index 372d51d..59a1abe 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "code-context-control"
-version = "2.156.0"
+version = "2.157.0"
description = "Local MCP code-intelligence for AI coding tools: surgical search/read/edit, agent-config version history, path-level access + masking guards, and a multi-project hub."
readme = "README.md"
requires-python = ">=3.10"
diff --git a/services/fleet_overview.py b/services/fleet_overview.py
new file mode 100644
index 0000000..f95645d
--- /dev/null
+++ b/services/fleet_overview.py
@@ -0,0 +1,479 @@
+"""Cross-project read models shared by the Hub and the Oracle mobile gateway.
+
+The Hub's overview routes (Locks, Permissions, Rules, Built-in guards) and the
+gateway's (API v7) answer the same question for two audiences: C3 Desk on
+loopback and a paired phone over the tailnet. Each function here takes the
+project list its caller trusts — the Hub's registry, the gateway's scanner —
+and builds the payload, so the two surfaces cannot drift apart field by field.
+
+Every builder isolates rows: one unreadable project reports ``error`` (or
+``initialized: false``) instead of blanking the page, and never reads as the
+empty/clear state, because "no leases" or "no rules" is a claim about a repo
+we could not see.
+
+None of these touch a keyring except ``builtin`` (modes and leases are
+attested there), so the bulk loops stay cheap.
+"""
+from __future__ import annotations
+
+import json
+import time
+from pathlib import Path
+from typing import Iterable
+
+LOCKS_COVERAGE_NOTE = (
+ "Leases gate C3 tool surfaces only. A raw shell redirect, a non-Claude "
+ "agent, or a human in an editor is not covered.")
+
+ENFORCEMENT_COVERAGE_NOTE = (
+ "Tool discipline only governs whether native Edit/Write are pushed "
+ "through c3_edit. At every mode — including off — Access Guard path "
+ "rules, the credential-vault write guard, and agent locks still enforce. "
+ "The edit ledger records native writes either way; strict additionally "
+ "gets c3_edit's pre-edit snapshot.")
+
+OVERRIDE_POLICY_NOTE = (
+ "Project and global policy merge by tightening only: a layer is "
+ "escalatable iff both scopes allow it, and the numeric limits take the "
+ "smaller value.")
+
+
+def _rows(projects: Iterable[dict]) -> list[tuple[str, str]]:
+ out = []
+ for p in projects or []:
+ path = str(p.get("path") or "")
+ if path:
+ out.append((str(p.get("name") or Path(path).name), path))
+ return out
+
+
+def _initialized(path: str) -> bool:
+ return (Path(path) / ".c3").is_dir()
+
+
+# ── Locks ──────────────────────────────────────────────────────────────────
+
+def locks(projects: Iterable[dict]) -> dict:
+ """Every project's agent leases. Read-only; force-release lives elsewhere."""
+ from services import agent_locks as al
+
+ rows, total = [], 0
+ for name, ppath in _rows(projects):
+ row = {"name": name, "path": ppath, "initialized": True, "error": None,
+ "enabled": True, "mode": "advisory", "locks": [], "count": 0,
+ "fencing": 0}
+ try:
+ if not _initialized(ppath):
+ row["initialized"] = False
+ else:
+ cfg = al.config(ppath)
+ row["enabled"] = cfg["enabled"]
+ row["mode"] = cfg["mode"]
+ snap = al.store_for(ppath).snapshot()
+ row["locks"] = snap["locks"]
+ row["count"] = snap["count"]
+ row["fencing"] = snap["fencing"]
+ total += snap["count"]
+ except Exception as e:
+ row["error"] = str(e)
+ rows.append(row)
+ return {"projects": rows, "total": total,
+ "coverage_note": LOCKS_COVERAGE_NOTE}
+
+
+# ── Access Guard path rules ────────────────────────────────────────────────
+
+def access(projects: Iterable[dict]) -> dict:
+ """Each project's own path rules, plus the global scope and the builtins.
+
+ A corrupt scope is reported as corrupt: it evaluates deny-all until
+ repaired by hand, the opposite of what an empty list would suggest.
+ Builtin rules are read once, from the global realm.
+ """
+ from services import access_guard
+
+ empty = {"deny": [], "read_only": [], "confirm": [], "mask": 0,
+ "corrupt": False}
+ rows = []
+ for name, ppath in _rows(projects):
+ row = {"name": name, "path": ppath, "initialized": True,
+ "error": None, "rules": dict(empty)}
+ try:
+ if not _initialized(ppath):
+ row["initialized"] = False
+ else:
+ row["rules"] = access_guard.scope_rules("project", ppath)
+ except Exception as e:
+ row["error"] = str(e)
+ rows.append(row)
+
+ try:
+ global_rules = {**access_guard.scope_rules("global"), "error": None}
+ except Exception as e:
+ global_rules = {**empty, "error": str(e)}
+ try:
+ home = access_guard._global_base()
+ b = access_guard.list_rules(str(home) if home else ".")["builtin"]
+ builtin_rules = {"deny": list(b.get("deny") or []),
+ "read_only": list(b.get("read_only") or []),
+ "confirm": list(b.get("confirm") or []), "error": None}
+ except Exception as e:
+ builtin_rules = {"deny": [], "read_only": [], "confirm": [],
+ "error": str(e)}
+
+ return {
+ "projects": rows,
+ "global": global_rules,
+ "builtin": builtin_rules,
+ # Strictest last. "allow" is the absence of a rule, not a kind.
+ "kinds": ["confirm", "read_only", "deny"],
+ "coverage_note": access_guard.COVERAGE_MATRIX,
+ }
+
+
+# ── Tool discipline ────────────────────────────────────────────────────────
+
+def enforcement(projects: Iterable[dict]) -> dict:
+ """Each project's effective discipline mode, its source, and denials.
+
+ A project we could not read reports ``error`` rather than a mode —
+ claiming 'strict' for a repo we failed to read would be a guess.
+ """
+ from services import access_telemetry as at
+ from services import enforcement_policy as ep
+
+ rows = []
+ totals = {"discipline": 0, "access": 0}
+ for name, ppath in _rows(projects):
+ row = {"name": name, "path": ppath, "initialized": True,
+ "error": None, "mode": None, "scope": "", "set_by": "",
+ "signal_ttl_s": ep.DEFAULT_SIGNAL_TTL_S, "warnings": [],
+ "tier": "", "tier_implies": "", "denials": {}, "denial_total": 0}
+ try:
+ if not _initialized(ppath):
+ row["initialized"] = False
+ else:
+ policy = ep.resolve(ppath)
+ row["mode"] = policy.mode
+ row["scope"] = policy.scope
+ row["set_by"] = policy.set_by
+ row["signal_ttl_s"] = policy.signal_ttl_s
+ row["warnings"] = list(policy.warnings)
+ row["blocked_tools"] = sorted(policy.blocked_tools)
+
+ cfg_path = Path(ppath) / ".c3" / "config.json"
+ try:
+ cfg = json.loads(cfg_path.read_text(encoding="utf-8"))
+ tier = str(cfg.get("permission_tier") or "")
+ except Exception:
+ tier = ""
+ row["tier"] = tier
+ row["tier_implies"] = ep.derive_from_tier(tier) if tier else ""
+
+ agg = at.aggregate(ppath)
+ row["denials"] = agg["by_layer"]
+ row["denial_total"] = agg["total"]
+ row["top_denials"] = [
+ {**r, "fix": at.suggest(r)} for r in agg["rows"][:5]
+ ]
+ for layer, count in agg["by_layer"].items():
+ totals[layer] = totals.get(layer, 0) + count
+ except Exception as e:
+ row["error"] = str(e)
+ rows.append(row)
+
+ # The global (~/.c3) section. A corrupt global config must not blank the
+ # page — report the error inside the object.
+ try:
+ g = ep.resolve_global()
+ global_policy = {
+ "configured": g.scope == "global",
+ "mode": g.mode or None,
+ "set_by": g.set_by,
+ "signal_ttl_s": g.signal_ttl_s,
+ "blocked_tools": sorted(g.blocked_tools),
+ "warnings": list(g.warnings),
+ }
+ except Exception as e:
+ global_policy = {"configured": False, "mode": None, "error": str(e)}
+
+ return {
+ "projects": rows,
+ "modes": [{"id": m, "help": ep.MODE_HELP[m]} for m in ep.MODES],
+ "default_mode": ep.DEFAULT_MODE,
+ "tier_map": ep.TIER_TO_MODE,
+ "totals": totals,
+ "global_policy": global_policy,
+ # Stated so a client can never imply that turning discipline down
+ # also turns the security boundaries down. Mirrors docs/enforcement.md.
+ "coverage_note": ENFORCEMENT_COVERAGE_NOTE,
+ }
+
+
+# ── Override policy ────────────────────────────────────────────────────────
+
+def override_policy(projects: Iterable[dict]) -> dict:
+ """Each project's effective override policy, plus the global scope.
+
+ ``project_keys`` names what the project pins itself, so a screen can tell
+ an inherited value from a chosen one. The global scope is READ-ONLY on
+ every surface: ``write_section`` is project-scope by design.
+ """
+ from services import override_policy as opol
+
+ rows = []
+ for name, ppath in _rows(projects):
+ row = {"name": name, "path": ppath, "initialized": True,
+ "error": None, "policy": None, "configured": False,
+ "corrupt": False, "project_keys": []}
+ try:
+ if not _initialized(ppath):
+ row["initialized"] = False
+ else:
+ keys, present, corrupt = opol.scope_keys(ppath)
+ row["project_keys"] = keys
+ row["configured"] = present
+ row["corrupt"] = corrupt
+ row["policy"] = opol.resolve(ppath).as_dict()
+ except Exception as e:
+ row["error"] = str(e)
+ rows.append(row)
+
+ try:
+ keys, present, corrupt = opol.global_scope_keys()
+ global_policy = {"configured": present, "corrupt": corrupt,
+ "keys": keys, "error": None,
+ "policy": opol.resolve_global().as_dict()}
+ except Exception as e:
+ global_policy = {"configured": False, "corrupt": False, "keys": [],
+ "error": str(e), "policy": None}
+
+ return {
+ "projects": rows,
+ "global_policy": global_policy,
+ "defaults": {k: v for k, v in opol.DEFAULTS.items()
+ if k != opol.WAKE_KEY},
+ "layers": list(opol.LAYER_KEYS),
+ "typed_confirm_layers": sorted(opol.TYPED_CONFIRM_LAYERS),
+ "hard_max_ttl_s": opol.HARD_MAX_TTL_S,
+ "coverage_note": OVERRIDE_POLICY_NOTE,
+ # What the surface can do beyond the per-project GET/POST, so a client
+ # built against a newer C3 can hide what an older one cannot serve.
+ "features": ["clear"],
+ }
+
+
+# ── Built-in (Tier-1) guards ───────────────────────────────────────────────
+
+def builtin(projects: Iterable[dict]) -> dict:
+ """Every Tier-1 builtin with its live global mode, each project's own
+ modes, and the live temporary leases.
+
+ Modes are read through the keyring (config and attestation must agree),
+ so a mode listed here is one evaluation honours. ``projects`` holds only
+ projects that set a mode of their own, or could not be read.
+ ``strictness`` ranks each mode 0 (allow) to 3 (full deny), so a client
+ can tell loosening from tightening without knowing the tiers.
+
+ Raises when the global realm cannot be read — the caller answers 500
+ rather than presenting every guard at its default.
+ """
+ from services import access_guard, builtin_leases
+
+ global_modes = access_guard._scope_modes("global")
+ home = access_guard._global_base()
+ guards = []
+ for glob in access_guard.DISABLEABLE_BUILTINS:
+ canon = access_guard._norm_builtin(glob)
+ guards.append({
+ "glob": canon,
+ "global": global_modes.get(canon, "default"),
+ "strictness": {m: access_guard.builtin_strictness(canon, m)
+ for m in (*access_guard.BUILTIN_MODES, "default")},
+ })
+ rows = []
+ for name, ppath in _rows(projects):
+ row = {"name": name, "path": ppath, "modes": {}, "error": None}
+ try:
+ base = Path(ppath).resolve()
+ if (base / ".c3").is_dir() and base != home:
+ row["modes"] = access_guard._scope_modes("project", ppath)
+ except Exception as e:
+ row["error"] = str(e)
+ if row["modes"] or row["error"]:
+ rows.append(row)
+ return {"guards": guards, "projects": rows,
+ "leases": builtin_leases.active()}
+
+
+# ── Now: what is running across the fleet ──────────────────────────────────
+
+_STATS_TAIL_BYTES = 256 * 1024
+_CI_MAX_PROJECTS = 8
+
+
+def _tail_jsonl(path: Path, max_bytes: int = _STATS_TAIL_BYTES) -> list[dict]:
+ """The parsed rows of the last ``max_bytes`` of a JSONL file."""
+ try:
+ size = path.stat().st_size
+ with open(path, "rb") as fh:
+ if size > max_bytes:
+ fh.seek(size - max_bytes)
+ fh.readline() # drop the partial first line
+ block = fh.read()
+ except OSError:
+ return []
+ out = []
+ for raw in block.splitlines():
+ try:
+ row = json.loads(raw)
+ except Exception:
+ continue
+ if isinstance(row, dict):
+ out.append(row)
+ return out
+
+
+def _newest_stats(ppath: str) -> dict[str, dict]:
+ """Newest session_stats row per host session id.
+
+ Rows are cumulative per session, so the newest one IS the total — they
+ are never summed.
+ """
+ newest: dict[str, dict] = {}
+ for row in _tail_jsonl(Path(ppath) / ".c3" / "session_stats.jsonl"):
+ sid = str(row.get("session_id") or "")
+ if sid:
+ newest[sid] = row
+ return newest
+
+
+def _last_c3_call(ppath: str) -> tuple[str, dict]:
+ """``(host_session_id, {ts, tool})`` from enforcement_state.json.
+
+ Only the session id and the last call are read; the file's unlocked-path
+ list stays where it is.
+ """
+ try:
+ data = json.loads((Path(ppath) / ".c3" / "enforcement_state.json")
+ .read_text(encoding="utf-8"))
+ except Exception:
+ return "", {}
+ call = data.get("last_c3_call") if isinstance(data, dict) else None
+ if not isinstance(call, dict):
+ return "", {}
+ return (str(data.get("session_id") or ""),
+ {"ts": str(call.get("ts") or ""), "tool": str(call.get("tool") or "")})
+
+
+def _map_state(ppath: str, now: float) -> dict:
+ c3 = Path(ppath) / ".c3"
+ meta = c3 / "map.meta.json"
+ if not meta.exists():
+ return {"state": "missing", "generated_at": None, "age_s": None}
+ generated_at, age = None, None
+ try:
+ data = json.loads(meta.read_text(encoding="utf-8"))
+ generated_at = data.get("generated_at") or None
+ except Exception:
+ pass
+ if generated_at:
+ try:
+ from datetime import datetime
+ ts = datetime.fromisoformat(str(generated_at).replace("Z", "+00:00"))
+ age = max(0, int(now - ts.timestamp()))
+ except Exception:
+ age = None
+ state = "dirty" if (c3 / "map.dirty").exists() else (
+ "fresh" if generated_at else "unknown")
+ return {"state": state, "generated_at": generated_at, "age_s": age}
+
+
+def _session_row(beat: dict, stats: dict[str, dict], last_sid: str,
+ last_call: dict) -> dict:
+ host = str(beat.get("host_session_id") or "")
+ c3id = str(beat.get("session_id") or "")
+ st = stats.get(host) or stats.get(c3id) or {}
+ tool = last_call if last_call and last_sid and last_sid in (host, c3id) else {}
+ tokens_in = int(st.get("input_tokens") or 0)
+ tokens_out = int(st.get("output_tokens") or 0)
+ return {
+ "session_id": c3id,
+ "host_session_id": host,
+ "short": (host or c3id)[:8],
+ "ide": str(beat.get("ide") or ""),
+ "heartbeat_age_s": round(float(beat.get("age_seconds") or 0.0), 1),
+ "last_tool": tool.get("tool") or None,
+ "last_tool_at": tool.get("ts") or None,
+ "model": st.get("model") or None,
+ "tokens_in": tokens_in,
+ "tokens_out": tokens_out,
+ "tokens_total": tokens_in + tokens_out,
+ "cost_usd": st.get("cost_usd"),
+ }
+
+
+def now(projects: Iterable[dict], *, job_rows: list[dict] | None = None,
+ ci_runs=None) -> dict:
+ """Live sessions, running jobs, last CI and map state, across projects.
+
+ Liveness is the MCP heartbeat under ``.c3/live`` (services.session_live),
+ never an inference from activity rows. Only projects with a live session
+ or a running job are listed in full; every project still counts toward
+ the totals. ``job_rows`` is the caller's already-filtered job list (the
+ store is shared machine-wide); ``ci_runs(path, limit)`` is injectable for
+ tests and asked for at most eight projects.
+ """
+ from services import session_live
+
+ if ci_runs is None:
+ from services.ci_runner import list_runs as ci_runs
+
+ t = time.time()
+ running_by_project: dict[str, list[dict]] = {}
+ for job in job_rows or []:
+ if job.get("status") in ("queued", "running"):
+ key = str(Path(str(job.get("project_path") or "")).resolve()).lower()
+ running_by_project.setdefault(key, []).append({
+ "id": job.get("id"), "status": job.get("status"),
+ "cmd_display": job.get("cmd_display"),
+ "started_at": job.get("started_at") or job.get("created_at"),
+ })
+
+ rows, errors = [], []
+ live_total = running_total = 0
+ for name, ppath in _rows(projects):
+ if not _initialized(ppath):
+ continue
+ try:
+ beats = session_live.live_sessions(ppath, prune=False)
+ except Exception as e:
+ errors.append({"path": ppath, "error": str(e)})
+ beats = []
+ try:
+ jobs = running_by_project.get(str(Path(ppath).resolve()).lower(), [])
+ except OSError:
+ jobs = []
+ if not beats and not jobs:
+ continue
+ stats = _newest_stats(ppath) if beats else {}
+ last_sid, last_call = _last_c3_call(ppath) if beats else ("", {})
+ sessions = [_session_row(b, stats, last_sid, last_call) for b in beats]
+ live_total += len(sessions)
+ running_total += len(jobs)
+ rows.append({"name": name, "path": ppath, "sessions": sessions,
+ "running_jobs": jobs, "map": _map_state(ppath, t),
+ "ci": None})
+
+ rows.sort(key=lambda r: (-len(r["sessions"]), -len(r["running_jobs"]),
+ r["name"].lower()))
+ for row in rows[:_CI_MAX_PROJECTS]:
+ try:
+ runs = ci_runs(row["path"], limit=1) or []
+ row["ci"] = runs[0] if runs else None
+ except Exception as e:
+ errors.append({"path": row["path"], "error": f"ci: {e}"})
+
+ return {"generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(t)),
+ "projects": rows, "live_sessions": live_total,
+ "running_jobs": running_total, "errors": errors}
diff --git a/tests/test_mobile_clients.py b/tests/test_mobile_clients.py
index 8fc35d5..3c95c95 100644
--- a/tests/test_mobile_clients.py
+++ b/tests/test_mobile_clients.py
@@ -352,7 +352,7 @@ def test_client_token_authenticates_and_info_names_the_principal(self):
r = self.client.get("/api/mobile/info", headers=self.bearer(tok))
self.assertEqual(r.status_code, 200)
body = r.get_json()
- self.assertEqual(body["api_version"], 6)
+ self.assertEqual(body["api_version"], 7)
self.assertIn("clients", body["capabilities"])
self.assertEqual(body["client"], {"kind": "desk", "client_id": cid})
diff --git a/tests/test_mobile_fleet.py b/tests/test_mobile_fleet.py
new file mode 100644
index 0000000..403029e
--- /dev/null
+++ b/tests/test_mobile_fleet.py
@@ -0,0 +1,536 @@
+"""Gateway API v7: the fleet views C3 Desk reads from the Hub, for the phone.
+
+What these pin, beyond "the route answers":
+
+- every view is cross-project but never names a project the scanner does not
+ register, and goes dark (404 + gone from /info) with its subsystem switch;
+- what the phone may change is narrower than Desk: built-in guards only
+ tighten (a looser mode is a 403 for EVERY token kind, Desk's included),
+ "inherit" asks for the typed word when it loosens, locks stay read-only;
+- the session catalog carries no transcript text unless
+ mobile_sessions_transcripts is on, and never offers to resume on the PC;
+- /now reports heartbeats and tokens without leaking the unlocked-path list
+ that sits in the same enforcement_state.json.
+"""
+from __future__ import annotations
+
+import json
+import os
+import sys
+import tempfile
+import time
+import types
+import unittest
+from pathlib import Path
+from unittest import mock
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO_ROOT))
+
+os.environ.setdefault("C3_ORACLE_API_KEY", "mobile-fleet-key")
+
+import oracle.oracle_server as srv # noqa: E402
+from oracle.services import client_tokens, mobile_api # noqa: E402
+from services import access_guard as ag # noqa: E402
+from services import enforcement_policy as ep # noqa: E402
+from services import fleet_overview, session_live # noqa: E402
+from services import override_policy as opol # noqa: E402
+from services import session_catalog as sc # noqa: E402
+
+ENV = "**/.env*"
+GIT = "**/.git/**"
+
+
+class _StubScanner:
+ def __init__(self, projects):
+ self.projects = projects
+
+ def discover(self, force=False):
+ return [dict(p) for p in self.projects]
+
+
+class _FakeKeyring:
+ def __init__(self):
+ self.store = {}
+
+ def set_password(self, service, account, value):
+ self.store[(service, account)] = value
+
+ def get_password(self, service, account):
+ return self.store.get((service, account))
+
+
+def _entry(path, has_c3=True):
+ return {"path": str(path), "name": Path(path).name, "tags": [],
+ "active": False, "has_c3": has_c3, "fact_count": 0}
+
+
+class FleetBase(unittest.TestCase):
+ """Two registered projects, one ghost the scanner never lists, an isolated
+ home (global realm, keyring) and every switch at its shipped default."""
+
+ def setUp(self):
+ self._tmp = tempfile.TemporaryDirectory()
+ root = Path(self._tmp.name)
+ self.home = root / "home"
+ (self.home / ".c3").mkdir(parents=True)
+ self.proj = root / "proj"
+ self.other = root / "other"
+ self.ghost = root / "ghost"
+ for p in (self.proj, self.other, self.ghost):
+ (p / ".c3").mkdir(parents=True)
+
+ fake = _FakeKeyring()
+ keyring = types.ModuleType("keyring")
+ keyring.set_password, keyring.get_password = fake.set_password, fake.get_password
+ self._patches = [
+ mock.patch.dict(sys.modules, {"keyring": keyring}),
+ mock.patch.object(ag, "_global_base", return_value=self.home),
+ mock.patch.object(opol, "_global_base", return_value=self.home),
+ mock.patch.dict(os.environ, {"C3_HOME": str(self.home)}),
+ ]
+ for p in self._patches:
+ p.start()
+
+ self._prior_cfg = srv._cfg
+ srv._cfg = {
+ "mobile_api_enabled": True,
+ "api_rate_limit_per_min": 0,
+ "mobile_security_rate_limit_per_min": 0,
+ "api_audit_enabled": False,
+ "mobile_access_enabled": True,
+ "mobile_access_write": True,
+ "mobile_access_global_scope": False,
+ "mobile_override_enabled": True,
+ "mobile_override_write": True,
+ "mobile_builtin_enabled": True,
+ "mobile_builtin_tighten": True,
+ "mobile_sessions_enabled": True,
+ "mobile_sessions_write": True,
+ "mobile_sessions_transcripts": False,
+ "hub_url": "http://127.0.0.1:9",
+ }
+ mobile_api._sec_limiter = None
+ mobile_api._sec_limiter_key = None
+ mobile_api.init_services(scanner=_StubScanner(
+ [_entry(self.proj), _entry(self.other)]))
+ srv.app.config["TESTING"] = True
+ self.client = srv.app.test_client()
+ self.auth = {"Authorization": "Bearer " + os.environ["C3_ORACLE_API_KEY"]}
+
+ def tearDown(self):
+ srv._cfg = self._prior_cfg
+ for p in reversed(self._patches):
+ p.stop()
+ self._tmp.cleanup()
+
+ def get(self, path, **params):
+ r = self.client.get(path, headers=self.auth, query_string=params)
+ return r.status_code, r.get_json()
+
+ def post(self, path, body, headers=None):
+ r = self.client.post(path, json=body, headers=headers or self.auth)
+ return r.status_code, r.get_json()
+
+ def paths(self, rows):
+ return {str(Path(r["path"]).resolve()) for r in rows}
+
+
+class TestCapabilities(FleetBase):
+ NEW = ("locks_overview", "access_overview", "enforcement_overview",
+ "enforcement_clear", "override_policy_overview",
+ "override_policy_clear", "builtin", "builtin_tighten",
+ "sessions", "sessions_write", "now")
+
+ def test_v7_advertises_every_fleet_capability(self):
+ _, info = self.get("/api/mobile/info")
+ self.assertEqual(info["api_version"], 7)
+ for cap in self.NEW:
+ self.assertIn(cap, info["capabilities"])
+
+ def test_each_view_goes_dark_with_its_subsystem(self):
+ srv._cfg["mobile_access_enabled"] = False
+ srv._cfg["mobile_override_enabled"] = False
+ srv._cfg["mobile_builtin_enabled"] = False
+ srv._cfg["mobile_sessions_enabled"] = False
+ _, info = self.get("/api/mobile/info")
+ for cap in ("access_overview", "enforcement_overview",
+ "override_policy_overview", "builtin", "sessions"):
+ self.assertNotIn(cap, info["capabilities"])
+ for path in ("/api/mobile/access/overview",
+ "/api/mobile/enforcement/overview",
+ "/api/mobile/overrides/policy/overview",
+ "/api/mobile/access/builtin",
+ "/api/mobile/sessions/overview"):
+ self.assertEqual(self.get(path)[0], 404, path)
+ # Ops reads like /locks and /status have no switch.
+ self.assertIn("locks_overview", info["capabilities"])
+ self.assertIn("now", info["capabilities"])
+
+ def test_every_new_route_needs_the_bearer(self):
+ for path in ("/api/mobile/locks/overview", "/api/mobile/now",
+ "/api/mobile/sessions", "/api/mobile/access/builtin"):
+ self.assertEqual(self.client.get(path).status_code, 401, path)
+
+
+class TestLocksOverview(FleetBase):
+ def test_lists_registered_projects_and_their_leases_only(self):
+ from services import agent_locks as al
+ al.store_for(str(self.proj)).acquire(["src/a.py"], agent_id="agent-1",
+ session_id="s1", intent="refactor")
+ al.store_for(str(self.ghost)).acquire(["x.py"], agent_id="agent-2",
+ session_id="s2")
+ status, body = self.get("/api/mobile/locks/overview")
+ self.assertEqual(status, 200)
+ self.assertEqual(self.paths(body["projects"]),
+ {str(self.proj.resolve()), str(self.other.resolve())})
+ self.assertEqual(body["total"], 1)
+ row = next(r for r in body["projects"] if r["count"])
+ self.assertEqual(row["locks"][0]["agent_id"], "agent-1")
+ self.assertIn("coverage_note", body)
+
+ def test_there_is_no_force_release_on_the_phone(self):
+ rules = [str(r) for r in srv.app.url_map.iter_rules()
+ if str(r).startswith("/api/mobile/")]
+ self.assertFalse([r for r in rules
+ if "force-release" in r or r.endswith("/release")])
+
+
+class TestAccessOverview(FleetBase):
+ def test_project_rules_and_writable_scopes(self):
+ ag.set_rule("**/*.pem", "deny", "project", str(self.proj))
+ status, body = self.get("/api/mobile/access/overview")
+ self.assertEqual(status, 200)
+ row = next(r for r in body["projects"]
+ if Path(r["path"]).resolve() == self.proj.resolve())
+ self.assertIn("**/*.pem", row["rules"]["deny"])
+ self.assertEqual(body["writable"], {"project": True, "global": False})
+ srv._cfg["mobile_access_global_scope"] = True
+ self.assertTrue(self.get("/api/mobile/access/overview")[1]
+ ["writable"]["global"])
+ srv._cfg["mobile_access_write"] = False
+ self.assertEqual(self.get("/api/mobile/access/overview")[1]["writable"],
+ {"project": False, "global": False})
+
+
+class TestEnforcement(FleetBase):
+ def test_overview_reports_each_project_and_never_global_writable(self):
+ ep.set_mode("advisory", str(self.proj), set_by=ep.SET_BY_USER,
+ scope="project")
+ status, body = self.get("/api/mobile/enforcement/overview")
+ self.assertEqual(status, 200)
+ modes = {Path(r["path"]).name: r["mode"] for r in body["projects"]}
+ self.assertEqual(modes["proj"], "advisory")
+ self.assertFalse(body["writable"]["global"])
+
+ def test_clear_inherits_again(self):
+ ep.set_mode("advisory", str(self.proj), set_by=ep.SET_BY_USER,
+ scope="project")
+ status, body = self.post("/api/mobile/enforcement/clear",
+ {"project": str(self.proj)})
+ self.assertEqual(status, 200, body)
+ self.assertTrue(body["cleared"])
+ self.assertEqual(ep.resolve(str(self.proj)).mode or ep.DEFAULT_MODE,
+ body["mode"])
+
+ def test_clear_that_would_inherit_off_needs_the_word(self):
+ ep.set_mode("off", ".", set_by=ep.SET_BY_USER, scope="global")
+ ep.set_mode("strict", str(self.proj), set_by=ep.SET_BY_USER,
+ scope="project")
+ status, body = self.post("/api/mobile/enforcement/clear",
+ {"project": str(self.proj)})
+ self.assertEqual(status, 400)
+ self.assertEqual(body["confirm_with"], "off")
+ self.assertEqual(ep.resolve(str(self.proj)).mode, "strict")
+ status, body = self.post("/api/mobile/enforcement/clear",
+ {"project": str(self.proj), "confirm": "off"})
+ self.assertEqual(status, 200)
+ self.assertEqual(body["mode"], "off")
+
+ def test_clear_refuses_an_unregistered_project(self):
+ status, _ = self.post("/api/mobile/enforcement/clear",
+ {"project": str(self.ghost)})
+ self.assertEqual(status, 404)
+
+
+class TestOverridePolicy(FleetBase):
+ def test_overview_and_clear(self):
+ opol.write_section(self.proj, {"enabled": False})
+ status, body = self.get("/api/mobile/overrides/policy/overview")
+ self.assertEqual(status, 200)
+ row = next(r for r in body["projects"]
+ if Path(r["path"]).resolve() == self.proj.resolve())
+ self.assertIn("enabled", row["project_keys"])
+ self.assertFalse(body["writable"]["global"])
+ self.assertNotIn("wake", body["defaults"])
+
+ status, body = self.post("/api/mobile/overrides/policy/clear",
+ {"project": str(self.proj)})
+ self.assertEqual(status, 200, body)
+ self.assertTrue(body["cleared"])
+
+ def test_clear_that_widens_needs_widen(self):
+ # Global turns the feature on; the project had pinned it off.
+ gcfg = self.home / ".c3" / "config.json"
+ gcfg.write_text(json.dumps({"override": {"enabled": True}}),
+ encoding="utf-8")
+ opol.write_section(self.proj, {"enabled": False})
+ self.assertTrue(opol.resolve_global().enabled)
+ status, body = self.post("/api/mobile/overrides/policy/clear",
+ {"project": str(self.proj)})
+ self.assertEqual(status, 400, body)
+ self.assertEqual(body.get("confirm_with"), "widen")
+ status, body = self.post("/api/mobile/overrides/policy/clear",
+ {"project": str(self.proj), "confirm": "widen"})
+ self.assertEqual(status, 200, body)
+
+
+class TestBuiltinGuards(FleetBase):
+ def mode(self, glob, scope="project"):
+ modes = (ag._scope_modes("global") if scope == "global"
+ else ag.effective_builtin_modes(str(self.proj)))
+ return modes.get(glob, "default")
+
+ def test_read_lists_guards_and_says_loosen_is_never_offered(self):
+ status, body = self.get("/api/mobile/access/builtin")
+ self.assertEqual(status, 200)
+ globs = {g["glob"] for g in body["guards"]}
+ self.assertIn(ENV, globs)
+ self.assertFalse(body["writable"]["loosen"])
+ self.assertFalse(body["writable"]["tighten_global"])
+
+ def test_tightening_a_project_guard_applies(self):
+ status, body = self.post("/api/mobile/access/builtin/mode", {
+ "scope": "project", "project": str(self.proj),
+ "glob": GIT, "mode": "deny"})
+ self.assertEqual(status, 200, body)
+ self.assertFalse(body["loosened"])
+ self.assertEqual(self.mode(GIT), "deny")
+
+ def test_loosening_is_refused_for_every_token_kind(self):
+ desk = {"Authorization": "Bearer desk-token"}
+ with mock.patch.object(client_tokens, "principal_for",
+ side_effect=lambda t: {"kind": "desk",
+ "client_id": "desk-1"}
+ if t == "desk-token" else None):
+ for headers in (self.auth, desk):
+ status, body = self.post("/api/mobile/access/builtin/mode", {
+ "scope": "project", "project": str(self.proj),
+ "glob": ENV, "mode": "allow", "confirm": ENV}, headers)
+ self.assertEqual(status, 403, body)
+ self.assertTrue(body["needs_desk"])
+ self.assertIn("c3 access builtin mode", body["command"])
+ self.assertEqual(self.mode(ENV), "default")
+
+ def test_global_scope_needs_the_global_switch_even_to_tighten(self):
+ status, _ = self.post("/api/mobile/access/builtin/mode", {
+ "scope": "global", "glob": GIT, "mode": "deny"})
+ self.assertEqual(status, 403)
+ srv._cfg["mobile_access_global_scope"] = True
+ status, body = self.post("/api/mobile/access/builtin/mode", {
+ "scope": "global", "glob": GIT, "mode": "deny"})
+ self.assertEqual(status, 200, body)
+ self.assertEqual(self.mode(GIT, "global"), "deny")
+
+ def test_tighten_switch_off_leaves_the_read(self):
+ srv._cfg["mobile_builtin_tighten"] = False
+ self.assertEqual(self.get("/api/mobile/access/builtin")[0], 200)
+ status, _ = self.post("/api/mobile/access/builtin/mode", {
+ "scope": "project", "project": str(self.proj),
+ "glob": GIT, "mode": "deny"})
+ self.assertEqual(status, 404)
+
+ def test_ending_a_lease_early(self):
+ from services import builtin_leases
+ lease = builtin_leases.mint(ENV, "confirm", scope="project",
+ project_path=str(self.proj), ttl_s=600,
+ created_by="desk-1")
+ status, body = self.get("/api/mobile/access/builtin")
+ self.assertIn(lease["id"], {row["id"] for row in body["leases"]})
+ status, body = self.post("/api/mobile/access/builtin/lease/revoke",
+ {"id": lease["id"]})
+ self.assertEqual(status, 200, body)
+ self.assertEqual(body["revoked"], lease["id"])
+ status, _ = self.post("/api/mobile/access/builtin/lease/revoke",
+ {"id": lease["id"]})
+ self.assertEqual(status, 404)
+
+ def test_a_lease_on_an_unregistered_project_is_not_revocable_here(self):
+ from services import builtin_leases
+ lease = builtin_leases.mint(ENV, "confirm", scope="project",
+ project_path=str(self.ghost), ttl_s=600,
+ created_by="desk-1")
+ status, _ = self.post("/api/mobile/access/builtin/lease/revoke",
+ {"id": lease["id"]})
+ self.assertEqual(status, 404)
+ self.assertIn(lease["id"], {r["id"] for r in builtin_leases.active()})
+
+
+def _row(project, **over):
+ row = {
+ "id": "11111111-2222-3333-4444-555555555555", "short": "11111111",
+ "provider": "claude", "project": {"name": project.name,
+ "path": str(project)},
+ "title": "fix the parser", "title_source": "ai",
+ "first_prompt": "my first prompt", "last_prompt": "my last prompt",
+ "started": "2026-10-14T10:00:00+00:00",
+ "last_active": "2026-10-14T11:00:00+00:00", "branch": "main",
+ "live": False, "stale": None, "hints": [], "note": None,
+ "links": {"c3_sessions": [], "snapshots": 0, "tasks": 0,
+ "decisions": 0, "predecessor": "", "successor": ""},
+ "resume": {"command": "claude --resume 1111", "cwd": str(project),
+ "remote_url": "https://claude.ai/code/session_abc",
+ "can_launch": True, "why_not": ""},
+ }
+ row.update(over)
+ return row
+
+
+class TestSessions(FleetBase):
+ def test_list_withholds_transcript_text_by_default(self):
+ rows = [_row(self.proj),
+ _row(self.proj, id="9" * 36, title="my first prompt",
+ title_source="first_prompt")]
+ with mock.patch.object(sc, "_build", return_value=(rows, {})):
+ status, body = self.get("/api/mobile/sessions",
+ project=str(self.proj))
+ self.assertEqual(status, 200, body)
+ self.assertFalse(body["transcripts"])
+ for row in body["sessions"]:
+ self.assertEqual(row["first_prompt"], "")
+ self.assertEqual(row["last_prompt"], "")
+ self.assertEqual(set(row["resume"]),
+ {"remote_url", "can_launch", "why_not"})
+ self.assertFalse(row["resume"]["can_launch"])
+ titles = {r["title_source"]: r["title"] for r in body["sessions"]}
+ self.assertEqual(titles["ai"], "fix the parser")
+ self.assertEqual(titles["withheld"], "(untitled)")
+ self.assertEqual(body["sessions"][0]["resume"]["remote_url"],
+ "https://claude.ai/code/session_abc")
+
+ def test_the_switch_lets_prompts_through(self):
+ srv._cfg["mobile_sessions_transcripts"] = True
+ with mock.patch.object(sc, "_build",
+ return_value=([_row(self.proj)], {})):
+ _, body = self.get("/api/mobile/sessions", project=str(self.proj))
+ self.assertTrue(body["transcripts"])
+ self.assertEqual(body["sessions"][0]["first_prompt"], "my first prompt")
+
+ def test_detail_drops_the_preview_by_default(self):
+ detail = dict(_row(self.proj), decisions=[], snapshot=None, tasks=[],
+ preview=[{"role": "user", "text": "secret turn"}])
+ with mock.patch.object(sc, "get_session", return_value=detail):
+ status, body = self.get("/api/mobile/sessions/detail",
+ project=str(self.proj), id="11111111")
+ self.assertEqual(status, 200)
+ self.assertEqual(body["preview"], [])
+ self.assertNotIn("secret turn", json.dumps(body))
+
+ def test_unregistered_project_is_a_404(self):
+ status, _ = self.get("/api/mobile/sessions", project=str(self.ghost))
+ self.assertEqual(status, 404)
+
+ def test_bad_stale_filter_is_a_400(self):
+ self.assertEqual(self.get("/api/mobile/sessions", stale="nope")[0], 400)
+
+ def test_mark_records_a_user_mark(self):
+ seen = {}
+
+ def fake_mark(project, ref, op, **kw):
+ seen.update(project=project, ref=ref, op=op, **kw)
+ return {"mark": {"op": op, "by": kw["by"]},
+ "row": _row(self.proj), "warning": ""}
+
+ with mock.patch.object(sc, "mark", side_effect=fake_mark):
+ status, body = self.post("/api/mobile/sessions/mark", {
+ "project": str(self.proj), "id": "11111111", "op": "stale",
+ "reason": "superseded"})
+ self.assertEqual(status, 200, body)
+ self.assertEqual(seen["by"], "user")
+ self.assertEqual(seen["op"], "stale")
+ self.assertEqual(body["row"]["first_prompt"], "")
+
+ def test_mark_switch_off_is_a_404(self):
+ srv._cfg["mobile_sessions_write"] = False
+ status, _ = self.post("/api/mobile/sessions/mark", {
+ "project": str(self.proj), "id": "11111111", "op": "stale"})
+ self.assertEqual(status, 404)
+
+ def test_overview_counts_registered_projects(self):
+ with mock.patch.object(sc, "overview", return_value={
+ "counts": {"total": 3, "live": 1, "stale": 0, "idle": 1},
+ "last_active": "2026-10-14T11:00:00+00:00",
+ "newest": {"id": "x", "title": "my first prompt",
+ "title_source": "first_prompt"},
+ "transcripts": True}):
+ status, body = self.get("/api/mobile/sessions/overview")
+ self.assertEqual(status, 200)
+ self.assertEqual(self.paths(body["projects"]),
+ {str(self.proj.resolve()), str(self.other.resolve())})
+ self.assertEqual(body["projects"][0]["newest"]["title"], "(untitled)")
+ self.assertNotIn("resume", body["features"])
+
+
+class TestNow(FleetBase):
+ def test_live_session_with_tokens_and_last_tool(self):
+ host = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
+ session_live.beat(str(self.proj), "c3-session-1", host_session_id=host,
+ ide="claude-code")
+ c3 = self.proj / ".c3"
+ (c3 / "session_stats.jsonl").write_text(
+ json.dumps({"session_id": host, "input_tokens": 10,
+ "output_tokens": 100, "model": "m1"}) + "\n"
+ + json.dumps({"session_id": host, "input_tokens": 20,
+ "output_tokens": 300, "model": "m1"}) + "\n",
+ encoding="utf-8")
+ (c3 / "enforcement_state.json").write_text(json.dumps({
+ "session_id": host,
+ "last_c3_call": {"ts": "2026-10-14T12:00:00+00:00",
+ "tool": "c3_shell"},
+ "unlocked_files": {"secret/path.py": ["edit"]}}), encoding="utf-8")
+ (c3 / "map.meta.json").write_text(json.dumps(
+ {"generated_at": "2026-10-14T11:00:00+00:00"}), encoding="utf-8")
+
+ status, body = self.get("/api/mobile/now")
+ self.assertEqual(status, 200, body)
+ self.assertEqual(body["live_sessions"], 1)
+ self.assertNotIn("secret/path.py", json.dumps(body))
+ project = body["projects"][0]
+ self.assertEqual(Path(project["path"]).resolve(), self.proj.resolve())
+ s = project["sessions"][0]
+ self.assertEqual(s["host_session_id"], host)
+ self.assertEqual(s["last_tool"], "c3_shell")
+ # Newest row, never a sum: rows are cumulative per session.
+ self.assertEqual(s["tokens_out"], 300)
+ self.assertEqual(s["tokens_total"], 320)
+ self.assertEqual(project["map"]["state"], "fresh")
+ self.assertEqual(body["services"]["oracle"]["api_version"], 7)
+ self.assertFalse(body["services"]["hub"]["available"])
+
+ def test_quiet_projects_and_ghosts_are_not_listed(self):
+ session_live.beat(str(self.ghost), "ghost-session")
+ status, body = self.get("/api/mobile/now")
+ self.assertEqual(status, 200)
+ self.assertEqual(body["projects"], [])
+ self.assertEqual(body["live_sessions"], 0)
+
+ def test_running_jobs_list_their_project(self):
+ job = {"id": "j1", "project_path": str(self.other), "status": "running",
+ "cmd_display": "pytest", "created_at": "2026-10-14T12:00:00Z"}
+ body = fleet_overview.now([{"name": "other", "path": str(self.other)}],
+ job_rows=[job], ci_runs=lambda p, limit: [])
+ self.assertEqual(body["running_jobs"], 1)
+ self.assertEqual(body["projects"][0]["running_jobs"][0]["id"], "j1")
+
+ def test_a_stale_heartbeat_is_not_live(self):
+ session_live.beat(str(self.proj), "old-session")
+ path = session_live.heartbeat_path(str(self.proj), "old-session")
+ data = json.loads(path.read_text(encoding="utf-8"))
+ data["ts"] = time.time() - 3600
+ path.write_text(json.dumps(data), encoding="utf-8")
+ _, body = self.get("/api/mobile/now")
+ self.assertEqual(body["live_sessions"], 0)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_mobile_override_costs.py b/tests/test_mobile_override_costs.py
index 174d365..bb45152 100644
--- a/tests/test_mobile_override_costs.py
+++ b/tests/test_mobile_override_costs.py
@@ -112,7 +112,7 @@ def test_capability_is_advertised_and_follows_the_override_switch(self):
def test_api_version_is_not_bumped(self):
# Additive: an older client must not be told the API moved.
- self.assertEqual(mobile_api.API_VERSION, 6)
+ self.assertEqual(mobile_api.API_VERSION, 7)
def test_switch_off_is_404(self):
srv._cfg["mobile_override_enabled"] = False
diff --git a/tests/test_mobile_security_api.py b/tests/test_mobile_security_api.py
index 5a78e92..e2ffddb 100644
--- a/tests/test_mobile_security_api.py
+++ b/tests/test_mobile_security_api.py
@@ -34,7 +34,7 @@
os.environ.setdefault("C3_ORACLE_API_KEY", "mobile-sec-key")
import oracle.oracle_server as srv # noqa: E402
-from oracle.services import mobile_api # noqa: E402
+from oracle.services import mobile_api, mobile_fleet # noqa: E402
from services import access_guard as ag # noqa: E402
from services import credential_store as cs # noqa: E402
@@ -711,7 +711,9 @@ class TestMobileSurfaceInvariants(unittest.TestCase):
"""Structural assertions — stronger than any canary, because they hold for
values the tests never constructed."""
- IDENTS = _module_identifiers(mobile_api)
+ # mobile_fleet registers routes on the same blueprint (API v7), so the
+ # plaintext and builtin-disable bans cover it too.
+ IDENTS = _module_identifiers(mobile_api) | _module_identifiers(mobile_fleet)
def _mobile_rules(self):
return [str(r) for r in srv.app.url_map.iter_rules()
@@ -729,9 +731,19 @@ def test_no_plaintext_accessor_in_module(self):
f"{banned} gives this module access to plaintext")
def test_no_builtin_disable_route_or_reference(self):
+ # Invariant 2 since API v7: the gateway may read builtin modes, tighten
+ # one and end a lease early — never loosen, disable or mint a lease.
+ # The allowlist is exact so a new builtin route is a deliberate edit
+ # here, not something that slips in beside these three.
self.assertNotIn("set_builtin_disabled", self.IDENTS)
- for rule in self._mobile_rules():
- self.assertNotIn("builtin", rule)
+ # builtin_leases.mint is how a lease (a temporary loosening) is made.
+ self.assertNotIn("mint", _module_identifiers(mobile_fleet))
+ builtin_rules = {r for r in self._mobile_rules() if "builtin" in r}
+ self.assertEqual(builtin_rules, {
+ "/api/mobile/access/builtin",
+ "/api/mobile/access/builtin/mode",
+ "/api/mobile/access/builtin/lease/revoke",
+ })
def test_no_raw_content_preview_route(self):
for rule in self._mobile_rules():