File size: 7,933 Bytes
4879fc7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 | from __future__ import annotations
from app.config import Settings
from app.errors import InvalidFrontmatter, InvalidPath
from app.naming import (
AGENT_ID_RE,
RESERVED_CHANNEL_NAMES,
SLUG_RE,
SourceURI,
agent_id_from_bucket,
parse_source_uri,
)
BLOCKED_TARGETS = {
"README.md",
"LEADERBOARD.md",
"shared_resources/README.md",
}
BLOCKED_PREFIXES = ("audit/", "inbox/", "taskforces/", "channels/")
# The human-* namespace identifies human participants in inbox routing
# (§16.4): @human-<name> delivers without a registration check, so no agent
# may register inside it (bare "human" included, so it can't be squatted
# either — it routes nowhere).
HUMAN_HANDLE_PREFIX = "human-"
# Per-channel notification level, stored as `notify:` on the membership marker
# (WATCH_DESIGN.md §4.3). Subscription means "I can read this"; the level means
# "this may wake me", and they are deliberately decoupled — joining a channel is
# never a notification commitment, so the default is the quiet one and an
# ABSENT key reads as `mentions` (every pre-existing membership included).
NOTIFY_MENTIONS = "mentions"
NOTIFY_ALL = "all"
NOTIFY_LEVELS = (NOTIFY_MENTIONS, NOTIFY_ALL)
def is_human_handle(handle: str) -> bool:
return handle.startswith(HUMAN_HANDLE_PREFIX) and len(handle) > len(HUMAN_HANDLE_PREFIX)
def validate_notify_level(value: str) -> str:
"""Normalise a caller-supplied `notify` level, rejecting anything else. The
value is written verbatim into marker frontmatter, so a typo must fail loud
rather than silently read back as the quiet default."""
level = value.strip().lower()
if level not in NOTIFY_LEVELS:
raise InvalidFrontmatter(
f"`notify` must be one of {list(NOTIFY_LEVELS)}, got {value!r}"
)
return level
def stored_notify_level(frontmatter: dict) -> str:
"""The level a membership marker's frontmatter *means* — the lenient read
side of `validate_notify_level`'s strict write side. `all` only for an
explicit `notify: all`; an absent (or unrecognised, or hand-edited) value
reads as the quiet default, which is what makes every pre-existing
membership correct without a migration."""
level = str(frontmatter.get("notify", "")).strip().lower()
return level if level in NOTIFY_LEVELS else NOTIFY_MENTIONS
def validate_agent_id(agent_id: str) -> None:
if agent_id != agent_id.lower():
raise InvalidPath(
f"agent_id must be lowercase: {agent_id!r}",
hint=f"use '{agent_id.lower()}' instead",
)
if not AGENT_ID_RE.match(agent_id):
raise InvalidPath(f"invalid agent_id: {agent_id!r}")
def validate_registerable_agent_id(agent_id: str) -> None:
"""Format check plus the reserved-namespace check — registration only.
Read paths (inbox, digest) accept human-* handles, so they use the plain
format check; minting an identity must not be able to squat the namespace.
"""
validate_agent_id(agent_id)
if agent_id == "human" or agent_id.startswith(HUMAN_HANDLE_PREFIX):
raise InvalidPath(
f"agent_id '{agent_id}' is reserved: 'human-<name>' handles identify "
"human participants in inbox routing",
hint="pick an agent_id that does not start with 'human-'",
)
def validate_slug(slug: str) -> None:
if not SLUG_RE.match(slug):
raise InvalidPath(f"invalid slug: {slug!r}")
def validate_path_components(path: str) -> None:
if not path:
raise InvalidPath("empty path")
if path.startswith("/"):
raise InvalidPath("path must not be absolute")
for part in path.rstrip("/").split("/"):
if part in ("", ".", ".."):
raise InvalidPath(f"invalid path component: {part!r}")
if part.startswith("."):
raise InvalidPath(f"path component must not start with '.': {part!r}")
if any(ord(c) < 32 for c in part):
raise InvalidPath("path contains control characters")
def check_dest_not_blocked(target: str) -> None:
norm = target.lstrip("/")
if norm in BLOCKED_TARGETS:
raise InvalidPath(f"target path blocked: {norm}", hint="this path is reserved")
for prefix in BLOCKED_PREFIXES:
if norm.startswith(prefix):
raise InvalidPath(f"target path blocked: {norm}", hint=f"prefix '{prefix}' is reserved")
def resolve_source(settings: Settings, source: str) -> tuple[SourceURI, str]:
"""Parse a source URI and confirm it points inside a valid agent bucket.
Returns (parsed_uri, agent_id). Raises InvalidPath otherwise.
"""
parsed = parse_source_uri(source)
if parsed is None:
raise InvalidPath(f"source must be an hf://buckets/... URI, got: {source!r}")
if parsed.org != settings.org:
raise InvalidPath(
f"source must be under org '{settings.org}', got '{parsed.org}'",
hint="agents post from buckets in this org only",
)
agent_id = agent_id_from_bucket(parsed.bucket, settings.collab_slug)
if agent_id is None:
raise InvalidPath(
f"source bucket '{parsed.bucket}' does not match '{settings.collab_slug}-<agent_id>'",
hint="source must be under your own scratch bucket",
)
if parsed.path:
validate_path_components(parsed.path)
return parsed, agent_id
def _validate_agent_marker(dest_path: str, agent_id: str, what: str) -> None:
"""Attribution-by-construction: the `_{agent_id}` marker must appear in the
dest path, checked against the *resolved* source identity — so only the
same agent can overwrite their own file."""
leaf = dest_path.rsplit("/", 1)[-1]
marker = f"_{agent_id}"
leaf_no_ext = leaf.rsplit(".", 1)[0]
if marker not in leaf_no_ext and marker not in dest_path:
raise InvalidPath(
f"{what} dest path must include '_{agent_id}' in the leaf component",
hint=f"e.g. 'tokenizers/{agent_id}_bpe.json' or 'plots/curve_{agent_id}.png'",
)
def validate_shared_dest_path(dest_path: str, agent_id: str) -> None:
validate_path_components(dest_path)
_validate_agent_marker(dest_path, agent_id, "shared_resources")
full_target = f"shared_resources/{dest_path}"
check_dest_not_blocked(full_target)
def validate_channel_name(name: str) -> None:
if name != name.lower():
raise InvalidPath(
f"channel name must be lowercase: {name!r}",
hint=f"use '{name.lower()}' instead",
)
if not SLUG_RE.match(name):
raise InvalidPath(
f"invalid channel name: {name!r}",
hint="kebab-case, 1-40 chars: [a-z0-9] with internal hyphens",
)
if name in RESERVED_CHANNEL_NAMES:
raise InvalidPath(
f"channel name '{name}' is reserved (it is an API path segment)",
hint="pick a different name",
)
def validate_taskforce_name(name: str) -> None:
if name != name.lower():
raise InvalidPath(
f"taskforce name must be lowercase: {name!r}",
hint=f"use '{name.lower()}' instead",
)
if not SLUG_RE.match(name):
raise InvalidPath(
f"invalid taskforce name: {name!r}",
hint="kebab-case, 1-40 chars: [a-z0-9] with internal hyphens",
)
def validate_taskforce_dest_path(dest_path: str, agent_id: str) -> None:
"""Named taskforce files (§18.3): shared-resources marker rule, plus the
README leaf is reserved for the create/update endpoint."""
validate_path_components(dest_path)
leaf = dest_path.rsplit("/", 1)[-1]
if leaf.lower() == "readme.md":
raise InvalidPath(
"README.md is reserved: the taskforce README is managed via POST /v1/taskforces",
hint="pick a different filename for your content",
)
_validate_agent_marker(dest_path, agent_id, "taskforce")
|