agent_profile.py 7.57 KB
"""Org Agent identity: compiles migrated Hermes personas into Claude Code
subagent definitions and deploys them into a target project (U-004, A-2 assumption).

Source content: `agents/<org_node>/SOUL.md` (+ `AGENTS.md` for validator) in
graphlemon's own repo -- the migrated/adapted personas, see those files' own
migration notes for what changed and why.

Deploy target: `<target_folder_path>/.claude/agents/<org_node>.md`. Confirmed
with user 2026-07-31: 03-C §3.3 ("让整个会话以目标项目 .claude/agents/<org_node>.md
定义的 subagent 身份运行") and 05 §5.1 are the authoritative SDD contract here --
Claude Code's `--agent <name>` discovers subagents relative to the invocation
cwd, so the file has to physically be inside whatever directory `claude` is
run from (the target project), not graphlemon's own repo. Architecture doc
§13.1's "agents/ is a platform-only asset" framing is about the *canonical*
copy (this module's source), not the deployed one -- the two aren't the same
file.

agent_profile (01-M 1.9) is a singleton per org_node, has no target-project
reference -- it tracks graphlemon's own canonical definition, not a per-deploy
record, so deploy_agent_definitions() doesn't touch it; record_agent_profiles()
does, separately.

**H (2026-08-06, Role Package abstraction)**: what used to be two hardcoded
Python dicts (`FRONTMATTER`/`_DESCRIPTIONS`/`_MIGRATED_FROM_PROFILE`) is now
`roles/<role_id>/role.json` -- a declarative manifest, not code. `ORG_NODES`
is no longer a fixed 3-tuple; it's discovered by scanning `roles/` for
directories containing a `role.json`, so a new role becomes recognized by
adding one JSON file, no schema.py/agent_profile.py edit. role.json is JSON,
not YAML, despite Skill files using YAML frontmatter -- pure-stdlib (05
§5.7) rules out PyYAML, and unlike Skill frontmatter (which only Claude Code
itself ever parses, never this codebase), role.json genuinely needs to be
read by this module's own Python code; `json.load` is the zero-risk choice,
no hand-rolled parser. role.json content is deliberately a REFERENCE
manifest, not a content copy: `identity.soul_md_path`/`agents_md_path` still
point at the same `agents/<org_node>/SOUL.md`/`AGENTS.md` files, unmoved --
see docs/done/role-package-abstraction-design.md §2 for why keeping the
existing identity files as the single source of prose content makes the
migration's "deployed output must stay byte-identical" requirement close to
automatic (only the frontmatter's data SOURCE changed, not the body-assembly
logic). Every public function here takes an optional `roles_dir` override
(defaults to the real `roles/` directory) precisely so a test can prove "a
new role needs no code change" against a throwaway fixture directory instead
of mutating this repo's own `roles/`.
"""

import json
import os

_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
_AGENTS_SOURCE_DIR = os.path.join(_REPO_ROOT, "agents")
_ROLES_SOURCE_DIR = os.path.join(_REPO_ROOT, "roles")


def _discover_role_ids(roles_dir=None):
    """role_id == directory name, for every `<roles_dir>/<role_id>/role.json`
    that exists. Sorted for deterministic iteration order (matches the old
    hardcoded tuple's alphabetical `("coder", "planner", "validator")`...
    actually the old tuple was declared as ("planner", "coder", "validator"),
    not alphabetical -- callers that relied on that exact order didn't exist
    (grep confirms every real use is either membership-testing or a `for`
    loop that doesn't care about order), so sorted() here is a deliberate,
    harmless simplification, not a compatibility requirement)."""
    roles_dir = roles_dir or _ROLES_SOURCE_DIR
    if not os.path.isdir(roles_dir):
        return ()
    return tuple(sorted(
        name for name in os.listdir(roles_dir)
        if os.path.isfile(os.path.join(roles_dir, name, "role.json"))
    ))


