diff --git a/backend/shared/persimmons.py b/backend/shared/persimmons.py
new file mode 100644
index 0000000..8d3e0fd
--- /dev/null
+++ b/backend/shared/persimmons.py
@@ -0,0 +1,594 @@
+from __future__ import annotations
+
+import json
+import re
+import secrets
+from datetime import datetime, timezone
+from typing import Any, Awaitable, Callable
+
+LEGACY_REPO = "matcha/persimmons"
+META_PREFIX = "PERSIM_META_"
+SECRET_PREFIX = "PERSIM_SECRET_"
+JUICE_PREFIX = "PERSIM_JUICE_"
+ID_RE = re.compile(r"[A-F0-9]{12}")
+REPO_RE = re.compile(r"[A-Za-z0-9_.-]{1,100}/[A-Za-z0-9_.-]{1,100}")
+SECRET_NAME_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
+LEGACY_ID_PREFIX = "secret:"
+GiteaRequest = Callable[..., Awaitable[Any]]
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
+
+
+def connection_id(value: str) -> str:
+    value = str(value or "").strip().upper()
+    if not ID_RE.fullmatch(value):
+        raise ValueError("invalid connection id")
+    return value
+
+def secret_name(value: str) -> str:
+    value = re.sub(r"[^A-Za-z0-9_]+", "_", str(value or "").strip().upper()).strip("_")
+    if value and value[0].isdigit():
+        value = f"P_{value}"
+    value = value[:100]
+    if not value or not SECRET_NAME_RE.fullmatch(value):
+        raise ValueError("valid secret name is required")
+    return value
+
+
+def metadata_secret_name(metadata: dict[str, Any], cid: str) -> str:
+    stored = str(metadata.get("secret_name") or "").strip()
+    if stored:
+        return secret_name(stored)
+    return f"{SECRET_PREFIX}{connection_id(cid)}"
+
+
+
+def repository_name(value: str) -> str:
+    value = str(value or "").strip()
+    if not REPO_RE.fullmatch(value):
+        raise ValueError("invalid repository")
+    return value
+
+
+def _items(data: Any, key: str) -> list[dict[str, Any]]:
+    if isinstance(data, list):
+        return [item for item in data if isinstance(item, dict)]
+    if isinstance(data, dict) and isinstance(data.get(key), list):
+        return [item for item in data[key] if isinstance(item, dict)]
+    return []
+
+
+def _scopes(value: Any) -> list[str]:
+    raw = value.split(",") if isinstance(value, str) else value if isinstance(value, list) else []
+    result = [str(item).strip() for item in raw if str(item).strip()]
+    if len(result) > 40 or any(len(item) > 100 for item in result):
+        raise ValueError("invalid scopes")
+    return result
+
+
+def vault_from_input(data: dict[str, Any]) -> dict[str, Any]:
+    raw_kind = str(data.get("vault_kind") or "").strip().lower()
+    if not raw_kind:
+        return {"kind": "repository", "repository": LEGACY_REPO, "implicit_legacy": True}
+    if raw_kind == "user":
+        return {"kind": "user"}
+    if raw_kind == "repository":
+        return {"kind": "repository", "repository": repository_name(str(data.get("repository") or ""))}
+    raise ValueError("invalid vault")
+
+
+def clean_vault(vault: dict[str, Any]) -> dict[str, Any]:
+    if vault["kind"] == "user":
+        return {"kind": "user"}
+    return {"kind": "repository", "repository": vault["repository"]}
+
+
+def metadata_from_input(data: dict[str, Any], *, vault: dict[str, Any], created_at: str | None = None) -> dict[str, Any]:
+    provider = str(data.get("provider") or "").strip()
+    credential_type = str(data.get("credential_type") or "").strip()
+    account = str(data.get("account") or "").strip()
+    name = str(data.get("name") or "").strip() or provider
+    notes = str(data.get("notes") or "").strip()
+    expires = str(data.get("expires") or "").strip() or None
+    if not provider or not credential_type:
+        raise ValueError("provider and credential type are required")
+    for value in (name, provider, account, credential_type):
+        if len(value) > 200:
+            raise ValueError("field too long")
+    if len(notes) > 1000:
+        raise ValueError("notes too long")
+    now = utc_now()
+    return {
+        "kind": "credential",
+        "name": name,
+        "provider": provider,
+        "account": account,
+        "credential_type": credential_type,
+        "scopes": _scopes(data.get("scopes")),
+        "created_at": created_at or now,
+        "updated_at": now,
+        "expires": expires,
+        "notes": notes,
+        "status": "active",
+        "vault": clean_vault(vault),
+    }
+
+
+def _metadata(item: dict[str, Any]) -> dict[str, Any]:
+    raw = item.get("data") if item.get("data") is not None else item.get("value")
+    try:
+        value = json.loads(str(raw or "{}"))
+    except json.JSONDecodeError as exc:
+        raise ValueError("invalid Persimmons metadata") from exc
+    if not isinstance(value, dict):
+        raise ValueError("invalid Persimmons metadata")
+    return value
+
+
+def metadata_vault(metadata: dict[str, Any], fallback_repository: str | None = None) -> dict[str, Any]:
+    vault = metadata.get("vault")
+    if isinstance(vault, dict) and vault.get("kind") == "user":
+        return {"kind": "user"}
+    if isinstance(vault, dict) and vault.get("kind") == "repository":
+        return {"kind": "repository", "repository": repository_name(str(vault.get("repository") or ""))}
+    return {"kind": "repository", "repository": fallback_repository or LEGACY_REPO}
+
+
+def public_connection(cid: str, metadata: dict[str, Any], *, available: bool, fallback_repository: str | None = None) -> dict[str, Any]:
+    vault = metadata_vault(metadata, fallback_repository)
+    stored_name = metadata_secret_name(metadata, cid)
+    display_name = str(metadata.get("name") or metadata.get("provider") or stored_name)
+    return {
+        "id": cid,
+        "kind": str(metadata.get("kind") or "credential"),
+        "name": display_name,
+        "secret_name": stored_name,
+        "provider": str(metadata.get("provider") or ""),
+        "account": str(metadata.get("account") or ""),
+        "credential_type": str(metadata.get("credential_type") or ""),
+        "scopes": metadata.get("scopes") if isinstance(metadata.get("scopes"), list) else [],
+        "created_at": metadata.get("created_at"),
+        "updated_at": metadata.get("updated_at"),
+        "expires": metadata.get("expires") or None,
+        "notes": str(metadata.get("notes") or ""),
+        "status": "active" if available else "revoked",
+        "available": bool(available),
+        "vault_kind": vault["kind"],
+        "repository": vault.get("repository"),
+    }
+
+
+def legacy_secret_name(value: str) -> str | None:
+    raw = str(value or "")
+    if not raw.startswith(LEGACY_ID_PREFIX):
+        return None
+    name = raw[len(LEGACY_ID_PREFIX):]
+    if not SECRET_NAME_RE.fullmatch(name):
+        raise ValueError("invalid legacy secret")
+    return name
+
+
+def public_legacy_secret(
+    name: str,
+    vault: dict[str, Any],
+    *,
+    available: bool = True,
+    created_at: str | None = None,
+    description: str = "",
+) -> dict[str, Any]:
+    if not SECRET_NAME_RE.fullmatch(name):
+        raise ValueError("invalid legacy secret")
+    return {
+        "id": f"{LEGACY_ID_PREFIX}{name}",
+        "kind": "credential",
+        "name": name,
+        "provider": "",
+        "account": "",
+        "credential_type": "Existing Gitea secret",
+        "scopes": [],
+        "created_at": created_at,
+        "updated_at": created_at,
+        "expires": None,
+        "notes": description,
+        "status": "active" if available else "revoked",
+        "available": bool(available),
+        "vault_kind": vault["kind"],
+        "repository": vault.get("repository"),
+        "legacy": True,
+        "secret_name": name,
+    }
+
+
+def _prefix(vault: dict[str, Any]) -> str:
+    return "user/actions" if vault["kind"] == "user" else f"repos/{vault['repository']}/actions"
+
+
+async def _validate_repository(request: GiteaRequest, repository: str) -> dict[str, Any]:
+    repo = await request("GET", f"repos/{repository}")
+    if not isinstance(repo, dict):
+        raise ValueError("repository unavailable")
+    permissions = repo.get("permissions") if isinstance(repo.get("permissions"), dict) else {}
+    if permissions and not (permissions.get("admin") or permissions.get("push")):
+        raise ValueError("repository is not writable")
+    return repo
+
+
+async def list_repositories(request: GiteaRequest) -> dict[str, Any]:
+    repos: list[dict[str, Any]] = []
+    for page in range(1, 6):
+        data = await request("GET", "user/repos", params={"limit": 100, "page": page})
+        batch = _items(data, "data")
+        if not batch:
+            break
+        for repo in batch:
+            full_name = str(repo.get("full_name") or "")
+            permissions = repo.get("permissions") if isinstance(repo.get("permissions"), dict) else {}
+            if full_name and (not permissions or permissions.get("admin") or permissions.get("push")):
+                repos.append({
+                    "full_name": full_name,
+                    "private": bool(repo.get("private")),
+                    "owner": str((repo.get("owner") or {}).get("login") or ""),
+                })
+        if len(batch) < 100:
+            break
+    repos.sort(key=lambda item: item["full_name"].lower())
+    return {"repositories": repos}
+
+
+async def _variable_get(request: GiteaRequest, vault: dict[str, Any], cid: str, *, prefix: str = META_PREFIX) -> dict[str, Any] | None:
+    try:
+        return await request("GET", f"{_prefix(vault)}/variables/{prefix}{cid}")
+    except RuntimeError as exc:
+        if "HTTP 404" in str(exc):
+            return None
+        raise
+
+
+async def _variable_put(request: GiteaRequest, vault: dict[str, Any], cid: str, metadata: dict[str, Any], *, exists: bool, prefix: str = META_PREFIX) -> None:
+    name = f"{prefix}{cid}"
+    path = f"{_prefix(vault)}/variables/{name}"
+    body = {
+        "value": json.dumps(metadata, separators=(",", ":"), ensure_ascii=False),
+        "description": "Persimmons object metadata. Contains no protected value.",
+    }
+    await request("PUT" if exists else "POST", path, json=body)
+
+
+async def _list_vault(request: GiteaRequest, vault: dict[str, Any]) -> list[dict[str, Any]]:
+    if vault["kind"] == "repository" and vault["repository"] != LEGACY_REPO:
+        await _validate_repository(request, vault["repository"])
+    variable_data = await request("GET", f"{_prefix(vault)}/variables", params={"limit": 100})
+
+    secret_items: list[dict[str, Any]] = []
+    secret_listing_supported = True
+    try:
+        secret_data = await request("GET", f"{_prefix(vault)}/secrets", params={"limit": 100})
+        secret_items = _items(secret_data, "secrets")
+    except RuntimeError as exc:
+        if vault["kind"] == "user" and "HTTP 404" in str(exc):
+            secret_listing_supported = False
+        else:
+            raise
+    secret_names = {
+        str(item.get("name") or "")
+        for item in secret_items
+        if str(item.get("name") or "")
+    }
+
+    result: list[dict[str, Any]] = []
+    metadata_secret_names: set[str] = set()
+    for item in _items(variable_data, "variables"):
+        name = str(item.get("name") or "")
+        if not name.startswith(META_PREFIX):
+            continue
+        cid = name[len(META_PREFIX):]
+        if not ID_RE.fullmatch(cid):
+            continue
+        try:
+            metadata = _metadata(item)
+        except ValueError:
+            continue
+        stored_name = metadata_secret_name(metadata, cid)
+        metadata_secret_names.add(stored_name)
+        available = (
+            stored_name in secret_names
+            if secret_listing_supported
+            else str(metadata.get("status") or "active") == "active"
+        )
+        result.append(
+            public_connection(
+                cid,
+                metadata,
+                available=available,
+                fallback_repository=vault.get("repository"),
+            )
+        )
+
+    for item in secret_items:
+        name = str(item.get("name") or "")
+        if not name or name in metadata_secret_names or not SECRET_NAME_RE.fullmatch(name):
+            continue
+        result.append(
+            public_legacy_secret(
+                name,
+                clean_vault(vault),
+                available=True,
+                created_at=str(item.get("created_at") or "") or None,
+                description=str(item.get("description") or ""),
+            )
+        )
+    return result
+
+
+async def list_connections(
+    request: GiteaRequest,
+    *,
+    repository: str | None = None,
+    include_legacy: bool = True,
+    include_user: bool = True,
+) -> dict[str, Any]:
+    result: list[dict[str, Any]] = []
+    if include_user:
+        result.extend(await _list_vault(request, {"kind": "user"}))
+    target = repository_name(repository) if repository else None
+    if target:
+        result.extend(await _list_vault(request, {"kind": "repository", "repository": target}))
+    elif include_legacy:
+        try:
+            result.extend(await _list_vault(request, {"kind": "repository", "repository": LEGACY_REPO}))
+        except (RuntimeError, ValueError):
+            pass
+    result.sort(key=lambda item: str(item.get("updated_at") or item.get("created_at") or ""), reverse=True)
+    return {"connections": result}
+
+
+async def create_connection(request: GiteaRequest, data: dict[str, Any]) -> dict[str, Any]:
+    credential = str(data.get("credential") or "")
+    if not credential or len(credential) > 16384:
+        raise ValueError("valid credential value is required")
+    vault = vault_from_input(data)
+    if vault["kind"] == "repository" and not vault.get("implicit_legacy"):
+        await _validate_repository(request, vault["repository"])
+    cid = secrets.token_hex(6).upper()
+    metadata = metadata_from_input(data, vault=vault)
+    stored_name = secret_name(metadata["name"])
+    metadata["name"] = stored_name
+    metadata["secret_name"] = stored_name
+
+    existing_data = await request("GET", f"{_prefix(vault)}/secrets", params={"limit": 100})
+    existing_names = {
+        str(item.get("name") or "")
+        for item in _items(existing_data, "secrets")
+        if str(item.get("name") or "")
+    }
+    if stored_name in existing_names:
+        raise ValueError(f"secret {stored_name} already exists")
+
+    secret_path = f"{_prefix(vault)}/secrets/{stored_name}"
+    await request("PUT", secret_path, json={"data": credential, "description": "Persimmons write-only credential"})
+    try:
+        await _variable_put(request, vault, cid, metadata, exists=False)
+    except Exception:
+        try:
+            await request("DELETE", secret_path)
+        except Exception:
+            pass
+        raise
+    return public_connection(cid, metadata, available=True)
+
+
+async def _existing(request: GiteaRequest, cid: str, vault: dict[str, Any]) -> dict[str, Any]:
+    if vault["kind"] == "repository" and not vault.get("implicit_legacy"):
+        await _validate_repository(request, vault["repository"])
+    variable = await _variable_get(request, vault, cid)
+    if variable is None:
+        raise ValueError("connection not found")
+    metadata = _metadata(variable)
+    if metadata_vault(metadata, vault.get("repository")) != clean_vault(vault):
+        raise ValueError("connection vault does not match")
+    return metadata
+
+
+async def rotate_connection(request: GiteaRequest, cid: str, credential: str, vault_data: dict[str, Any] | None = None) -> dict[str, Any]:
+    credential = str(credential or "")
+    if not credential or len(credential) > 16384:
+        raise ValueError("valid credential value is required")
+    vault = vault_from_input(vault_data or {})
+    legacy_name = legacy_secret_name(cid)
+    if legacy_name:
+        if vault["kind"] == "repository" and not vault.get("implicit_legacy"):
+            await _validate_repository(request, vault["repository"])
+        await request(
+            "PUT",
+            f"{_prefix(vault)}/secrets/{legacy_name}",
+            json={"data": credential, "description": "Existing Gitea secret managed by Persimmons"},
+        )
+        return public_legacy_secret(legacy_name, clean_vault(vault), available=True)
+    cid = connection_id(cid)
+    metadata = await _existing(request, cid, vault)
+    stored_name = metadata_secret_name(metadata, cid)
+    await request("PUT", f"{_prefix(vault)}/secrets/{stored_name}", json={"data": credential, "description": "Persimmons write-only credential"})
+    metadata["updated_at"] = utc_now()
+    metadata["status"] = "active"
+    await _variable_put(request, vault, cid, metadata, exists=True)
+    return public_connection(cid, metadata, available=True)
+
+
+async def revoke_connection(request: GiteaRequest, cid: str, vault_data: dict[str, Any] | None = None) -> dict[str, Any]:
+    vault = vault_from_input(vault_data or {})
+    legacy_name = legacy_secret_name(cid)
+    if legacy_name:
+        if vault["kind"] == "repository" and not vault.get("implicit_legacy"):
+            await _validate_repository(request, vault["repository"])
+        try:
+            await request("DELETE", f"{_prefix(vault)}/secrets/{legacy_name}")
+        except RuntimeError as exc:
+            if "HTTP 404" not in str(exc):
+                raise
+        return public_legacy_secret(legacy_name, clean_vault(vault), available=False)
+    cid = connection_id(cid)
+    metadata = await _existing(request, cid, vault)
+    stored_name = metadata_secret_name(metadata, cid)
+    try:
+        await request("DELETE", f"{_prefix(vault)}/secrets/{stored_name}")
+    except RuntimeError as exc:
+        if "HTTP 404" not in str(exc):
+            raise
+    metadata["updated_at"] = utc_now()
+    metadata["status"] = "revoked"
+    await _variable_put(request, vault, cid, metadata, exists=True)
+    return public_connection(cid, metadata, available=False)
+
+
+async def delete_connection(request: GiteaRequest, cid: str, vault_data: dict[str, Any] | None = None) -> dict[str, Any]:
+    vault = vault_from_input(vault_data or {})
+    legacy_name = legacy_secret_name(cid)
+    if legacy_name:
+        if vault["kind"] == "repository" and not vault.get("implicit_legacy"):
+            await _validate_repository(request, vault["repository"])
+        try:
+            await request("DELETE", f"{_prefix(vault)}/secrets/{legacy_name}")
+        except RuntimeError as exc:
+            if "HTTP 404" not in str(exc):
+                raise
+        return {
+            "id": cid,
+            "deleted": True,
+            "vault_kind": vault["kind"],
+            "repository": vault.get("repository"),
+            "legacy": True,
+        }
+    cid = connection_id(cid)
+    metadata = await _existing(request, cid, vault)
+    stored_name = metadata_secret_name(metadata, cid)
+    for path in (
+        f"{_prefix(vault)}/secrets/{stored_name}",
+        f"{_prefix(vault)}/variables/{META_PREFIX}{cid}",
+    ):
+        try:
+            await request("DELETE", path)
+        except RuntimeError as exc:
+            if "HTTP 404" not in str(exc):
+                raise
+    return {"id": cid, "deleted": True, "vault_kind": vault["kind"], "repository": vault.get("repository")}
+
+
+def juice_vault(data: dict[str, Any]) -> dict[str, Any]:
+    """Juice defaults to My vault, never the implicit legacy repository."""
+    repository = data.get("repository")
+    kind = data.get("vault_kind") or ("repository" if repository else "user")
+    if kind == "user" and repository:
+        raise ValueError("Juice vault does not match repository")
+    return vault_from_input({"vault_kind": kind, "repository": repository})
+
+
+def _juice_policy(data: dict[str, Any]) -> dict[str, Any]:
+    subject = data.get("subject")
+    scopes = data.get("scopes")
+    if not isinstance(subject, str) or not subject.strip() or len(subject) > 200:
+        raise ValueError("valid subject is required (maximum 200 characters)")
+    if not isinstance(scopes, list) or not scopes or any(
+        not isinstance(scope, str) or not scope.strip() for scope in scopes
+    ):
+        raise ValueError("nonempty scopes array is required")
+    expires = data.get("expires_at")
+    if expires is not None:
+        try:
+            if not isinstance(expires, str) or len(expires) > 40:
+                raise ValueError
+            parsed = datetime.fromisoformat(expires.replace("Z", "+00:00"))
+            if parsed.tzinfo is None:
+                raise ValueError
+            expires = parsed.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
+        except ValueError as exc:
+            raise ValueError("expires_at must be a timestamp with timezone or null") from exc
+    return {"subject": subject.strip(), "scopes": _scopes(scopes), "expires_at": expires}
+
+
+def public_juice(jid: str, metadata: dict[str, Any], vault: dict[str, Any]) -> dict[str, Any]:
+    if metadata.get("vault") != clean_vault(vault):
+        raise ValueError("Juice vault does not match")
+    if metadata.get("kind") != "juice" or metadata.get("status") not in {"active", "revoked"}:
+        raise ValueError("invalid Juice metadata")
+    source = metadata.get("source")
+    if not isinstance(source, str) or not source or len(source) > 200:
+        raise ValueError("invalid Juice source")
+    # Explicit projection: never return arbitrary stored fields or source metadata.
+    return {
+        "id": jid, "kind": "juice",
+        "source_id": connection_id(metadata.get("source_id")), "source": source,
+        **_juice_policy(metadata), "created_at": metadata.get("created_at"),
+        "status": metadata["status"], "revocable": True,
+        "vault": clean_vault(vault), "vault_kind": vault["kind"],
+        "repository": vault.get("repository"),
+    }
+
+
+async def create_juice(request: GiteaRequest, data: dict[str, Any]) -> dict[str, Any]:
+    vault = juice_vault(data)
+    cid = connection_id(data.get("source_id"))
+    policy = _juice_policy(data)
+    if policy["expires_at"] is not None and datetime.fromisoformat(
+        policy["expires_at"].replace("Z", "+00:00")
+    ) <= datetime.now(timezone.utc):
+        raise ValueError("expires_at must be in the future")
+    source = await _existing(request, cid, vault)
+    if source.get("status", "active") != "active":
+        raise ValueError("source connection is revoked")
+    # Gitea's collection returns names only; no individual secret GET is allowed.
+    try:
+        names = await request("GET", f"{_prefix(vault)}/secrets", params={"limit": 100})
+        if metadata_secret_name(source, cid) not in {item.get("name") for item in _items(names, "secrets")}:
+            raise ValueError("source connection is unavailable")
+    except RuntimeError as exc:
+        # Older Gitea user APIs lack secret listing; use the saved active metadata.
+        if vault["kind"] != "user" or "HTTP 404" not in str(exc):
+            raise
+    jid = secrets.token_hex(6).upper()
+    metadata = public_juice(jid, {
+        "kind": "juice", "source_id": cid, "source": source.get("provider"),
+        **policy, "created_at": utc_now(), "status": "active", "vault": clean_vault(vault),
+    }, vault)
+    await _variable_put(request, vault, jid, metadata, exists=False, prefix=JUICE_PREFIX)
+    return metadata
+
+
+async def list_juice(
+    request: GiteaRequest, *, repository: str | None = None, source_id: str | None = None,
+) -> dict[str, Any]:
+    vault = juice_vault({"repository": repository})
+    cid = connection_id(source_id) if source_id is not None else None
+    if vault["kind"] == "repository":
+        await _validate_repository(request, vault["repository"])
+    variables = await request("GET", f"{_prefix(vault)}/variables", params={"limit": 100})
+    result = []
+    for item in _items(variables, "variables"):
+        name = str(item.get("name") or "")
+        if not name.startswith(JUICE_PREFIX) or not ID_RE.fullmatch(name[len(JUICE_PREFIX):]):
+            continue
+        try:
+            juice = public_juice(name[len(JUICE_PREFIX):], _metadata(item), vault)
+        except ValueError:
+            continue
+        if cid is None or juice["source_id"] == cid:
+            result.append(juice)
+    result.sort(key=lambda item: str(item.get("created_at") or ""), reverse=True)
+    return {"juice": result}
+
+
+async def revoke_juice(
+    request: GiteaRequest, jid: str, vault_data: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    jid = connection_id(jid)
+    vault = juice_vault(vault_data or {})
+    if vault["kind"] == "repository":
+        await _validate_repository(request, vault["repository"])
+    variable = await _variable_get(request, vault, jid, prefix=JUICE_PREFIX)
+    if variable is None:
+        raise ValueError("Juice not found")
+    metadata = public_juice(jid, _metadata(variable), vault)
+    metadata["status"] = "revoked"
+    await _variable_put(request, vault, jid, metadata, exists=True, prefix=JUICE_PREFIX)
+    return metadata