feat(manifest): trusted agent forge identity and guidance
Implements PRD prd-new-trusted-agent-forge-identity. Moves author identity and named forge configurations onto the trusted, host-only agent definition, and lets a bottle repository optionally associate a git-gate repo with one of the agent's forge aliases. - Agent-owned identity: new `author` (name/email) and `forge-accounts` (alias -> canonical Gitea /api/v1 origin + host `token_secret` ref) on the agent manifest. `author` populates the bottle's git user.name/user.email. - Remove `git-gate.user` from both agents and bottles; fail with a migration pointer to `author`. `git-gate` is no longer accepted on an agent. - Bottle git-gate repos gain optional `forge: <alias>`, resolved against the selected agent's forge-accounts at composition (fail-closed on unknown). - Fail-closed Gitea API URL validation (https, no userinfo/query/fragment, /api/v1 base, host lowercased, trailing slash normalized, host dedup). - Proxy-held credential: synthesize one inspected, token-authenticated egress route per referenced forge alias, scoped to the origin + API prefix. The token is resolved from the host env at launch into the egress proxy only — never the bottle env, prompt, gitconfig, or workspace. - Generated, non-secret forge workflow guidance appended to the agent prompt for associated repos (API base, branch-backed PR flow, AGit-ref prohibition, mutation verification); omitted when no repo declares a forge. - Agents become home-only: cwd `.bot-bottle/agents` no longer contributes, overrides, or is selectable; warned-and-ignored like cwd bottles. - Docs: README examples, PRD 0011 supersession note, manifest schema docstring. - Tests: new test_forge_identity suite; legacy git-gate.user/cwd tests updated to the agent-owned identity model. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,387 @@
|
||||
"""Unit: trusted agent forge identity & guidance
|
||||
(PRD prd-new-trusted-agent-forge-identity).
|
||||
|
||||
Covers the net-new surface: agent `author`, agent `forge-accounts` (Gitea API
|
||||
URL validation + host token reference), the repo→forge association resolved at
|
||||
composition, the synthesized proxy-held egress route, and the generated,
|
||||
non-secret prompt guidance. Legacy git-gate.user / cwd-agent behavior lives in
|
||||
the manifest test modules.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from typing import Callable
|
||||
from unittest.mock import patch
|
||||
|
||||
from bot_bottle.egress import (
|
||||
egress_forge_routes,
|
||||
egress_render_routes,
|
||||
egress_resolve_token_values,
|
||||
egress_routes_for_bottle,
|
||||
egress_token_env_map,
|
||||
)
|
||||
from bot_bottle.manifest import ManifestError, ManifestIndex
|
||||
from bot_bottle.manifest.forge import (
|
||||
ManifestForgeAccount,
|
||||
canonicalize_forge_url,
|
||||
render_forge_guidance,
|
||||
)
|
||||
|
||||
|
||||
GITEA = {
|
||||
"url": "https://gitea.dideric.is/api/v1",
|
||||
"auth": {"type": "token", "token_secret": "GITEA_CLAUDE_TOKEN"},
|
||||
}
|
||||
|
||||
|
||||
def _repo(*, forge: str | None = None) -> dict[str, object]:
|
||||
entry: dict[str, object] = {
|
||||
"url": "ssh://git@100.78.141.42:30009/didericis/bot-bottle.git",
|
||||
"key": {"provider": "static", "path": "/k"},
|
||||
}
|
||||
if forge is not None:
|
||||
entry["forge"] = forge
|
||||
return entry
|
||||
|
||||
|
||||
def _index(
|
||||
*,
|
||||
author: dict[str, object] | None = None,
|
||||
forge_accounts: dict[str, object] | None = None,
|
||||
repo_forge: str | None = None,
|
||||
) -> ManifestIndex:
|
||||
"""Build an eager index with one agent 'claude' + bottle 'bb'."""
|
||||
agent: dict[str, object] = {"bottle": "bb", "prompt": ""}
|
||||
if author is not None:
|
||||
agent["author"] = author
|
||||
if forge_accounts is not None:
|
||||
agent["forge-accounts"] = forge_accounts
|
||||
bottle: dict[str, object] = {
|
||||
"git-gate": {"repos": {"bot-bottle": _repo(forge=repo_forge)}}
|
||||
}
|
||||
return ManifestIndex.from_json_obj(
|
||||
{"bottles": {"bb": bottle}, "agents": {"claude": agent}}
|
||||
)
|
||||
|
||||
|
||||
def _error(
|
||||
callable_: Callable[..., object], *args: object, **kwargs: object,
|
||||
) -> str:
|
||||
try:
|
||||
callable_(*args, **kwargs)
|
||||
except ManifestError as e:
|
||||
return str(e)
|
||||
raise AssertionError("expected ManifestError was not raised")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# author
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestAuthor(unittest.TestCase):
|
||||
def test_populates_git_identity(self) -> None:
|
||||
idx = _index(author={"name": "didericis-claude", "email": "e+c@x.is"})
|
||||
m = idx.load_for_agent("claude", ())
|
||||
self.assertEqual("didericis-claude", m.bottle.git_user.name)
|
||||
self.assertEqual("e+c@x.is", m.bottle.git_user.email)
|
||||
self.assertEqual(
|
||||
"name=didericis-claude, email=e+c@x.is",
|
||||
m.git_identity_summary(),
|
||||
)
|
||||
|
||||
def test_absent_author_no_identity(self) -> None:
|
||||
m = _index().load_for_agent("claude", ())
|
||||
self.assertTrue(m.bottle.git_user.is_empty())
|
||||
self.assertIsNone(m.git_identity_summary())
|
||||
|
||||
def test_name_required(self) -> None:
|
||||
msg = _error(_index, author={"email": "e@x.is"})
|
||||
self.assertIn("author.name must be a non-empty string", msg)
|
||||
|
||||
def test_email_required(self) -> None:
|
||||
msg = _error(_index, author={"name": "n"})
|
||||
self.assertIn("author.email must be a non-empty string", msg)
|
||||
|
||||
def test_email_rejects_whitespace(self) -> None:
|
||||
msg = _error(_index, author={"name": "n", "email": "a b@x.is"})
|
||||
self.assertIn("must not contain whitespace", msg)
|
||||
|
||||
def test_name_rejects_newline(self) -> None:
|
||||
msg = _error(_index, author={"name": "a\nb", "email": "e@x.is"})
|
||||
self.assertIn("author.name must not contain newlines", msg)
|
||||
|
||||
def test_unknown_key(self) -> None:
|
||||
msg = _error(
|
||||
_index, author={"name": "n", "email": "e@x.is", "role": "x"}
|
||||
)
|
||||
self.assertIn("author has unknown key", msg)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# forge-accounts URL validation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestForgeUrl(unittest.TestCase):
|
||||
def test_canonical_ok(self) -> None:
|
||||
canonical, origin, host, prefix = canonicalize_forge_url(
|
||||
"a", "g", "https://Gitea.Dideric.is/api/v1/"
|
||||
)
|
||||
# trailing slash normalized; host lowercased.
|
||||
self.assertEqual("https://gitea.dideric.is/api/v1", canonical)
|
||||
self.assertEqual("https://gitea.dideric.is", origin)
|
||||
self.assertEqual("gitea.dideric.is", host)
|
||||
self.assertEqual("/api/v1", prefix)
|
||||
|
||||
def test_port_preserved(self) -> None:
|
||||
canonical, origin, _host, _ = canonicalize_forge_url(
|
||||
"a", "g", "https://gitea.local:3000/api/v1"
|
||||
)
|
||||
self.assertEqual("https://gitea.local:3000/api/v1", canonical)
|
||||
self.assertEqual("https://gitea.local:3000", origin)
|
||||
|
||||
def test_http_fails(self) -> None:
|
||||
self.assertIn("must use https", _error(
|
||||
canonicalize_forge_url, "a", "g", "http://gitea/api/v1"))
|
||||
|
||||
def test_userinfo_fails(self) -> None:
|
||||
self.assertIn("userinfo", _error(
|
||||
canonicalize_forge_url, "a", "g", "https://u:p@gitea/api/v1"))
|
||||
|
||||
def test_query_fails(self) -> None:
|
||||
self.assertIn("query string", _error(
|
||||
canonicalize_forge_url, "a", "g", "https://gitea/api/v1?x=1"))
|
||||
|
||||
def test_fragment_fails(self) -> None:
|
||||
self.assertIn("fragment", _error(
|
||||
canonicalize_forge_url, "a", "g", "https://gitea/api/v1#x"))
|
||||
|
||||
def test_missing_host_fails(self) -> None:
|
||||
self.assertIn("hostname", _error(
|
||||
canonicalize_forge_url, "a", "g", "https:///api/v1"))
|
||||
|
||||
def test_bad_path_fails(self) -> None:
|
||||
self.assertIn("Gitea API base", _error(
|
||||
canonicalize_forge_url, "a", "g", "https://gitea/api/v2"))
|
||||
|
||||
def test_non_string_fails(self) -> None:
|
||||
self.assertIn("required", _error(
|
||||
canonicalize_forge_url, "a", "g", None))
|
||||
|
||||
def test_malformed_url_fails(self) -> None:
|
||||
# An unparseable URL (invalid IPv6 literal) trips urlsplit's ValueError.
|
||||
self.assertIn("not a valid URL", _error(
|
||||
canonicalize_forge_url, "a", "g", "https://[/api/v1"))
|
||||
|
||||
|
||||
class TestForgeAccount(unittest.TestCase):
|
||||
def test_parses(self) -> None:
|
||||
acct = ManifestForgeAccount.from_dict("a", "didericis-gitea", GITEA)
|
||||
self.assertEqual("didericis-gitea", acct.alias)
|
||||
self.assertEqual("gitea.dideric.is", acct.host)
|
||||
self.assertEqual("token", acct.auth_type)
|
||||
self.assertEqual("GITEA_CLAUDE_TOKEN", acct.token_secret)
|
||||
|
||||
def test_non_kebab_alias_fails(self) -> None:
|
||||
self.assertIn("valid alias", _error(
|
||||
ManifestForgeAccount.from_dict, "a", "Bad_Alias", GITEA))
|
||||
|
||||
def test_auth_required(self) -> None:
|
||||
self.assertIn("missing required 'auth'", _error(
|
||||
ManifestForgeAccount.from_dict, "a", "g",
|
||||
{"url": "https://gitea/api/v1"}))
|
||||
|
||||
def test_bad_auth_type_fails(self) -> None:
|
||||
self.assertIn("auth.type must be", _error(
|
||||
ManifestForgeAccount.from_dict, "a", "g",
|
||||
{"url": "https://gitea/api/v1",
|
||||
"auth": {"type": "basic", "token_secret": "T"}}))
|
||||
|
||||
def test_token_secret_required(self) -> None:
|
||||
self.assertIn("token_secret must be", _error(
|
||||
ManifestForgeAccount.from_dict, "a", "g",
|
||||
{"url": "https://gitea/api/v1", "auth": {"type": "token"}}))
|
||||
|
||||
def test_unknown_key_fails(self) -> None:
|
||||
self.assertIn("unknown key", _error(
|
||||
ManifestForgeAccount.from_dict, "a", "g",
|
||||
{**GITEA, "bogus": 1}))
|
||||
|
||||
def test_unknown_auth_key_fails(self) -> None:
|
||||
self.assertIn("auth has unknown key", _error(
|
||||
ManifestForgeAccount.from_dict, "a", "g",
|
||||
{"url": "https://gitea/api/v1",
|
||||
"auth": {"type": "token", "token_secret": "T", "bogus": 1}}))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# repo -> forge association (composition)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestForgeAssociations(unittest.TestCase):
|
||||
def test_resolves_when_repo_references_alias(self) -> None:
|
||||
idx = _index(
|
||||
forge_accounts={"didericis-gitea": GITEA},
|
||||
repo_forge="didericis-gitea",
|
||||
)
|
||||
m = idx.load_for_agent("claude", ())
|
||||
self.assertEqual(1, len(m.forge_associations))
|
||||
assoc = m.forge_associations[0]
|
||||
self.assertEqual("didericis-gitea", assoc.alias)
|
||||
self.assertEqual(("bot-bottle",), assoc.repo_names)
|
||||
|
||||
def test_unreferenced_account_yields_no_association(self) -> None:
|
||||
idx = _index(forge_accounts={"didericis-gitea": GITEA}) # repo has no forge
|
||||
m = idx.load_for_agent("claude", ())
|
||||
self.assertEqual((), m.forge_associations)
|
||||
|
||||
def test_unknown_alias_fails_closed(self) -> None:
|
||||
idx = _index(
|
||||
forge_accounts={"didericis-gitea": GITEA}, repo_forge="nope"
|
||||
)
|
||||
msg = _error(idx.load_for_agent, "claude", ())
|
||||
self.assertIn("references forge alias 'nope'", msg)
|
||||
self.assertIn("not defined on agent", msg)
|
||||
|
||||
def test_forge_without_account_fails_closed(self) -> None:
|
||||
idx = _index(repo_forge="didericis-gitea") # no forge-accounts at all
|
||||
self.assertIn("(none)", _error(idx.load_for_agent, "claude", ()))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# synthesized egress route (proxy-held credential)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestForgeEgressRoutes(unittest.TestCase):
|
||||
def _assoc(self):
|
||||
idx = _index(
|
||||
forge_accounts={"didericis-gitea": GITEA},
|
||||
repo_forge="didericis-gitea",
|
||||
)
|
||||
return idx.load_for_agent("claude", ())
|
||||
|
||||
def test_route_shape(self) -> None:
|
||||
routes = egress_forge_routes(self._assoc().forge_associations)
|
||||
self.assertEqual(1, len(routes))
|
||||
r = routes[0]
|
||||
self.assertEqual("gitea.dideric.is", r.host)
|
||||
self.assertEqual("token", r.auth_scheme)
|
||||
self.assertEqual("GITEA_CLAUDE_TOKEN", r.token_ref)
|
||||
self.assertTrue(r.inspect)
|
||||
# scoped to the API prefix.
|
||||
self.assertEqual("/api/v1", r.matches[0].paths[0].value)
|
||||
|
||||
def test_token_slot_and_resolution(self) -> None:
|
||||
m = self._assoc()
|
||||
forge_routes = egress_forge_routes(m.forge_associations)
|
||||
routes = egress_routes_for_bottle(m.bottle, (), forge_routes)
|
||||
token_map = egress_token_env_map(routes)
|
||||
# one slot mapping the proxy env slot -> host env var name.
|
||||
self.assertEqual({"EGRESS_TOKEN_0": "GITEA_CLAUDE_TOKEN"}, token_map)
|
||||
resolved = egress_resolve_token_values(
|
||||
token_map, {"GITEA_CLAUDE_TOKEN": "sekret-value"}
|
||||
)
|
||||
self.assertEqual({"EGRESS_TOKEN_0": "sekret-value"}, resolved)
|
||||
|
||||
def test_rendered_routes_hide_secret_name_and_value(self) -> None:
|
||||
m = self._assoc()
|
||||
routes = egress_routes_for_bottle(
|
||||
m.bottle, (), egress_forge_routes(m.forge_associations)
|
||||
)
|
||||
rendered = egress_render_routes(routes)
|
||||
self.assertIn("gitea.dideric.is", rendered)
|
||||
self.assertIn("EGRESS_TOKEN_0", rendered) # slot, not the host var name
|
||||
self.assertNotIn("GITEA_CLAUDE_TOKEN", rendered)
|
||||
self.assertNotIn("sekret-value", rendered)
|
||||
|
||||
def test_dedup_by_host(self) -> None:
|
||||
# Two repos referencing the same alias share one route/credential.
|
||||
idx = ManifestIndex.from_json_obj({
|
||||
"bottles": {"bb": {"git-gate": {"repos": {
|
||||
"r1": _repo(forge="g"),
|
||||
"r2": {
|
||||
"url": "ssh://git@h/x/other.git",
|
||||
"key": {"provider": "static", "path": "/k"},
|
||||
"forge": "g",
|
||||
},
|
||||
}}}},
|
||||
"agents": {"claude": {"bottle": "bb", "prompt": "",
|
||||
"forge-accounts": {"g": GITEA}}},
|
||||
})
|
||||
m = idx.load_for_agent("claude", ())
|
||||
routes = egress_forge_routes(m.forge_associations)
|
||||
self.assertEqual(1, len(routes))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# generated prompt guidance
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestForgeGuidance(unittest.TestCase):
|
||||
def _assoc(self):
|
||||
idx = _index(
|
||||
forge_accounts={"didericis-gitea": GITEA},
|
||||
repo_forge="didericis-gitea",
|
||||
)
|
||||
return idx.load_for_agent("claude", ()).forge_associations
|
||||
|
||||
def test_empty_when_no_associations(self) -> None:
|
||||
self.assertEqual("", render_forge_guidance(()))
|
||||
|
||||
def test_names_account_repo_and_workflow(self) -> None:
|
||||
text = render_forge_guidance(self._assoc())
|
||||
self.assertIn("didericis-gitea", text)
|
||||
self.assertIn("https://gitea.dideric.is/api/v1", text)
|
||||
self.assertIn("bot-bottle", text)
|
||||
# branch-backed PR workflow + AGit prohibition.
|
||||
self.assertIn("refs/heads/", text)
|
||||
self.assertIn("refs/for/*", text)
|
||||
self.assertIn("pull request", text)
|
||||
|
||||
def test_no_token_secret_name_or_value(self) -> None:
|
||||
text = render_forge_guidance(self._assoc())
|
||||
self.assertNotIn("GITEA_CLAUDE_TOKEN", text)
|
||||
|
||||
def test_prompt_file_appends_guidance(self) -> None:
|
||||
from bot_bottle.backend.resolve_common import prepare_agent_state_dir
|
||||
|
||||
idx = _index(
|
||||
author={"name": "n", "email": "e@x.is"},
|
||||
forge_accounts={"didericis-gitea": GITEA},
|
||||
repo_forge="didericis-gitea",
|
||||
)
|
||||
# Give the agent a real prompt body.
|
||||
m = idx.load_for_agent("claude", ())
|
||||
from dataclasses import replace
|
||||
m = replace(m, agent=replace(m.agent, prompt="BASE PROMPT"))
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
with patch.dict(os.environ, {"BOT_BOTTLE_ROOT": td}):
|
||||
_dir, prompt_file = prepare_agent_state_dir("slug1", m)
|
||||
body = Path(prompt_file).read_text()
|
||||
self.assertIn("BASE PROMPT", body)
|
||||
self.assertIn("Forge access", body)
|
||||
self.assertIn("https://gitea.dideric.is/api/v1", body)
|
||||
|
||||
def test_prompt_file_no_guidance_without_forge(self) -> None:
|
||||
from bot_bottle.backend.resolve_common import prepare_agent_state_dir
|
||||
|
||||
m = _index(author={"name": "n", "email": "e@x.is"}).load_for_agent(
|
||||
"claude", ()
|
||||
)
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
with patch.dict(os.environ, {"BOT_BOTTLE_ROOT": td}):
|
||||
_dir, prompt_file = prepare_agent_state_dir("slug2", m)
|
||||
body = Path(prompt_file).read_text()
|
||||
self.assertNotIn("Forge access", body)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user