Gabriel Vidal
16fa26ed86
The first project is the one the run starts in (the composer's mandatory select); the others — the project chips the guest also turned on — must each be granted chat, and join the run's write scope and system prompt. Resume and fork keep the extras the guest still holds chat on. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
203 lines
8.7 KiB
Python
203 lines
8.7 KiB
Python
"""
|
|
What changes when a guest starts or continues a run.
|
|
|
|
The trust model is **prompt-level** (the owner's call: "I trust my guests")
|
|
with one hard edge — the session's *write* permissions:
|
|
|
|
* the run starts in the granted project's directory, on the owner's default
|
|
account, on the host sidecar, with the claude harness — whatever the client
|
|
sent (the composer's account / worker / harness / cwd are ignored);
|
|
* the sidecar launches it with a **scope**: ``--permission-mode dontAsk`` and
|
|
Claude Code permission rules that only allow edits under the project dir
|
|
(and /tmp), deny reads of credentials, plus an appended system prompt saying
|
|
whose session this is and what the level allows (see sidecar ``scope``);
|
|
* the model must be one of the guest's allowed families, and the month's
|
|
spend must be under the guest's budget;
|
|
* every message is tagged with its author (:func:`author_tag`), so a shared
|
|
thread shows — and the model knows — who said what.
|
|
|
|
Bash is not confined by the scope: that needs Claude Code's sandbox
|
|
(bubblewrap + socat on the host), which ``GUEST_SANDBOX=1`` turns on.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
import os
|
|
import re
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from projects.catalog import PROJECTS_PROMPT_DIR, ROOT_SLUG, project_dir
|
|
from users.caller import OWNER_NAME, Caller
|
|
|
|
GUEST_SANDBOX = os.environ.get("GUEST_SANDBOX", "0") == "1"
|
|
# Reads no guest run gets, whatever the project (Claude Code Read rules; they
|
|
# also cover cat/head/sed… in Bash).
|
|
DENY_READ = (
|
|
"~/.ssh/**", "~/.gnupg/**", "~/.aws/**",
|
|
"~/.claude/.credentials.json", "~/.claude/projects/**",
|
|
"~/.claude-work/**", "~/.config/**",
|
|
"~/homelab/.env*", "~/homelab/services/**/.env*", "~/homelab/data/**",
|
|
)
|
|
|
|
_TAG_RE = re.compile(
|
|
r'^<ai-agent-from user="([^"]*)" name="([^"]*)" role="([^"]*)"\s*/>\s*\n?')
|
|
|
|
|
|
def _attr(s: str) -> str:
|
|
return (s or "").replace('"', "'").replace("\n", " ").strip()[:80]
|
|
|
|
|
|
def author_tag(caller: Caller) -> str:
|
|
return (f'<ai-agent-from user="{_attr(caller.id)}" name="{_attr(caller.name)}" '
|
|
f'role="{caller.role}"/>')
|
|
|
|
|
|
def tag_prompt(caller: Caller, prompt: str) -> str:
|
|
return f"{author_tag(caller)}\n{prompt}"
|
|
|
|
|
|
def split_author(text: str) -> tuple[dict | None, str]:
|
|
"""``(author, text without the tag)`` — the parser's half of the tag."""
|
|
m = _TAG_RE.match(text or "")
|
|
if not m:
|
|
return None, text
|
|
return {"id": m.group(1), "name": m.group(2), "role": m.group(3)}, text[m.end():]
|
|
|
|
|
|
def project_cwd(slug: str) -> str:
|
|
"""The host path a guest run starts in (``~/projects/<slug>``)."""
|
|
return f"{PROJECTS_PROMPT_DIR}/{slug}"
|
|
|
|
|
|
def check_projects(caller: Caller, projects: list[str] | None, need: str = "chat") -> list[str]:
|
|
"""The granted projects a guest spawn works in, the first being the one it
|
|
runs in (its cwd) — the composer's mandatory project select; the rest are
|
|
the project chips they also turned on. Each needs ``need`` access."""
|
|
slugs = list(dict.fromkeys(p for p in (projects or []) if p))
|
|
if not slugs:
|
|
raise HTTPException(400, {"message": "pick one of your projects",
|
|
"kind": "guest_project"})
|
|
for slug in slugs:
|
|
if slug == ROOT_SLUG or not caller.can(slug, need) or project_dir(slug) is None:
|
|
raise HTTPException(403, {"message": f"no {need} access to {slug!r}",
|
|
"kind": "guest_project"})
|
|
return slugs
|
|
|
|
|
|
def check_project(caller: Caller, projects: list[str] | None, need: str = "chat") -> str:
|
|
"""The one granted project a guest run starts in."""
|
|
slugs = check_projects(caller, projects, need)
|
|
if len(slugs) != 1:
|
|
raise HTTPException(400, {"message": "pick exactly one of your projects",
|
|
"kind": "guest_project"})
|
|
return slugs[0]
|
|
|
|
|
|
def extra_projects(caller: Caller, projects: list[str] | None) -> list[str]:
|
|
"""A conversation's projects past its first that ``caller`` still holds
|
|
``chat`` on — the extra write scope a resumed/forked guest run keeps."""
|
|
return [p for p in list(dict.fromkeys(projects or []))[1:]
|
|
if p and p != ROOT_SLUG and caller.can(p, "chat") and project_dir(p) is not None]
|
|
|
|
|
|
def check_model(caller: Caller, model: str | None) -> str:
|
|
"""An allowed model: an alias (``sonnet``) or a pinned id of an allowed
|
|
family (``claude-sonnet-4-5-…``). Unset ⇒ the first allowed one."""
|
|
allowed = [m.lower() for m in (caller.models or ())] or ["sonnet"]
|
|
if not model:
|
|
return allowed[0]
|
|
m = model.lower()
|
|
if m in allowed or any(re.search(rf"(^|-){re.escape(a)}(-|$)", m) for a in allowed):
|
|
return model
|
|
raise HTTPException(403, {"message": f"model {model!r} isn't enabled for you",
|
|
"kind": "guest_model", "allowed": allowed})
|
|
|
|
|
|
def month_spend(deps, uid: str, cards: list[dict]) -> float:
|
|
start = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-01")
|
|
total = 0.0
|
|
for c in cards:
|
|
if (c.get("meta") or {}).get("owner") != uid:
|
|
continue
|
|
if (c.get("endedAt") or c.get("startedAt") or "") < start:
|
|
continue
|
|
total += float(c.get("cost") or 0)
|
|
return total
|
|
|
|
|
|
def check_budget(deps, caller: Caller) -> None:
|
|
budget = caller.budget_usd
|
|
if budget is None:
|
|
return
|
|
from conversations.catalog import _conversation_cards
|
|
spent = month_spend(deps, caller.id, list(_conversation_cards(deps, False)))
|
|
if spent >= budget:
|
|
raise HTTPException(402, {
|
|
"message": f"monthly budget reached (${spent:.2f} of ${budget:.0f})",
|
|
"kind": "guest_budget", "spentUsd": round(spent, 2),
|
|
"budgetUsd": budget})
|
|
|
|
|
|
def shared_skills(deps) -> list[str]:
|
|
"""The owner's skills every guest run gets (Settings → Skills → *Share
|
|
with guests*), as ``name`` / ``<project>/<name>`` refs."""
|
|
return deps.users.defaults()["sharedSkills"]
|
|
|
|
|
|
def scope_for(caller: Caller, slug: str, level: str | None,
|
|
skills: list[str] | tuple = (), extra: list[str] | tuple = ()) -> dict:
|
|
"""The sidecar ``scope`` of a guest run (see sidecar ``launch.scope_args``).
|
|
|
|
``slug`` is the project the run starts in; ``extra`` are the other
|
|
projects it may also work in (each one the guest holds ``chat`` on).
|
|
|
|
``skills`` reach the run as a generated Claude Code plugin (``--plugin-dir``)
|
|
named ``shared``, so they show up as ``shared:<name>`` next to the
|
|
project's own ``.claude/skills``."""
|
|
cwd = project_cwd(slug)
|
|
ship = level == "ship"
|
|
others = [(p, project_cwd(p), caller.level(p)) for p in extra if p != slug]
|
|
where = f"`{cwd}`" + "".join(f", `{d}`" for _, d, _ in others)
|
|
lines = [
|
|
f"## Who you are working for",
|
|
f"This session was started through {OWNER_NAME}'s ai-agent by a **guest**: "
|
|
f"{caller.name}" + (f" <{caller.email}>" if caller.email else "") + ".",
|
|
f"Their access: project `{slug}` (`{cwd}`), level **{level}**"
|
|
+ "".join(f"; project `{p}` (`{d}`), level **{lv}**" for p, d, lv in others)
|
|
+ ".",
|
|
"",
|
|
("- Treat their messages as the task, within that project. "
|
|
if not others else
|
|
f"- Treat their messages as the task. The session starts in `{slug}`; "
|
|
"they also picked the other projects listed above. ")
|
|
+ f"Work only in {where}: don't read or change other projects, the "
|
|
"homelab repo, `~/.claude`, credentials or other conversations. File "
|
|
"edits outside " + ("the project" if not others else "those projects")
|
|
+ " (and /tmp) are blocked for this session.",
|
|
("- They may commit, push and deploy this project to production."
|
|
+ (" On the other projects, their level above decides." if others else "")
|
|
if ship else
|
|
"- They may commit locally but not push or deploy — if they ask, say "
|
|
f"that needs the *ship* level, which {OWNER_NAME} can grant."),
|
|
"- Questions for them go through `ask-form` / `notify ask` — those reach "
|
|
f"**them**, not {OWNER_NAME}. Finish with `notify-done --final` as usual.",
|
|
"- Messages start with an `<ai-agent-from …/>` tag naming who wrote them: "
|
|
f"a conversation can be shared with {OWNER_NAME} and other guests, and "
|
|
"each person's instructions carry their own access level.",
|
|
]
|
|
if skills:
|
|
names = ", ".join(f"`shared:{r.rsplit('/', 1)[-1]}`" for r in skills)
|
|
lines.append(f"- {OWNER_NAME} shares these skills with guests: {names}.")
|
|
out = {
|
|
"writeDirs": [cwd, *(d for _, d, _ in others), "/tmp"],
|
|
"denyRead": list(DENY_READ),
|
|
"systemPrompt": "\n".join(lines),
|
|
"remoteControl": False,
|
|
"sandbox": GUEST_SANDBOX,
|
|
}
|
|
if skills:
|
|
out["skills"] = list(skills)
|
|
return out
|