feat(security): agent registry — canonical identity + signed tokens (SP-A) - #705
Conversation
…ns (SP-A)
Implements the taOS side of the agent identity/governance design:
- AgentRegistryStore (SQLite, BaseStore pattern): persists canonical agent
records with {canonical_id, display_name, framework, user_id, origin,
handle, role, capabilities, created_ts, revoked_at}.
- Canonical-ID minting: {slug}-{YYYYMMDD}-{HHMMSS} UTC, immutable, with a
2-char hex collision suffix for same-slug same-second duplicates.
- Ed25519 signing keypair: generated once on first boot, private key stored
at data_dir/agent_registry_signing.pem (mode 0600), mirroring the VAPID
keypair approach.
- Signed token: compact EdDSA JWT (header.payload.signature base64url) with
sub/iss/iat claims, issued at registration time.
- Routes: POST /api/agents/registry/register, GET /api/agents/registry/pubkey
(unauthenticated, for bus-side key fetch), GET /api/agents/registry,
GET /api/agents/registry/{id}, DELETE /api/agents/registry/{id}.
- 25 tests: store, keypair idempotency, token round-trip + tamper/wrong-key
rejection, and all five routes.
Bus-side token verification and (user_id, canonical_id) memory keying are
out of scope — those are @taOSmd's pending co-design.
Closes: part of agent-identity/governance epic.
Signed-off-by: jaylfc <jaylfc25@gmail.com>
|
Warning Review limit reached
More reviews will be available in 24 minutes and 23 seconds. Learn how PR review limits work. Your organization has run out of usage credits. Purchase more in the billing tab. ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans include higher PR review limits than trial, open-source, and free plans. In all cases, reviews become available again over time. During sustained high-volume PR review activity, CodeRabbit may temporarily slow when the next review becomes available. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (3)
📝 WalkthroughWalkthroughAdds a SQLite-backed Agent Registry with Ed25519 signing (mint/verify), canonical-id generation with collision handling, FastAPI endpoints (register/pubkey/list/get/revoke), app lifecycle wiring for store/keypair, comprehensive tests, and a design spec document. ChangesAgent Registry Feature
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (2)
tests/test_agent_registry.py (2)
88-93: ⚡ Quick winTighten the collision suffix assertion.
The design here is specifically “2-character hex suffix,” but this test only proves “some suffix exists.” Values like
-1,-xyz, or-0001would still pass. Assert the exact shape so the canonical ID contract stays locked down.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_agent_registry.py` around lines 88 - 93, Tighten the assertion that validates the collision suffix on canonical IDs: after calling store.register (the second call that returns rec) assert rec["canonical_id"] equals base_id + "-" followed by exactly two hex characters (0-9a-f or A-F) rather than just checking startswith; use a regex match or explicit pattern check against rec["canonical_id"] and reference rec, base_id, and the register call to locate the test code to update.
241-267: ⚡ Quick winAdd one unauthenticated
/pubkeytest.
registry_clientalways sends ataos_sessioncookie, so the current coverage only proves authenticated access. Since this endpoint is explicitly meant to stay public for bus-side key fetches, a middleware/auth regression would slip through unnoticed without a no-cookie request.Also applies to: 307-313
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_agent_registry.py` around lines 241 - 267, Add a new test that calls the public pubkey endpoint without authentication: create an ASGITransport(app=app) and AsyncClient(transport=transport, base_url="http://test") with no cookies (i.e., do not reuse the registry_client fixture which sets taos_session), then perform client.get("/pubkey") and assert a successful response and that the body contains the public key value from app.state.agent_registry (or the expected JSON/text). Reference the existing registry_client fixture and the "/pubkey" endpoint to locate where to add this unauthenticated test.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@tinyagentos/agent_registry_store.py`:
- Around line 152-167: The base64 decode calls to _b64url_decode(sig_b64) and
_b64url_decode(payload_b64) can raise non-ValueError exceptions for malformed
tokens; wrap both decodes in a try/except that catches the base64/binary decode
errors (e.g., Exception from the helper) and re-raise a ValueError (e.g.,
ValueError("token must be a valid base64url-encoded JWT")) from None so callers
that only catch ValueError see a normalized error before you call
load_pem_public_key or public_key.verify.
- Around line 77-92: The code can race by reading a partially-written PEM at
pem_path; update the write path so the key is published atomically (or retry
reads until a full PEM parses): after generating private_key and pem_bytes,
write to a temp file (e.g., same directory with a unique suffix) with mode
0o600, fsync the file descriptor, close it, then atomically rename (os.replace)
to pem_path; keep the existing FileExistsError branch to load the winner's key,
and if you prefer the retry approach instead, wrap
load_pem_private_key(pem_path.read_bytes(), ...) in a short retry loop catching
parsing exceptions and OSError, sleeping briefly between attempts before
failing.
- Around line 245-271: The current TOCTOU pre-check loop around canonical_id
(variables base_id, canonical_id, suffix_n and the SELECT) can still race;
instead attempt the INSERT and on UNIQUE/IntegrityError generate a new
canonical_id (e.g., increment suffix_n and set canonical_id =
f"{base_id}-{suffix_n:02x}") and retry the INSERT until it succeeds or a retry
limit is reached. Wrap each failed attempt with appropriate rollback if needed
(use the same self._db.execute for the INSERT into agent_registry and catch the
DB driver's IntegrityError), then commit once the INSERT succeeds.
In `@tinyagentos/app.py`:
- Around line 1229-1232: The comment above the assignments to
app.state.agent_registry and app.state.agent_registry_keypair is misleading:
update it to state that agent_registry_store and agent_registry_keypair are
created in create_app (not in the lifespan) and that the lifespan only
initializes/publishes them; remove the incorrect remark about setting None for
pre-startup existence (unlike mcp_supervisor which is intentionally set to
None). Reference the assignments to app.state.agent_registry =
agent_registry_store and app.state.agent_registry_keypair =
agent_registry_keypair, and mention create_app and lifespan to make the
lifecycle clear.
In `@tinyagentos/routes/agent_registry.py`:
- Around line 119-126: The route revoke_registry_entry currently assumes
store.revoke(canonical_id) returning a record means success, but the store
contract can return the record even if it was already revoked; update
revoke_registry_entry to detect an already-revoked record (e.g., check
record.get("revoked_at") or equivalent on the returned record) and return a 404
with the same error payload when revoked_at is present, while still returning
the revoked response only when revoked_at was newly set; reference the
revoke_registry_entry handler and the store.revoke(canonical_id) call to
implement this guard.
---
Nitpick comments:
In `@tests/test_agent_registry.py`:
- Around line 88-93: Tighten the assertion that validates the collision suffix
on canonical IDs: after calling store.register (the second call that returns
rec) assert rec["canonical_id"] equals base_id + "-" followed by exactly two hex
characters (0-9a-f or A-F) rather than just checking startswith; use a regex
match or explicit pattern check against rec["canonical_id"] and reference rec,
base_id, and the register call to locate the test code to update.
- Around line 241-267: Add a new test that calls the public pubkey endpoint
without authentication: create an ASGITransport(app=app) and
AsyncClient(transport=transport, base_url="http://test") with no cookies (i.e.,
do not reuse the registry_client fixture which sets taos_session), then perform
client.get("/pubkey") and assert a successful response and that the body
contains the public key value from app.state.agent_registry (or the expected
JSON/text). Reference the existing registry_client fixture and the "/pubkey"
endpoint to locate where to add this unauthenticated test.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 8f8c322a-76d4-4f60-b821-3459af698a76
📒 Files selected for processing (5)
tests/test_agent_registry.pytinyagentos/agent_registry_store.pytinyagentos/app.pytinyagentos/routes/__init__.pytinyagentos/routes/agent_registry.py
| if pem_path.exists(): | ||
| private_key = load_pem_private_key(pem_path.read_bytes(), password=None) | ||
| else: | ||
| private_key = Ed25519PrivateKey.generate() | ||
| pem_bytes = private_key.private_bytes( | ||
| Encoding.PEM, PrivateFormat.PKCS8, NoEncryption() | ||
| ) | ||
| try: | ||
| fd = os.open(str(pem_path), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) | ||
| try: | ||
| os.write(fd, pem_bytes) | ||
| finally: | ||
| os.close(fd) | ||
| except FileExistsError: | ||
| # Lost the race — load the winner's key instead. | ||
| private_key = load_pem_private_key(pem_path.read_bytes(), password=None) |
There was a problem hiding this comment.
Avoid loading a half-written PEM.
These branches read agent_registry_signing.pem as soon as the path exists. A competing process can observe the file after O_EXCL creates it but before the writer finishes os.write(...)/os.close(...), which turns first-boot startup into an intermittent PEM deserialization failure. Publish the key atomically, or at least retry reads until a complete PEM parses.
🔧 Minimal mitigation
+ def _load_private_key_with_retry():
+ last_exc = None
+ for _ in range(10):
+ try:
+ return load_pem_private_key(pem_path.read_bytes(), password=None)
+ except ValueError as exc:
+ last_exc = exc
+ time.sleep(0.05)
+ raise last_exc
+
if pem_path.exists():
- private_key = load_pem_private_key(pem_path.read_bytes(), password=None)
+ private_key = _load_private_key_with_retry()
else:
private_key = Ed25519PrivateKey.generate()
pem_bytes = private_key.private_bytes(
Encoding.PEM, PrivateFormat.PKCS8, NoEncryption()
)
@@
except FileExistsError:
# Lost the race — load the winner's key instead.
- private_key = load_pem_private_key(pem_path.read_bytes(), password=None)
+ private_key = _load_private_key_with_retry()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tinyagentos/agent_registry_store.py` around lines 77 - 92, The code can race
by reading a partially-written PEM at pem_path; update the write path so the key
is published atomically (or retry reads until a full PEM parses): after
generating private_key and pem_bytes, write to a temp file (e.g., same directory
with a unique suffix) with mode 0o600, fsync the file descriptor, close it, then
atomically rename (os.replace) to pem_path; keep the existing FileExistsError
branch to load the winner's key, and if you prefer the retry approach instead,
wrap load_pem_private_key(pem_path.read_bytes(), ...) in a short retry loop
catching parsing exceptions and OSError, sleeping briefly between attempts
before failing.
| parts = token.split(".") | ||
| if len(parts) != 3: | ||
| raise ValueError("token must have three dot-separated parts") | ||
|
|
||
| header_b64, payload_b64, sig_b64 = parts | ||
| signing_input = f"{header_b64}.{payload_b64}".encode() | ||
| sig_bytes = _b64url_decode(sig_b64) | ||
|
|
||
| public_key = load_pem_public_key(public_key_pem) | ||
| try: | ||
| public_key.verify(sig_bytes, signing_input) | ||
| except InvalidSignature: | ||
| raise ValueError("token signature verification failed") from None | ||
|
|
||
| payload = json.loads(_b64url_decode(payload_b64)) | ||
| return payload |
There was a problem hiding this comment.
Normalize malformed token input to ValueError.
_b64url_decode(sig_b64) can raise a raw base64 decoding exception here, even though this helper promises ValueError for invalid tokens. If callers only trap ValueError, a garbage token becomes a 500 instead of a clean verification failure.
🔧 Suggested fix
def verify_registry_token(token: str, public_key_pem: bytes) -> dict:
@@
+ import binascii
+
parts = token.split(".")
if len(parts) != 3:
raise ValueError("token must have three dot-separated parts")
header_b64, payload_b64, sig_b64 = parts
signing_input = f"{header_b64}.{payload_b64}".encode()
- sig_bytes = _b64url_decode(sig_b64)
+ try:
+ sig_bytes = _b64url_decode(sig_b64)
+ payload_bytes = _b64url_decode(payload_b64)
+ except (binascii.Error, ValueError) as exc:
+ raise ValueError("token is not valid base64url") from exc
public_key = load_pem_public_key(public_key_pem)
@@
- payload = json.loads(_b64url_decode(payload_b64))
+ try:
+ payload = json.loads(payload_bytes)
+ except ValueError as exc:
+ raise ValueError("token payload is not valid JSON") from exc
return payload🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tinyagentos/agent_registry_store.py` around lines 152 - 167, The base64
decode calls to _b64url_decode(sig_b64) and _b64url_decode(payload_b64) can
raise non-ValueError exceptions for malformed tokens; wrap both decodes in a
try/except that catches the base64/binary decode errors (e.g., Exception from
the helper) and re-raise a ValueError (e.g., ValueError("token must be a valid
base64url-encoded JWT")) from None so callers that only catch ValueError see a
normalized error before you call load_pem_public_key or public_key.verify.
| # Collision guard: if the same slug+second already exists, append a | ||
| # 2-char hex suffix to break the tie. | ||
| suffix_n = 0 | ||
| while True: | ||
| existing = await ( | ||
| await self._db.execute( | ||
| "SELECT id FROM agent_registry WHERE canonical_id = ?", | ||
| (canonical_id,), | ||
| ) | ||
| ).fetchone() | ||
| if existing is None: | ||
| break | ||
| suffix_n += 1 | ||
| canonical_id = f"{base_id}-{suffix_n:02x}" | ||
|
|
||
| caps_json = json.dumps(capabilities) | ||
| await self._db.execute( | ||
| """ | ||
| INSERT INTO agent_registry | ||
| (canonical_id, display_name, framework, user_id, origin, | ||
| handle, role, capabilities, created_ts) | ||
| VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) | ||
| """, | ||
| (canonical_id, display_name, framework, user_id, origin, | ||
| handle, role, caps_json, created_ts), | ||
| ) | ||
| await self._db.commit() |
There was a problem hiding this comment.
Retry on INSERT conflicts instead of pre-checking.
This collision guard is a TOCTOU. The existence check and INSERT are separated by several awaits, so two same-second registrations can both see the same canonical_id as free and the loser then hits the UNIQUE constraint instead of getting the documented hex-suffix fallback. The retry loop needs to be driven by INSERT conflicts, not the preflight SELECT.
🔧 Suggested fix
# Collision guard: if the same slug+second already exists, append a
# 2-char hex suffix to break the tie.
suffix_n = 0
- while True:
- existing = await (
- await self._db.execute(
- "SELECT id FROM agent_registry WHERE canonical_id = ?",
- (canonical_id,),
- )
- ).fetchone()
- if existing is None:
- break
- suffix_n += 1
- canonical_id = f"{base_id}-{suffix_n:02x}"
-
caps_json = json.dumps(capabilities)
- await self._db.execute(
- """
- INSERT INTO agent_registry
- (canonical_id, display_name, framework, user_id, origin,
- handle, role, capabilities, created_ts)
- VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
- """,
- (canonical_id, display_name, framework, user_id, origin,
- handle, role, caps_json, created_ts),
- )
+ while True:
+ try:
+ await self._db.execute(
+ """
+ INSERT INTO agent_registry
+ (canonical_id, display_name, framework, user_id, origin,
+ handle, role, capabilities, created_ts)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+ """,
+ (canonical_id, display_name, framework, user_id, origin,
+ handle, role, caps_json, created_ts),
+ )
+ break
+ except aiosqlite.IntegrityError:
+ suffix_n += 1
+ canonical_id = f"{base_id}-{suffix_n:02x}"
await self._db.commit()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tinyagentos/agent_registry_store.py` around lines 245 - 271, The current
TOCTOU pre-check loop around canonical_id (variables base_id, canonical_id,
suffix_n and the SELECT) can still race; instead attempt the INSERT and on
UNIQUE/IntegrityError generate a new canonical_id (e.g., increment suffix_n and
set canonical_id = f"{base_id}-{suffix_n:02x}") and retry the INSERT until it
succeeds or a retry limit is reached. Wrap each failed attempt with appropriate
rollback if needed (use the same self._db.execute for the INSERT into
agent_registry and catch the DB driver's IntegrityError), then commit once the
INSERT succeeds.
| # agent_registry and its keypair are created by the lifespan; None here | ||
| # ensures attribute-existence checks work during the pre-startup window. | ||
| app.state.agent_registry = agent_registry_store | ||
| app.state.agent_registry_keypair = agent_registry_keypair |
There was a problem hiding this comment.
Correct the misleading comment.
The comment states that agent_registry and its keypair "are created by the lifespan" but they are actually created in create_app (lines 245-246), not in the lifespan. The lifespan only initializes and publishes them (lines 371-373).
The comment also claims "None here ensures attribute-existence checks work during the pre-startup window" but the code assigns the actual objects, not None. Compare this to truly lifespan-created objects like mcp_supervisor (line 1217) which are set to None in the eager section.
📝 Proposed fix to align comment with code
- # agent_registry and its keypair are created by the lifespan; None here
- # ensures attribute-existence checks work during the pre-startup window.
+ # agent_registry and its keypair are created in create_app and initialized
+ # by the lifespan; eagerly set here so routes can check these attributes.
app.state.agent_registry = agent_registry_store
app.state.agent_registry_keypair = agent_registry_keypair📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| # agent_registry and its keypair are created by the lifespan; None here | |
| # ensures attribute-existence checks work during the pre-startup window. | |
| app.state.agent_registry = agent_registry_store | |
| app.state.agent_registry_keypair = agent_registry_keypair | |
| # agent_registry and its keypair are created in create_app and initialized | |
| # by the lifespan; eagerly set here so routes can check these attributes. | |
| app.state.agent_registry = agent_registry_store | |
| app.state.agent_registry_keypair = agent_registry_keypair |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tinyagentos/app.py` around lines 1229 - 1232, The comment above the
assignments to app.state.agent_registry and app.state.agent_registry_keypair is
misleading: update it to state that agent_registry_store and
agent_registry_keypair are created in create_app (not in the lifespan) and that
the lifespan only initializes/publishes them; remove the incorrect remark about
setting None for pre-startup existence (unlike mcp_supervisor which is
intentionally set to None). Reference the assignments to
app.state.agent_registry = agent_registry_store and
app.state.agent_registry_keypair = agent_registry_keypair, and mention
create_app and lifespan to make the lifecycle clear.
| @router.delete("/api/agents/registry/{canonical_id}") | ||
| async def revoke_registry_entry(request: Request, canonical_id: str): | ||
| """Revoke a registry entry (sets revoked_at, does not delete).""" | ||
| store = _get_store(request) | ||
| record = await store.revoke(canonical_id) | ||
| if record is None: | ||
| return JSONResponse({"error": "not found or already revoked"}, status_code=404) | ||
| return {"status": "revoked", "canonical_id": canonical_id, "revoked_at": record.get("revoked_at")} |
There was a problem hiding this comment.
Already-revoked deletes can still return 200.
tests/test_agent_registry.py currently codifies AgentRegistryStore.revoke() returning a record on the second call, so this None check only catches missing IDs. In that case, a repeated DELETE incorrectly returns success instead of the documented 404 for an already-revoked entry.
Suggested route-side guard if you want to keep the current store contract
`@router.delete`("/api/agents/registry/{canonical_id}")
async def revoke_registry_entry(request: Request, canonical_id: str):
"""Revoke a registry entry (sets revoked_at, does not delete)."""
store = _get_store(request)
- record = await store.revoke(canonical_id)
- if record is None:
+ existing = await store.get(canonical_id)
+ if existing is None or existing.get("revoked_at") is not None:
return JSONResponse({"error": "not found or already revoked"}, status_code=404)
+ record = await store.revoke(canonical_id)
return {"status": "revoked", "canonical_id": canonical_id, "revoked_at": record.get("revoked_at")}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tinyagentos/routes/agent_registry.py` around lines 119 - 126, The route
revoke_registry_entry currently assumes store.revoke(canonical_id) returning a
record means success, but the store contract can return the record even if it
was already revoked; update revoke_registry_entry to detect an already-revoked
record (e.g., check record.get("revoked_at") or equivalent on the returned
record) and return a 404 with the same error payload when revoked_at is present,
while still returning the revoked response only when revoked_at was newly set;
reference the revoke_registry_entry handler and the store.revoke(canonical_id)
call to implement this guard.
…er, generic-verify tests + SP-A spec
- mint_registry_token() now accepts user_id and framework keyword args and
embeds them as claims in the JWT payload alongside sub/iss/iat. The JWT
header is serialised with compact separators to exactly {"alg":"EdDSA","typ":"JWT"}.
- register route passes record["user_id"] and record["framework"] into the
token mint so the issued token carries the registering agent's identity.
- Tests: assert user_id + framework claims on a minted token; add a generic
Ed25519 verify round-trip (cryptography lib only, no tinyagentos import) to
prove taOSmd can verify tokens independently. All 27 tests pass (25 original
+ 2 new).
- docs/superpowers/specs/2026-06-09-agent-registry-sp-a-design.md: SP-A spec
documenting the claim set, standard compact EdDSA JWT format, GET /pubkey
bus-side fetch, Bearer-token presentation, sub==from check, and opt-in
verification (bus skips when no pubkey configured).
Signed-off-by: jaylfc <jaylfc25@gmail.com>
…A/B hardening) Signed-off-by: jaylfc <jaylfc25@gmail.com>
There was a problem hiding this comment.
🧹 Nitpick comments (3)
tests/test_agent_registry.py (2)
261-261: ⚡ Quick winTypo in test name:
edsashould beeddsa.The algorithm being verified is EdDSA (Edwards-curve Digital Signature Algorithm), but the test name uses
edsa. For clarity and consistency with the algorithm name, rename totest_generic_eddsa_verify_without_our_helperortest_generic_ed25519_verify_without_our_helper.📝 Proposed fix
- def test_generic_edsa_verify_without_our_helper(self, tmp_path): + def test_generic_eddsa_verify_without_our_helper(self, tmp_path):🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_agent_registry.py` at line 261, Rename the test function test_generic_edsa_verify_without_our_helper to test_generic_eddsa_verify_without_our_helper (or test_generic_ed25519_verify_without_our_helper) to correct the algorithm name; update the function definition and any references (imports, parametrizations, test IDs, or markers) that use the old name so pytest still discovers it and any related assertions/docstrings reflect "eddsa"/"ed25519".
261-317: 💤 Low valueConsider extracting base64 padding helper.
The base64 padding logic (lines 281-285, 293-297, 307-311) is duplicated within this test and across other tests (lines 226-228, 252-254). Extracting a helper function would reduce duplication and improve maintainability.
♻️ Example refactor
def _add_b64_padding(b64_str: str) -> str: """Add RFC 4648 padding to base64url string.""" padding = 4 - len(b64_str) % 4 return b64_str if padding == 4 else b64_str + "=" * padding # Usage: header_b64_padded = _add_b64_padding(header_b64) header = json.loads(base64.urlsafe_b64decode(header_b64_padded))🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_agent_registry.py` around lines 261 - 317, The duplicated base64url padding logic in test_generic_edsa_verify_without_our_helper (and similar blocks in the same test file) should be extracted into a small helper function (e.g., _add_b64_padding(b64_str: str) -> str) and used everywhere the padding calculation appears; replace the three local padding blocks in test_generic_edsa_verify_without_our_helper (header_b64, sig_b64, payload_b64) with calls to that helper and update other tests in tests/test_agent_registry.py that repeat the same pattern to call the helper as well so the padding logic is centralized and de-duplicated.docs/superpowers/specs/2026-06-09-agent-registry-sp-a-design.md (1)
47-52: 💤 Low valueAdd language specifiers to fenced code blocks for better rendering.
Several code blocks are missing language identifiers, which improves syntax highlighting and readability in most markdown renderers. Consider adding appropriate language tags:
- Line 47: no language needed (generic format example) or use
text- Line 128: use
httporbash- Line 147: use
pythonorpseudocode- Line 158: use
http- Line 183: use
textor no language (format template)Example fixes
Line 128:
-``` +```http GET http://<taos-host>/api/agents/registry/pubkey → public_key (PEM) ```Line 158:
-``` +```http Authorization: Bearer <compact-jwt> ```Also applies to: 128-131, 147-149, 158-160, 183-185
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@docs/superpowers/specs/2026-06-09-agent-registry-sp-a-design.md` around lines 47 - 52, Add explicit language specifiers to the fenced code blocks referenced in the spec: mark the JWT format block (the "base64url(header) . base64url(payload) . base64url(signature)" example) as text (or leave untagged), annotate the HTTP request example that shows "GET http://<taos-host>/api/agents/registry/pubkey" with ```http, tag the pseudocode/python example with ```python (or ```pseudocode), label the Authorization header snippet ("Authorization: Bearer <compact-jwt>") with ```http, and mark the format/template block near the end as text (or leave untagged); update the corresponding fenced code fences accordingly so markdown renderers get appropriate syntax highlighting.Source: Linters/SAST tools
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@docs/superpowers/specs/2026-06-09-agent-registry-sp-a-design.md`:
- Around line 47-52: Add explicit language specifiers to the fenced code blocks
referenced in the spec: mark the JWT format block (the "base64url(header) .
base64url(payload) . base64url(signature)" example) as text (or leave untagged),
annotate the HTTP request example that shows "GET
http://<taos-host>/api/agents/registry/pubkey" with ```http, tag the
pseudocode/python example with ```python (or ```pseudocode), label the
Authorization header snippet ("Authorization: Bearer <compact-jwt>") with
```http, and mark the format/template block near the end as text (or leave
untagged); update the corresponding fenced code fences accordingly so markdown
renderers get appropriate syntax highlighting.
In `@tests/test_agent_registry.py`:
- Line 261: Rename the test function test_generic_edsa_verify_without_our_helper
to test_generic_eddsa_verify_without_our_helper (or
test_generic_ed25519_verify_without_our_helper) to correct the algorithm name;
update the function definition and any references (imports, parametrizations,
test IDs, or markers) that use the old name so pytest still discovers it and any
related assertions/docstrings reflect "eddsa"/"ed25519".
- Around line 261-317: The duplicated base64url padding logic in
test_generic_edsa_verify_without_our_helper (and similar blocks in the same test
file) should be extracted into a small helper function (e.g.,
_add_b64_padding(b64_str: str) -> str) and used everywhere the padding
calculation appears; replace the three local padding blocks in
test_generic_edsa_verify_without_our_helper (header_b64, sig_b64, payload_b64)
with calls to that helper and update other tests in tests/test_agent_registry.py
that repeat the same pattern to call the helper as well so the padding logic is
centralized and de-duplicated.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 400f362f-ade8-4800-b72c-723e843e7698
📒 Files selected for processing (4)
docs/superpowers/specs/2026-06-09-agent-registry-sp-a-design.mdtests/test_agent_registry.pytinyagentos/agent_registry_store.pytinyagentos/routes/agent_registry.py
🚧 Files skipped from review as they are similar to previous changes (2)
- tinyagentos/routes/agent_registry.py
- tinyagentos/agent_registry_store.py
Summary
Implements SP-A (taOS side) of the agent identity/governance design.
BaseStorepattern): records{canonical_id, display_name, framework, user_id, origin, handle, role, capabilities, created_ts, revoked_at}.{slug}-{YYYYMMDD}-{HHMMSS}UTC, immutable, collision-safe with a 2-char hex suffix on same-slug same-second duplicates.data_dir/agent_registry_signing.pem(mode 0600), mirroring the VAPID keypair approach.header.payload.signature, base64url) withsub/iss/iatclaims, issued at registration time./api/agents/registry/):POST .../register— mint ID + issue tokenGET .../pubkey— unauthenticated, for bus-side key fetchGET ...— list all entriesGET .../{id}— read one entryDELETE .../{id}— soft-revokeTest results
25/25 tests pass (
tests/test_agent_registry.py). Full suite (1254 tests) unaffected; the singletest_agent_bridge::test_exec_commandSIGSEGV is a pre-existing macOS Python 3.14 fork-safety issue documented in conftest.py and passes in isolation.Design doc
docs/superpowers/specs/2026-06-09-agent-registry-sp-a-design.md— not committed (gitignored per repo convention for AI planning docs), but content is reproduced in this PR description.Explicit out-of-scope note
Bus-side token verification and
(user_id, canonical_id)memory keying are @taOSmd's pending co-design. The public key endpoint (/api/agents/registry/pubkey) andverify_registry_token()(importable fromtinyagentos.agent_registry_store) are ready for taOSmd to consume.Files changed
tinyagentos/agent_registry_store.pytinyagentos/routes/agent_registry.pytinyagentos/app.pytinyagentos/routes/__init__.py/api/agents/{name})tests/test_agent_registry.pySummary by CodeRabbit
New Features
Documentation
Tests