Apply per-user skills/connectors at launch; stamp installed skills' owner

claude_gen.apply now takes the launching project's owner and materializes
only shared rows plus that user's own; spawn and resume pass it through.
skill_install stamps imported rows with the requesting user from the
command payload (reinstalls keep the existing owner).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019ws7xj5Ej623hh4GXQCYYR

src/handler/control/claude_gen.py

@@ -137,8 +137,8 @@ def sync_user_skills(skills: list[dict], home: str | None = None) -> list[str]:
     return written
 
 
-def _load_skills(conn: Connection) -> list[dict]:
-    skills = repo.list_claude_skills(conn, enabled_only=True)
+def _load_skills(conn: Connection, visible_to) -> list[dict]:
+    skills = repo.list_claude_skills(conn, enabled_only=True, visible_to=visible_to)
     return [
         {
             **s,
@@ -150,15 +150,24 @@ def _load_skills(conn: Connection) -> list[dict]:
     ]
 
 
-def apply(working_dir: str, conn: Connection | None = None) -> dict:
-    """Apply the whole web-managed config for one launch; returns a small summary."""
+def apply(working_dir: str, conn: Connection | None = None, visible_to=None) -> dict:
+    """Apply the whole web-managed config for one launch; returns a small summary.
+
+    ``visible_to`` is the launching project's ``owner_user_id`` — the launch gets the
+    shared rows plus that user's own, so one user's skills and connectors never reach
+    another user's agents. ``None`` (a shared/legacy project) applies shared rows only,
+    which is exactly the pre-accounts behavior when nothing has an owner. Note the
+    skills sync target is the worker's user-level skills dir: each launch rewrites it
+    to its own visible set, so on a busy multi-user worker the set follows the most
+    recent launch (a bounded staleness, not a leak — a launch never *reads* another
+    user's skills into its own sync)."""
     if conn is None:
         with connection() as c:
-            connectors = repo.list_claude_connectors(c, enabled_only=True)
-            skills = _load_skills(c)
+            connectors = repo.list_claude_connectors(c, enabled_only=True, visible_to=visible_to)
+            skills = _load_skills(c, visible_to)
     else:
-        connectors = repo.list_claude_connectors(conn, enabled_only=True)
-        skills = _load_skills(conn)
+        connectors = repo.list_claude_connectors(conn, enabled_only=True, visible_to=visible_to)
+        skills = _load_skills(conn, visible_to)
     mcp_path = write_mcp_config(working_dir, [memory_server_connector()] + connectors)
     written = sync_user_skills(skills)
     return {"mcp_config": mcp_path, "skills_written": len(written)}

src/handler/control/skill_install.py

@@ -170,10 +170,13 @@ def _collect_skill(skill_dir: str) -> tuple[dict[str, str], dict[str, str], list
     return files_and_meta
 
 
-def import_staged(staging_dir: str, conn: Connection) -> list[dict]:
+def import_staged(
+    staging_dir: str, conn: Connection, owner_user_id: int | None = None
+) -> list[dict]:
     """Upsert every ``<staging>/<name>/SKILL.md`` as a managed skill row (matched by
     name — reinstalling a skill updates it in place) with its auxiliary files. Returns
-    one summary dict per skill."""
+    one summary dict per skill. New rows belong to ``owner_user_id`` (None = shared);
+    a reinstall keeps the existing row's owner."""
     results: list[dict] = []
     for entry in sorted(os.listdir(staging_dir)):
         skill_dir = os.path.join(staging_dir, entry)
@@ -190,7 +193,9 @@ def import_staged(staging_dir: str, conn: Connection) -> list[dict]:
         body = meta["__body__"].strip() + "\n"
         existing = repo.get_claude_skill_by_name(conn, name)
         if existing is None:
-            row = repo.create_claude_skill(conn, name, body, description=description)
+            row = repo.create_claude_skill(
+                conn, name, body, description=description, owner_user_id=owner_user_id
+            )
             action = "created"
         else:
             row = repo.update_claude_skill(
@@ -205,11 +210,12 @@ def import_staged(staging_dir: str, conn: Connection) -> list[dict]:
     return results
 
 
-def run(prompt: str) -> dict:
+def run(prompt: str, owner_user_id: int | None = None) -> dict:
     """The whole flow: stage, run the wrapped prompt through headless claude, import.
 
     Returns ``{"skills": [...], "summary": <claude's closing report>}``; raises
-    InstallError when the run fails or fetched nothing importable.
+    InstallError when the run fails or fetched nothing importable. Imported skills
+    belong to ``owner_user_id`` (None = shared).
     """
     prompt = (prompt or "").strip()
     if not prompt:
@@ -221,7 +227,7 @@ def run(prompt: str) -> dict:
         output = _run_claude(_WRAPPER.format(prompt=prompt), staging, settings_path)
         os.remove(settings_path)  # never importable, but keep the scan surface clean
         with connection() as conn:
-            skills = import_staged(staging, conn)
+            skills = import_staged(staging, conn, owner_user_id=owner_user_id)
     if not skills:
         raise InstallError(
             "the install run finished but no <skill>/SKILL.md landed in the staging "

src/handler/control/spawn.py

@@ -158,7 +158,8 @@ def spawn(
     # Materialize the web-managed Claude config (MCP connectors + user-level skills)
     # so this launch picks up what the operator configured in the dashboard. The skills
     # half also feeds pi-harness agents (their settings.json points at the same dir).
-    claude_gen.apply(working_dir)
+    # Scoped to the project's owner: shared rows plus theirs, nobody else's.
+    claude_gen.apply(working_dir, visible_to=project.get("owner_user_id"))
     env, harness = _agent_env(project, agent, token, role=role, mise_init=mise_init)
 
     # Verify the pinned forge version, if one is configured. Non-fatal: a version drift
@@ -284,7 +285,7 @@ def resume(agent: dict, answer: str, worker_id: str | None = None) -> tuple[bool
 
     working_dir = agent["working_dir"]
     settings_path = settings_gen.write_settings(working_dir)
-    claude_gen.apply(working_dir)
+    claude_gen.apply(working_dir, visible_to=project.get("owner_user_id"))
     try:
         token = None
         with connection() as conn:

src/handler/control/worker.py

@@ -251,12 +251,14 @@ def _cmd_login_submit(command: dict) -> dict:
 
 def _cmd_skill_install(command: dict) -> dict:
     """Run a pasted marketplace install prompt through a one-off headless claude and
-    import the fetched skills as managed rows (Claude page, Skills tab)."""
-    prompt = _payload(command).get("prompt")
+    import the fetched skills as managed rows (Claude page, Skills tab). Imported rows
+    belong to the requesting user (``owner_user_id`` in the payload; None = shared)."""
+    payload = _payload(command)
+    prompt = payload.get("prompt")
     if not prompt or not str(prompt).strip():
         raise CommandError("skill_install requires a 'prompt' in the payload")
     try:
-        return skill_install.run(str(prompt))
+        return skill_install.run(str(prompt), owner_user_id=payload.get("owner_user_id"))
     except skill_install.InstallError as exc:
         raise CommandError(str(exc)) from exc
 
@@ -488,7 +490,8 @@ def run(
             pass
         did_work = drain(worker_id) > 0
         now = time.monotonic()
-        if credsync_interval > 0 and (last_credsync == 0.0 or now - last_credsync >= credsync_interval):
+        credsync_due = last_credsync == 0.0 or now - last_credsync >= credsync_interval
+        if credsync_interval > 0 and credsync_due:
             # First pass runs immediately: a fresh worker container must materialize the
             # claude credentials before it claims its first spawn.
             try: