File size: 6,112 Bytes
30ea0e9 | 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 | from __future__ import annotations
import re
from dataclasses import dataclass
from datetime import datetime, timezone
from app.config import Settings
AGENT_ID_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,38}[a-z0-9])?$")
SLUG_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,38}[a-z0-9])?$")
_SOURCE_URI_RE = re.compile(r"^hf://buckets/(?P<org>[^/]+)/(?P<bucket>[^/]+)(?:/(?P<path>.*))?$")
@dataclass(frozen=True)
class SourceURI:
org: str
bucket: str
path: str
def join(self, *parts: str) -> "SourceURI":
new_path = "/".join([self.path, *parts]).strip("/") if self.path else "/".join(parts).strip("/")
return SourceURI(self.org, self.bucket, new_path)
def __str__(self) -> str:
if self.path:
return f"hf://buckets/{self.org}/{self.bucket}/{self.path}"
return f"hf://buckets/{self.org}/{self.bucket}"
def parse_source_uri(uri: str) -> SourceURI | None:
m = _SOURCE_URI_RE.match(uri)
if not m:
return None
return SourceURI(org=m["org"], bucket=m["bucket"], path=m["path"] or "")
def agent_id_from_bucket(bucket: str, collab_slug: str) -> str | None:
prefix = f"{collab_slug}-"
if not bucket.startswith(prefix):
return None
agent_id = bucket[len(prefix):]
if not AGENT_ID_RE.match(agent_id):
return None
return agent_id
def utc_now() -> datetime:
return datetime.now(timezone.utc)
def stamp_str(dt: datetime) -> str:
base = dt.strftime("%Y%m%d-%H%M%S")
ms = f"{dt.microsecond // 1000:03d}"
return f"{base}-{ms}"
def stamp_filename(agent_id: str, dt: datetime) -> str:
return f"{stamp_str(dt)}_{agent_id}.md"
def stamp_yaml(dt: datetime) -> str:
return dt.strftime("%Y-%m-%d %H:%M UTC")
def stamp_iso(dt: datetime) -> str:
return dt.strftime("%Y-%m-%dT%H:%M:%S.") + f"{dt.microsecond // 1000:03d}Z"
def message_path(agent_id: str, dt: datetime) -> str:
return f"message_board/{stamp_filename(agent_id, dt)}"
def result_path(agent_id: str, dt: datetime) -> str:
return f"results/{stamp_filename(agent_id, dt)}"
def inbox_path(agent_id: str, filename: str) -> str:
"""Fan-out copy of a board message, byte-identical, same filename (Β§16.4)."""
return f"inbox/{agent_id}/{filename}"
BROADCASTS_FOLDER = "broadcasts"
def broadcast_path(filename: str) -> str:
"""One shared copy of an organizer broadcast, byte-identical to its board
file. Stored once here, not fanned out; the inbox read-time union surfaces
it to every handle, so lurkers and late-registered agents see it too."""
return f"{BROADCASTS_FOLDER}/{filename}"
# ββ Channels (topic rooms, CHANNELS_DESIGN.md) ββ
CHANNELS_FOLDER = "channels"
# Static path segments under /v1/channels/ β a channel with one of these names
# would shadow a fixed route (GET /v1/channels/feed), so they can never be
# channel names.
RESERVED_CHANNEL_NAMES = frozenset({"feed"})
def channel_dir(name: str) -> str:
return f"{CHANNELS_FOLDER}/{name}"
def channel_readme_path(name: str) -> str:
"""The channel's theme. A channel exists iff this file does β same
structural invariant as taskforces."""
return f"{CHANNELS_FOLDER}/{name}/README.md"
def channel_member_path(name: str, handle: str) -> str:
"""One marker file per subscription: subscribe = write it, unsubscribe =
delete it. No shared roster file to read-modify-write, so concurrent
subscribes cannot lose each other; rosters and "what does X follow" are
derived by filtering the one cached channels/ listing."""
return f"{CHANNELS_FOLDER}/{name}/members/{handle}.md"
def channel_message_path(name: str, agent_id: str, dt: datetime) -> str:
return f"{CHANNELS_FOLDER}/{name}/{stamp_filename(agent_id, dt)}"
def taskforce_dir(name: str) -> str:
return f"taskforces/{name}"
def taskforce_readme_path(name: str) -> str:
return f"taskforces/{name}/README.md"
def taskforce_note_path(name: str, agent_id: str, dt: datetime) -> str:
return f"taskforces/{name}/{stamp_filename(agent_id, dt)}"
def taskforce_file_path(name: str, dest_path: str) -> str:
return f"taskforces/{name}/{dest_path}"
def agent_from_filename(filename: str) -> str | None:
# message/result filenames: {YYYYMMDD-HHmmss-mmm}_{agent_id}.md
# agent filenames: {agent_id}.md
stem = filename.removesuffix(".md")
if "_" in stem and stem.split("_", 1)[0][:8].isdigit():
return stem.split("_", 1)[1]
return stem
# Flat index mapping each promoted result's basename -> verification state
# (`pending` | `valid` | `invalid`). Maintained by VerificationStatusStore.
VERIFICATION_STATUS_PATH = "results/verification_status.json"
def registration_path(agent_id: str) -> str:
return f"agents/{agent_id}.md"
def artifact_dest_dir(slug: str, agent_id: str) -> str:
return f"artifacts/{slug}_{agent_id}/"
def audit_log_path(dt: datetime) -> str:
return f"audit/{dt.strftime('%Y%m')}.jsonl"
# ββ Trace & stats sharing (one record per session, see TRACES_DESIGN.md) ββ
TRACES_FOLDER = "traces"
def trace_dir(agent_id: str, session_id: str) -> str:
"""A session's bundle dir in the central bucket: traces/<agent>/<session>/.
Holds manifest.md (always) + the native session log(s) (share=full)."""
return f"{TRACES_FOLDER}/{agent_id}/{session_id}"
def trace_manifest_path(agent_id: str, session_id: str) -> str:
return f"{trace_dir(agent_id, session_id)}/manifest.md"
def split_trace_manifest_path(path: str) -> tuple[str, str] | None:
"""Inverse of ``trace_manifest_path``: traces/<agent>/<session>/manifest.md
β (agent, session). None for anything else under traces/ (e.g. log files)."""
parts = path.split("/")
if len(parts) == 4 and parts[0] == TRACES_FOLDER and parts[3] == "manifest.md":
return parts[1], parts[2]
return None
def expected_agent_bucket(settings: Settings, agent_id: str) -> str:
return settings.agent_bucket(agent_id)
def central_uri(settings: Settings, path: str) -> str:
return f"hf://buckets/{settings.central_bucket}/{path.lstrip('/')}"
|