ORG_NODES = _discover_role_ids()


def _role_json_path(role_id, roles_dir=None):
    return os.path.join(roles_dir or _ROLES_SOURCE_DIR, role_id, "role.json")


def _load_role(role_id, roles_dir=None):
    with open(_role_json_path(role_id, roles_dir), encoding="utf-8") as f:
        return json.load(f)


def _read(path):
    with open(path, encoding="utf-8") as f:
        return f.read()


def _frontmatter_yaml(role):
    # 05 §5.1: skills:/memory:/tools:/model: (and, for validator,
    # disallowedTools: -- the qa/AGENTS.md workspace-permission-contract
    # folded in) are the frontmatter fields U-004's Done criterion checks
    # for. role.json's own field names are snake_case (permissions.tools/
    # disallowed_tools); the deployed frontmatter keeps Claude Code's own
    # camelCase `disallowedTools:` key -- this function is the one place
    # that translates between the two, not role.json itself.
    permissions = role["permissions"]
    lines = ["---", "name: {}".format(role["role_id"]), "description: {}".format(role["description"])]
    lines.append("memory: {}".format(role["context"]["memory"]))
    lines.append("model: {}".format(role["model"]))
    lines.append("skills:")
    lines += ["  - {}".format(s) for s in role["skills"]]
    lines.append("tools:")
    lines += ["  - {}".format(t) for t in permissions["tools"]]
    if permissions.get("disallowed_tools"):
        lines.append("disallowedTools:")
        lines += ["  - {}".format(t) for t in permissions["disallowed_tools"]]
    lines.append("---")
    return "\n".join(lines)


def render_agent_definition(org_node, roles_dir=None):
    """Assemble the .claude/agents/<org_node>.md content: frontmatter + migrated body."""
    if org_node not in _discover_role_ids(roles_dir):
        raise ValueError(org_node)
    role = _load_role(org_node, roles_dir)
    identity = role["identity"]
    body = _read(os.path.join(_REPO_ROOT, identity["soul_md_path"]))
    if identity.get("agents_md_path"):
        body += "\n\n" + _read(os.path.join(_REPO_ROOT, identity["agents_md_path"]))
    return _frontmatter_yaml(role) + "\n\n" + body


def deploy_agent_definitions(target_folder_path, roles_dir=None):
    """Write .claude/agents/<role_id>.md into a target project for every
    discovered role (still exactly {planner, coder, validator} against this
    repo's real roles/ directory, but data-driven now -- see module
    docstring)."""
    out_dir = os.path.join(target_folder_path, ".claude", "agents")
    os.makedirs(out_dir, exist_ok=True)
    for org_node in _discover_role_ids(roles_dir):
        with open(os.path.join(out_dir, org_node + ".md"), "w", encoding="utf-8") as f:
            f.write(render_agent_definition(org_node, roles_dir))


def record_agent_profiles(conn, current_git_commit_sha, roles_dir=None):
    """Upsert the agent_profile (01-M 1.9) singleton rows for graphlemon's own repo."""
    for org_node in _discover_role_ids(roles_dir):
        role = _load_role(org_node, roles_dir)
        identity = role["identity"]
        agents_md_path = identity.get("agents_md_path")
        conn.execute(
            "INSERT INTO agent_profile "
            "(org_node, soul_file_path, agents_file_path, migrated_from_profile, current_git_commit_sha) "
            "VALUES (?, ?, ?, ?, ?) "
            "ON CONFLICT(org_node) DO UPDATE SET "
            "soul_file_path=excluded.soul_file_path, "
            "agents_file_path=excluded.agents_file_path, "
            "migrated_from_profile=excluded.migrated_from_profile, "
            "current_git_commit_sha=excluded.current_git_commit_sha",
            (
                org_node,
                identity["soul_md_path"],
                agents_md_path,
                role["migrated_from_profile"],
                current_git_commit_sha,
            ),
        )
    conn.commit()