@clawhub-gyuryongkim-bacb797b56
Gusnais (Ruby-China/Homeland compatible) API integration with web-parity behavior and permission-consistent UX. Use when users want to connect using only CLI...
---
name: gusnais-skill
description: Gusnais (Ruby-China/Homeland compatible) API integration with web-parity behavior and permission-consistent UX. Use when users want to connect using only CLIENT_ID and CLIENT_SECRET, auto-complete OAuth/API settings, keep capability differences identical to gusnais.com based on abilities and server authorization, or perform read/write API operations for plugin domains (press, note, jobs, site).
---
# Gusnais Skill
Implement Gusnais API integration that mirrors web behavior and permission boundaries.
## Require only two user inputs
- `CLIENT_ID`
- `CLIENT_SECRET`
Do not ask for base URL, OAuth paths, account IDs, scope defaults, pagination defaults, or serializer mappings unless discovery fails.
## Auto-complete platform config
Use these defaults:
- Site: `https://gusnais.com`
- OAuth Authorize: `/oauth/authorize`
- OAuth Token: `/oauth/token`
- OAuth Revoke: `/oauth/revoke`
- API Base: `/api/v3`
## Auth flow
1. Build authorization URL automatically.
2. Exchange authorization code for `access_token` and `refresh_token`.
3. Validate token with `GET /api/v3/users/me`.
4. Refresh once on 401; if refresh fails, request re-auth.
Prefer Authorization header for requests:
- `Authorization: Bearer <access_token>`
Keep `access_token` query fallback for compatibility with Homeland API behavior.
## Web parity contract
Match gusnais.com UX and permission behavior:
1. **Read abilities first when available**
- Resource-level actions must follow returned `abilities`.
2. **Dual check**
- UI check (visible/enabled) using abilities.
- Execution check with real API call and status code handling.
3. **No privilege escalation**
- Never assume admin/mod privileges in client logic.
4. **Respect hidden/inaccessible resources**
- 404/403 semantics should stay consistent with server behavior.
## Capability gating model
For each action produce:
- `visible`: `true|false`
- `enabled`: `true|false`
- `reason`: `ok|no_permission|auth_required|resource_unavailable|validation_error`
- `source`: `abilities|server_status|policy`
## Endpoint behavior alignment
Use endpoint mapping in `references/endpoints.md` and serializer notes for normalized outputs.
Keep defaults aligned with docs:
- offset default: `0`
- limit default: `20`
- limit range on list endpoints: `1..150` (or endpoint-specific documented max)
- topic list default `type=last_actived`
For plugin domain operations (press/note/site/jobs):
- Read plugin web-route parity and API contract in `references/endpoints.md`.
- Read permission nuances in `references/permission-parity.md`.
- Treat 404 on plugin API endpoints as `resource_unavailable` unless deployment has enabled those API routes.
## Topic action safety
For `POST /api/v3/topics/:id/action?type=:type` (`ban|excellent|unexcellent|close|open`):
- Gate by `abilities` if present.
- Enforce final server response.
- Never expose action as enabled when denied.
## Error mapping
Normalize API errors without changing meaning:
- 400 -> `validation_error`
- 401 -> `auth_required` (refresh then retry once)
- 403 -> `no_permission`
- 404 -> `resource_unavailable`
- 500 -> `server_error`
Return original server error text when available.
## Rate limiting / retries
- Respect `Retry-After` on 429.
- Use exponential backoff with jitter for transient 5xx.
- Avoid one-item tight loops for batch writes.
## Read these references before implementation
- `references/endpoints.md`
- `references/permission-parity.md`
## Bootstrap script
Use `scripts/gusnais_bootstrap.py` to initialize runtime config from `CLIENT_ID` and `CLIENT_SECRET`.
Recommended:
- set `TOKEN_STORE_PATH` when exchanging code, so refreshable tokens are persisted to JSON for long-lived automation.
## Plugin API client script
Use `scripts/gusnais_plugin_client.py` for plugin API read/write calls with:
- auto refresh before expiry and on 401;
- one retry after refresh;
- normalized status reason mapping;
- capability hint extraction from `abilities`;
- action-level payload guardrails to avoid avoidable 400/500 (e.g. press create summary fallback).
Current deployment notes (2026-03-19):
- Press API is mounted for read/write (`/api/v3/press/posts*`).
- Note API is mounted for read/write (`/api/v3/note/notes*`).
- Site API is mounted for `sites` CRUD + `site_nodes` list; `undestroy`/site_node writes are not mounted.
- Treat any unmounted plugin API route as `resource_unavailable` and avoid repeated retries.
FILE:scripts/gusnais_bootstrap.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Bootstrap Gusnais API config using only CLIENT_ID and CLIENT_SECRET.
Usage:
1) Export CLIENT_ID and CLIENT_SECRET.
2) Run once to get authorize_url.
3) Complete OAuth in browser and export OAUTH_CODE.
4) Run again to exchange token and verify identity via /api/v3/users/me.
5) (Optional) set TOKEN_STORE_PATH to persist refreshable token store JSON.
"""
import json
import os
import time
import urllib.parse
from pathlib import Path
from typing import Any, Dict, Optional
import requests
SITE = "https://gusnais.com"
AUTHORIZE_PATH = "/oauth/authorize"
TOKEN_PATH = "/oauth/token"
REVOKE_PATH = "/oauth/revoke"
API_BASE = "/api/v3"
class BootError(Exception):
pass
def env_required(name: str) -> str:
value = os.getenv(name, "").strip()
if not value:
raise BootError(f"missing env: {name}")
return value
def build_authorize_url(client_id: str, redirect_uri: str, state: str, scope: str = "") -> str:
params = {
"client_id": client_id,
"redirect_uri": redirect_uri,
"response_type": "code",
"state": state,
}
if scope:
params["scope"] = scope
return f"{SITE}{AUTHORIZE_PATH}?{urllib.parse.urlencode(params)}"
def post_token(data: Dict[str, str]) -> Dict[str, Any]:
res = requests.post(f"{SITE}{TOKEN_PATH}", data=data, timeout=20)
if res.status_code >= 400:
raise BootError(f"token request failed: {res.status_code} {res.text[:300]}")
payload = res.json()
if "access_token" not in payload:
raise BootError("token response missing access_token")
return payload
def exchange_code(client_id: str, client_secret: str, code: str, redirect_uri: str) -> Dict[str, Any]:
return post_token(
{
"grant_type": "authorization_code",
"client_id": client_id,
"client_secret": client_secret,
"code": code,
"redirect_uri": redirect_uri,
}
)
def refresh_access_token(client_id: str, client_secret: str, refresh_token: str) -> Dict[str, Any]:
return post_token(
{
"grant_type": "refresh_token",
"client_id": client_id,
"client_secret": client_secret,
"refresh_token": refresh_token,
}
)
def api_get(path: str, access_token: Optional[str] = None, params: Optional[Dict[str, Any]] = None) -> Any:
url = f"{SITE}{API_BASE}{path}"
query = dict(params or {})
headers = {}
if access_token:
headers["Authorization"] = f"Bearer {access_token}"
# compatibility with Homeland API docs
query["access_token"] = access_token
res = requests.get(url, params=query, headers=headers, timeout=20)
if res.status_code >= 400:
raise BootError(f"GET {path} failed: {res.status_code} {res.text[:300]}")
return res.json()
def extract_identity(me_payload: Any) -> Dict[str, Any]:
"""Normalize /users/me payload variants to a flat identity dict."""
def pick_fields(obj: Any) -> Optional[Dict[str, Any]]:
if not isinstance(obj, dict):
return None
keys = {"id", "login", "name", "email"}
if keys.intersection(obj.keys()):
return {
"id": obj.get("id"),
"login": obj.get("login"),
"name": obj.get("name"),
"email": obj.get("email"),
}
return None
# Most common cases
for key in ("user", "data", "profile", "attributes"):
picked = pick_fields(me_payload.get(key) if isinstance(me_payload, dict) else None)
if picked:
return picked
# Flat payload fallback
picked = pick_fields(me_payload)
if picked:
return picked
# Heuristic: first nested dict containing at least one identity field
if isinstance(me_payload, dict):
for value in me_payload.values():
picked = pick_fields(value)
if picked:
return picked
return {"id": None, "login": None, "name": None, "email": None}
def write_token_store(
token_store_path: str,
client_id: str,
client_secret: str,
redirect_uri: str,
token_data: Dict[str, Any],
) -> None:
"""Persist OAuth tokens for long-lived refresh flow."""
now = int(time.time())
expires_in = int(token_data.get("expires_in") or 0)
payload = {
"site": SITE,
"api_base": f"{SITE}{API_BASE}",
"oauth": {
"authorize_endpoint": f"{SITE}{AUTHORIZE_PATH}",
"token_endpoint": f"{SITE}{TOKEN_PATH}",
"revoke_endpoint": f"{SITE}{REVOKE_PATH}",
},
"client": {
"client_id": client_id,
"client_secret": client_secret,
"redirect_uri": redirect_uri,
},
"token": {
"access_token": token_data.get("access_token"),
"refresh_token": token_data.get("refresh_token"),
"token_type": token_data.get("token_type"),
"scope": token_data.get("scope"),
"expires_in": expires_in,
"expires_at": (now + expires_in) if expires_in > 0 else None,
"created_at": token_data.get("created_at", now),
},
}
path = Path(token_store_path).expanduser()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
os.chmod(path, 0o600)
def main() -> None:
client_id = env_required("CLIENT_ID")
client_secret = env_required("CLIENT_SECRET")
redirect_uri = os.getenv("REDIRECT_URI", "urn:ietf:wg:oauth:2.0:oob")
state = os.getenv("OAUTH_STATE", "gusnais-state")
scope = os.getenv("OAUTH_SCOPE", "")
code = os.getenv("OAUTH_CODE", "").strip()
token_store_path = os.getenv("TOKEN_STORE_PATH", "").strip()
output: Dict[str, Any] = {
"site": SITE,
"oauth": {
"authorize_endpoint": f"{SITE}{AUTHORIZE_PATH}",
"token_endpoint": f"{SITE}{TOKEN_PATH}",
"revoke_endpoint": f"{SITE}{REVOKE_PATH}",
},
"api_base": f"{SITE}{API_BASE}",
"authorize_url": build_authorize_url(client_id, redirect_uri, state, scope),
"token_ready": False,
}
if not code:
print(json.dumps(output, ensure_ascii=False, indent=2))
return
token_data = exchange_code(client_id, client_secret, code, redirect_uri)
access_token = token_data["access_token"]
me = api_get("/users/me", access_token=access_token)
output["token_ready"] = True
output["identity"] = extract_identity(me)
output["token_meta"] = {
"expires_in": token_data.get("expires_in"),
"has_refresh_token": bool(token_data.get("refresh_token")),
"token_type": token_data.get("token_type"),
}
if token_store_path:
write_token_store(token_store_path, client_id, client_secret, redirect_uri, token_data)
output["token_store"] = token_store_path
print(json.dumps(output, ensure_ascii=False, indent=2))
if __name__ == "__main__":
try:
main()
except BootError as err:
print(json.dumps({"ok": False, "error": str(err)}, ensure_ascii=False))
raise SystemExit(1)
FILE:scripts/gusnais_plugin_client.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Plugin API client for Gusnais/Homeland deployments.
Features:
- Load/save token store JSON (generated by gusnais_bootstrap.py)
- Auto-refresh access token before expiry or on 401
- One-shot action runner for plugin read/write endpoints
- Capability gating helper from `abilities`
- Action-specific payload guardrails (avoid obvious 400/500)
Usage examples:
python3 scripts/gusnais_plugin_client.py --token-store /tmp/gusnais-token.json --action jobs.topics.list
python3 scripts/gusnais_plugin_client.py --token-store /tmp/gusnais-token.json --action jobs.topics.create --json '{"title":"招聘:后端工程师","body":"职位描述..."}'
python3 scripts/gusnais_plugin_client.py --token-store /tmp/gusnais-token.json --action press.posts.list --query '{"limit":20}'
"""
from __future__ import annotations
import argparse
import json
import os
import time
from pathlib import Path
from typing import Any, Dict, Optional, Tuple
import requests
class ClientError(Exception):
pass
ACTIONS: Dict[str, Tuple[str, str, Dict[str, Any]]] = {
# Jobs plugin: use core topics API with fixed node_id=25
"jobs.topics.list": ("GET", "/api/v3/topics", {"inject_query": {"node_id": 25, "type": "last_actived"}}),
"jobs.topics.create": ("POST", "/api/v3/topics", {"inject_json": {"node_id": 25}}),
"jobs.topics.get": ("GET", "/api/v3/topics/{id}", {}),
"jobs.topics.update": ("PUT", "/api/v3/topics/{id}", {}),
"jobs.topics.delete": ("DELETE", "/api/v3/topics/{id}", {}),
# Press plugin API contract
"press.posts.list": ("GET", "/api/v3/press/posts", {}),
"press.posts.upcoming": ("GET", "/api/v3/press/posts/upcoming", {}),
"press.posts.get": ("GET", "/api/v3/press/posts/{id}", {}),
"press.posts.create": ("POST", "/api/v3/press/posts", {}),
"press.posts.update": ("PUT", "/api/v3/press/posts/{id}", {}),
"press.posts.delete": ("DELETE", "/api/v3/press/posts/{id}", {}),
"press.posts.publish": ("PUT", "/api/v3/press/posts/{id}/publish", {}),
"press.posts.preview": ("POST", "/api/v3/press/posts/preview", {}),
# Note plugin API contract
"note.notes.list": ("GET", "/api/v3/note/notes", {}),
"note.notes.get": ("GET", "/api/v3/note/notes/{id}", {}),
"note.notes.create": ("POST", "/api/v3/note/notes", {}),
"note.notes.update": ("PUT", "/api/v3/note/notes/{id}", {}),
"note.notes.delete": ("DELETE", "/api/v3/note/notes/{id}", {}),
"note.notes.preview": ("POST", "/api/v3/note/notes/preview", {}),
# Site plugin API contract (deployment-verified subset)
"site.sites.list": ("GET", "/api/v3/site/sites", {}),
"site.sites.get": ("GET", "/api/v3/site/sites/{id}", {}),
"site.sites.create": ("POST", "/api/v3/site/sites", {}),
"site.sites.update": ("PUT", "/api/v3/site/sites/{id}", {}),
"site.sites.delete": ("DELETE", "/api/v3/site/sites/{id}", {}),
"site.nodes.list": ("GET", "/api/v3/site/site_nodes", {}),
}
def _json_loads(text: str) -> Dict[str, Any]:
text = (text or "").strip()
if not text:
return {}
obj = json.loads(text)
if not isinstance(obj, dict):
raise ClientError("JSON input must be an object")
return obj
def _merge(base: Dict[str, Any], extra: Dict[str, Any]) -> Dict[str, Any]:
out = dict(base)
out.update(extra)
return out
def map_status(status: int) -> str:
if status == 400:
return "validation_error"
if status == 401:
return "auth_required"
if status == 403:
return "no_permission"
if status == 404:
return "resource_unavailable"
if status >= 500:
return "server_error"
return "ok"
def evaluate_capability(abilities: Optional[Dict[str, Any]], action: str) -> Dict[str, Any]:
if not abilities:
return {
"visible": True,
"enabled": True,
"reason": "ok",
"source": "server_status",
}
enabled = bool(abilities.get(action, False)) if action in abilities else True
return {
"visible": True,
"enabled": enabled,
"reason": "ok" if enabled else "no_permission",
"source": "abilities",
}
def _ensure_required(action: str, body: Dict[str, Any]) -> None:
if action in {"jobs.topics.create", "jobs.topics.update"}:
# Keep this loose to be compatible with topic partial updates.
if action.endswith("create") and not body.get("title"):
raise ClientError("jobs.topics.create requires json.title")
if action == "press.posts.create":
if not body.get("title"):
raise ClientError("press.posts.create requires json.title")
if not body.get("body"):
raise ClientError("press.posts.create requires json.body")
if not body.get("summary"):
# Server validates summary presence. Provide a safe default.
source = str(body.get("body") or "").strip() or str(body.get("title") or "")
body["summary"] = source[:120]
if action == "note.notes.create" and not body.get("body"):
raise ClientError("note.notes.create requires json.body")
if action == "site.sites.create":
for key in ("name", "url", "site_node_id"):
if body.get(key) in (None, ""):
raise ClientError(f"site.sites.create requires json.{key}")
def _extract_server_message(payload: Any) -> str:
if isinstance(payload, dict):
for k in ("message", "error", "errors", "raw"):
if k in payload and payload[k] not in (None, ""):
return str(payload[k])
return ""
class GusnaisPluginClient:
def __init__(self, token_store_path: str, timeout: int = 20) -> None:
self.path = Path(token_store_path).expanduser()
if not self.path.exists():
raise ClientError(f"token store not found: {self.path}")
self.store = json.loads(self.path.read_text(encoding="utf-8"))
self.timeout = timeout
self.site = self.store.get("site", "https://gusnais.com").rstrip("/")
self.token = self.store.get("token") or {}
self.client_cfg = self.store.get("client") or {}
self.oauth_cfg = self.store.get("oauth") or {}
def save(self) -> None:
self.store["token"] = self.token
self.path.write_text(json.dumps(self.store, ensure_ascii=False, indent=2), encoding="utf-8")
os.chmod(self.path, 0o600)
def token_expiring(self, skew_seconds: int = 300) -> bool:
expires_at = self.token.get("expires_at")
if not expires_at:
return False
return time.time() >= float(expires_at) - skew_seconds
def refresh(self) -> None:
refresh_token = self.token.get("refresh_token")
client_id = self.client_cfg.get("client_id")
client_secret = self.client_cfg.get("client_secret")
if not (refresh_token and client_id and client_secret):
raise ClientError("missing refresh_token/client_id/client_secret in token store")
token_endpoint = self.oauth_cfg.get("token_endpoint") or f"{self.site}/oauth/token"
res = requests.post(
token_endpoint,
data={
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"client_id": client_id,
"client_secret": client_secret,
},
timeout=self.timeout,
)
if res.status_code >= 400:
raise ClientError(f"refresh failed: {res.status_code} {res.text[:300]}")
payload = res.json()
now = int(time.time())
expires_in = int(payload.get("expires_in") or 0)
self.token.update(
{
"access_token": payload.get("access_token"),
"refresh_token": payload.get("refresh_token") or self.token.get("refresh_token"),
"token_type": payload.get("token_type") or self.token.get("token_type"),
"scope": payload.get("scope") or self.token.get("scope"),
"expires_in": expires_in,
"expires_at": (now + expires_in) if expires_in > 0 else None,
"created_at": payload.get("created_at", now),
}
)
self.save()
def request(self, method: str, path: str, *, query: Optional[Dict[str, Any]] = None, body: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
if self.token_expiring():
self.refresh()
access_token = self.token.get("access_token")
if not access_token:
raise ClientError("missing access_token in token store")
url = f"{self.site}{path}"
params = dict(query or {})
params["access_token"] = access_token
headers = {"Authorization": f"Bearer {access_token}"}
res = requests.request(method, url, params=params, json=body, headers=headers, timeout=self.timeout)
if res.status_code == 401:
self.refresh()
access_token = self.token.get("access_token")
params["access_token"] = access_token
headers["Authorization"] = f"Bearer {access_token}"
res = requests.request(method, url, params=params, json=body, headers=headers, timeout=self.timeout)
data: Any
try:
data = res.json()
except Exception:
data = {"raw": res.text}
out = {
"status": res.status_code,
"reason": map_status(res.status_code),
"ok": 200 <= res.status_code < 300,
"data": data,
}
if not out["ok"]:
msg = _extract_server_message(data)
if msg:
out["message"] = msg
return out
def run_action(self, action: str, *, query: Dict[str, Any], body: Dict[str, Any], target_id: Optional[str]) -> Dict[str, Any]:
if action not in ACTIONS:
raise ClientError(f"unknown action: {action}")
method, path_tmpl, behavior = ACTIONS[action]
path = path_tmpl
if "{id}" in path_tmpl:
if not target_id:
raise ClientError("action requires --id")
path = path_tmpl.replace("{id}", str(target_id))
q = _merge(behavior.get("inject_query", {}), query)
b = _merge(behavior.get("inject_json", {}), body)
_ensure_required(action, b)
result = self.request(method, path, query=q, body=b if method in {"POST", "PUT", "PATCH"} else None)
# Try to infer ability map for common payload shapes.
abilities = None
if isinstance(result.get("data"), dict):
payload = result["data"]
if isinstance(payload.get("abilities"), dict):
abilities = payload.get("abilities")
elif isinstance(payload.get("topic"), dict) and isinstance(payload["topic"].get("abilities"), dict):
abilities = payload["topic"].get("abilities")
elif isinstance(payload.get("post"), dict) and isinstance(payload["post"].get("abilities"), dict):
abilities = payload["post"].get("abilities")
elif isinstance(payload.get("note"), dict) and isinstance(payload["note"].get("abilities"), dict):
abilities = payload["note"].get("abilities")
elif isinstance(payload.get("site"), dict) and isinstance(payload["site"].get("abilities"), dict):
abilities = payload["site"].get("abilities")
verb = action.split(".")[-1]
result["capability"] = evaluate_capability(abilities, verb)
result["request"] = {
"action": action,
"method": method,
"path": path,
"query": q,
"body": b if method in {"POST", "PUT", "PATCH"} else None,
}
return result
def main() -> None:
parser = argparse.ArgumentParser(description="Run Gusnais plugin API actions with auto-refresh")
parser.add_argument("--token-store", required=True, help="Path to token store JSON")
parser.add_argument("--action", required=True, help="Action key, e.g. press.posts.list")
parser.add_argument("--id", help="Resource id for actions with /{id}")
parser.add_argument("--query", default="{}", help="Query JSON object")
parser.add_argument("--json", default="{}", help="Body JSON object")
args = parser.parse_args()
query = _json_loads(args.query)
body = _json_loads(args.json)
client = GusnaisPluginClient(args.token_store)
out = client.run_action(args.action, query=query, body=body, target_id=args.id)
print(json.dumps(out, ensure_ascii=False, indent=2))
if __name__ == "__main__":
try:
main()
except ClientError as err:
print(json.dumps({"ok": False, "reason": "client_error", "error": str(err)}, ensure_ascii=False))
raise SystemExit(1)
FILE:references/endpoints.md
# Gusnais API Endpoint Mapping (Homeland API Compatible)
Base:
- Site: `https://gusnais.com`
- API base: `/api/v3`
OAuth:
- GET `/oauth/authorize`
- POST `/oauth/token`
- POST `/oauth/revoke`
## Root
- GET `/api/v3/hello`
## Users
- GET `/api/v3/users`
- GET `/api/v3/users/:id`
- GET `/api/v3/users/me`
- GET `/api/v3/users/:id/replies`
- GET `/api/v3/users/:id/topics`
- POST `/api/v3/users/:id/block`
- POST `/api/v3/users/:id/unblock`
- GET `/api/v3/users/:id/blocked`
- POST `/api/v3/users/:id/follow`
- POST `/api/v3/users/:id/unfollow`
- GET `/api/v3/users/:id/following`
- GET `/api/v3/users/:id/followers`
- GET `/api/v3/users/:id/favorites`
## Nodes
- GET `/api/v3/nodes`
- GET `/api/v3/nodes/:id`
## Topics
- GET `/api/v3/topics`
- GET `/api/v3/topics/:id`
- POST `/api/v3/topics`
- PUT `/api/v3/topics/:id`
- DELETE `/api/v3/topics/:id`
- POST `/api/v3/topics/:id/replies`
- GET `/api/v3/topics/:id/replies`
- POST `/api/v3/topics/:id/follow`
- POST `/api/v3/topics/:id/unfollow`
- POST `/api/v3/topics/:id/favorite`
- POST `/api/v3/topics/:id/unfavorite`
- POST `/api/v3/topics/:id/action?type=:type`
Topic list params:
- `type`: `last_actived|recent|no_reply|popular|excellent` (default `last_actived`)
- `node_id`: optional node filter
- `offset`: default `0`
- `limit`: default `20`, range `1..150`
## Replies
- GET `/api/v3/replies/:id`
- POST `/api/v3/replies/:id`
- DELETE `/api/v3/replies/:id`
## Photos
- POST `/api/v3/photos` (multipart form, field: `file`)
## Likes
- POST `/api/v3/likes`
- DELETE `/api/v3/likes`
Like params:
- `obj_type`: `topic|reply`
- `obj_id`: integer
## Notifications
- GET `/api/v3/notifications`
- POST `/api/v3/notifications/read`
- GET `/api/v3/notifications/unread_count`
- DELETE `/api/v3/notifications/:id`
- DELETE `/api/v3/notifications/all`
Common pagination:
- `offset` default `0`
- `limit` default `20`
Common status semantics:
- `200/201` success
- `400` validation/request error
- `401` auth missing/expired
- `403` forbidden
- `404` resource not found
- `500` server error
## Plugin API surface (read/write contract)
> Goal: keep plugin capability parity with web endpoints and permission checks.
> On deployments that have not mounted plugin API endpoints yet, these paths may return 404.
### Press (`press-master`)
Web routes observed:
- `GET /posts`
- `GET /posts/upcoming`
- `GET /posts/:id`
- `POST /posts`
- `PUT /posts/:id`
- `DELETE /posts/:id`
- `PUT /posts/:id/publish`
- `POST /posts/preview`
- `GET /admin/posts`
API contract used by this skill:
- `GET /api/v3/press/posts`
- `GET /api/v3/press/posts/upcoming`
- `GET /api/v3/press/posts/:id`
- `POST /api/v3/press/posts`
- `PUT /api/v3/press/posts/:id`
- `DELETE /api/v3/press/posts/:id`
- `PUT /api/v3/press/posts/:id/publish`
- `POST /api/v3/press/posts/preview`
### Note (`note-master`)
Web routes observed:
- `GET /notes`
- `GET /notes/:id`
- `POST /notes`
- `PUT /notes/:id`
- `DELETE /notes/:id`
- `POST /notes/preview`
- `GET /admin/notes`
API contract used by this skill:
- `GET /api/v3/note/notes`
- `GET /api/v3/note/notes/:id`
- `POST /api/v3/note/notes`
- `PUT /api/v3/note/notes/:id`
- `DELETE /api/v3/note/notes/:id`
- `POST /api/v3/note/notes/preview`
### Site (`site-master`)
Web routes observed:
- `GET /sites`
- `POST /sites`
- `GET /admin/sites`
- `POST /admin/sites/:id/undestroy`
- `GET /admin/site_nodes`
API contract used by this skill (deployment-verified):
- `GET /api/v3/site/sites`
- `GET /api/v3/site/sites/:id`
- `POST /api/v3/site/sites`
- `PUT /api/v3/site/sites/:id`
- `DELETE /api/v3/site/sites/:id`
- `GET /api/v3/site/site_nodes`
Notes:
- `undestroy` and site_node write APIs are not mounted in the current deployment and should be treated as unavailable unless server routes are explicitly added later.
- `favorite` field is not present on current `Site` model; client payload should use `{name, desc, url, site_node_id}`.
### Jobs (`jobs-master`)
Jobs plugin is node-driven in web app (`/jobs`) and backed by core Topics at builtin node `25`.
API mapping used by this skill:
- List jobs: `GET /api/v3/topics?node_id=25&type=last_actived`
- Create job topic: `POST /api/v3/topics` with `node_id=25`
- Detail/update/delete: core topic endpoints (`/api/v3/topics/:id`)
FILE:references/permission-parity.md
# Permission Parity Rules (Gusnais)
Goal: keep API interaction behavior consistent with gusnais.com web UX across different account permissions.
## Source priority
1. Resource `abilities` in response payload (highest)
2. HTTP status (`401/403/404`)
3. Static defaults (fallback only)
## UI/action gating
For each actionable resource:
- If abilities indicate deny -> `enabled=false`, keep reason explicit.
- If abilities indicate allow -> `enabled=true`.
- If abilities absent -> allow tentative UI but enforce at submit.
## Must-follow cases
- Topic update/delete and special actions (`ban|excellent|unexcellent|close|open`) must be gated.
- Reply update/delete must be gated when abilities are available via detail payload.
## Denial mapping
- `403` => `no_permission`
- `404` => `resource_unavailable`
- `401` => `auth_required`
## Retry & refresh
- On 401: attempt token refresh once then retry once.
- On repeated 401: force re-auth.
- On 403 after optimistic gating: refresh abilities cache and keep denied.
## Cache policy
- abilities cache TTL: 60-180 seconds.
- Invalidate cache on:
- token refresh
- user switch
- explicit permission failure (403)
## Plugin-specific parity notes
### Press (`press-master`)
- Member can create post.
- Member can update/destroy own post only when status is `upcoming`.
- Admin/maintainer can manage posts and publish.
- Anonymous is read-only (`read`, `upcoming`).
### Note (`note-master`)
- Logged-in user can create notes.
- Logged-in user can read/update/destroy own notes.
- Everyone can read `publish=true` notes.
- Preview is allowed by plugin ability.
### Site (`site-master`)
- Anonymous/read-only for sites.
- Create/update/delete are restricted (current deployment uses admin-only style ability for write actions).
- API exposes `site_nodes` read for client-side validation (`site_node_id` lookup).
- Admin routes (`admin/sites`, `admin/site_nodes`) remain server-enforced.
### Jobs (`jobs-master`)
- Jobs listing is public (`/jobs` view).
- Job posting is constrained by builtin node policy (`node_id=25`):
- default deny for non-admin users;
- allow only HR role or whitelist login.
## Security
- Never attempt hidden admin actions without server-side permission.
- Never infer elevated roles from UI hints alone.
- Never bypass ability checks by direct ID probing.