agent_comm_protocol

View module diagram

Protocol schemas, builders, and template resolution for agent communication.

This package implements Layers 1-3 of the agent communication protocol:

• Layer 1: :class:TaskManifest — structured facts about a task.
• Layer 2: :func:resolve_instructions — role-specific work instructions.
• Layer 3: :class:WorkflowPolicies — composable workflow configuration.

All types and functions are framework-agnostic. Framework adapters (in task_runner/implementations/) consume these outputs.

DependencyInfo dataclass

Description of a dependency task for manifest purposes.

Attributes:

Name	Type	Description
description	Optional[str]	Human-readable description of the dependency task, or None if no description is available.

Source code in src/agentrelay/agent_comm_protocol/manifest.py

@dataclass(frozen=True)
class DependencyInfo:
    """Description of a dependency task for manifest purposes.

    Attributes:
        description: Human-readable description of the dependency task,
            or ``None`` if no description is available.
    """

    description: Optional[str]

TaskManifest dataclass

Frozen Layer-1 manifest: pure facts about a task.

Attributes:

Name	Type	Description
schema_version	str	Schema version string.
task_id	str	Unique task identifier.
role	AgentRole	Agent role for this task.
description	Optional[str]	Human-readable task description, or None.
tagged_paths	tuple[TaggedPath, ...]	Category-tagged file paths for this task.
branch_name	str	Feature branch for this task's work.
integration_branch	str	Branch this task's PR targets.
attempt_num	int	Current attempt number (0-indexed).
graph_name	Optional[str]	Name of the containing graph, or None.
dependencies	dict[str, DependencyInfo]	Mapping of dependency task IDs to :class:DependencyInfo.
tools	tuple[str, ...]	Declared tool names from the graph YAML.

Source code in src/agentrelay/agent_comm_protocol/manifest.py

@dataclass(frozen=True)
class TaskManifest:
    """Frozen Layer-1 manifest: pure facts about a task.

    Attributes:
        schema_version: Schema version string.
        task_id: Unique task identifier.
        role: Agent role for this task.
        description: Human-readable task description, or ``None``.
        tagged_paths: Category-tagged file paths for this task.
        branch_name: Feature branch for this task's work.
        integration_branch: Branch this task's PR targets.
        attempt_num: Current attempt number (0-indexed).
        graph_name: Name of the containing graph, or ``None``.
        dependencies: Mapping of dependency task IDs to :class:`DependencyInfo`.
        tools: Declared tool names from the graph YAML.
    """

    schema_version: str
    task_id: str
    role: AgentRole
    description: Optional[str]
    tagged_paths: tuple[TaggedPath, ...]
    branch_name: str
    integration_branch: str
    attempt_num: int
    graph_name: Optional[str]
    dependencies: dict[str, DependencyInfo]
    input_files: tuple[InputFileInfo, ...] = ()
    tools: tuple[str, ...] = ()

WorkflowPolicies dataclass

Frozen Layer-3 policies: composable workflow configuration.

Each optional field is independently None (behavior absent) or present (behavior active). Framework adapters read these policies and translate them into framework-specific actions.

Attributes:

Name	Type	Description
schema_version	str	Schema version string.
commit_policy	Optional[_CommitPolicy]	Commit and push policy, or None.
pr_policy	Optional[_PrPolicy]	PR creation policy, or None.
completion_gate	Optional[_CompletionGatePolicy]	Gate loop policy, or None.
review	Optional[_ReviewPolicy]	Self-review policy, or None.
adr	Optional[_AdrPolicy]	ADR writing policy, or None.
verification	Optional[_VerificationPolicy]	Verification command policy, or None.

Source code in src/agentrelay/agent_comm_protocol/policies.py

@dataclass(frozen=True)
class WorkflowPolicies:
    """Frozen Layer-3 policies: composable workflow configuration.

    Each optional field is independently ``None`` (behavior absent) or
    present (behavior active).  Framework adapters read these policies and
    translate them into framework-specific actions.

    Attributes:
        schema_version: Schema version string.
        commit_policy: Commit and push policy, or ``None``.
        pr_policy: PR creation policy, or ``None``.
        completion_gate: Gate loop policy, or ``None``.
        review: Self-review policy, or ``None``.
        adr: ADR writing policy, or ``None``.
        verification: Verification command policy, or ``None``.
    """

    schema_version: str
    commit_policy: Optional[_CommitPolicy]
    pr_policy: Optional[_PrPolicy]
    completion_gate: Optional[_CompletionGatePolicy]
    review: Optional[_ReviewPolicy]
    adr: Optional[_AdrPolicy]
    verification: Optional[_VerificationPolicy]

build_manifest(task, branch_name, integration_branch, graph_name, attempt_num, dependency_descriptions, tools=(), input_files=())

Build a :class:TaskManifest from task spec and contextual data.

Parameters:

Name	Type	Description	Default
task	Task	Frozen task specification.	required
branch_name	str	Git branch for this task's work.	required
integration_branch	str	Branch this task's PR targets.	required
graph_name	Optional[str]	Name of the containing task graph.	required
attempt_num	int	Current execution attempt (0-indexed).	required
dependency_descriptions	dict[str, Optional[str]]	Map of dependency task ID to description (None if the dependency has no description).	required
tools	tuple[str, ...]	Declared tool names from the graph YAML.	()

Returns:

Type	Description
TaskManifest	Frozen manifest with all task facts.

Source code in src/agentrelay/agent_comm_protocol/manifest.py

def build_manifest(
    task: Task,
    branch_name: str,
    integration_branch: str,
    graph_name: Optional[str],
    attempt_num: int,
    dependency_descriptions: dict[str, Optional[str]],
    tools: tuple[str, ...] = (),
    input_files: tuple[InputFileInfo, ...] = (),
) -> TaskManifest:
    """Build a :class:`TaskManifest` from task spec and contextual data.

    Args:
        task: Frozen task specification.
        branch_name: Git branch for this task's work.
        integration_branch: Branch this task's PR targets.
        graph_name: Name of the containing task graph.
        attempt_num: Current execution attempt (0-indexed).
        dependency_descriptions: Map of dependency task ID to description
            (``None`` if the dependency has no description).
        tools: Declared tool names from the graph YAML.

    Returns:
        Frozen manifest with all task facts.
    """
    return TaskManifest(
        schema_version=MANIFEST_SCHEMA_VERSION,
        task_id=task.id,
        role=task.role,
        description=task.description,
        tagged_paths=task.tagged_paths,
        branch_name=branch_name,
        integration_branch=integration_branch,
        attempt_num=attempt_num,
        graph_name=graph_name,
        dependencies={
            dep_id: DependencyInfo(description=desc)
            for dep_id, desc in dependency_descriptions.items()
        },
        input_files=input_files,
        tools=tools,
    )

manifest_to_dict(manifest)

Serialize a :class:TaskManifest to a JSON-compatible dict.

The dict structure matches the Layer-1 schema defined in docs/AGENT_COMM_PROTOCOL.md.

Parameters:

Name	Type	Description	Default
manifest	TaskManifest	Frozen manifest to serialize.	required

Returns:

Type	Description
dict[str, Any]	Nested dict with schema_version, task, paths,
dict[str, Any]	workspace, execution, and dependencies sections.

Source code in src/agentrelay/agent_comm_protocol/manifest.py

def manifest_to_dict(manifest: TaskManifest) -> dict[str, Any]:
    """Serialize a :class:`TaskManifest` to a JSON-compatible dict.

    The dict structure matches the Layer-1 schema defined in
    ``docs/AGENT_COMM_PROTOCOL.md``.

    Args:
        manifest: Frozen manifest to serialize.

    Returns:
        Nested dict with ``schema_version``, ``task``, ``paths``,
        ``workspace``, ``execution``, and ``dependencies`` sections.
    """
    return {
        "schema_version": manifest.schema_version,
        "task": {
            "id": manifest.task_id,
            "role": manifest.role.value,
            "description": manifest.description,
        },
        "paths": [
            {"path": str(tp.path), "category": tp.category}
            for tp in manifest.tagged_paths
        ],
        "workspace": {
            "branch_name": manifest.branch_name,
            "integration_branch": manifest.integration_branch,
        },
        "execution": {
            "attempt_num": manifest.attempt_num,
            "graph_name": manifest.graph_name,
        },
        "dependencies": {
            dep_id: {"description": info.description}
            for dep_id, info in manifest.dependencies.items()
        },
        "input_files": [
            {
                "path": str(f.path),
                "category": f.category,
                "source_task": f.source_task,
            }
            for f in manifest.input_files
        ],
        "tools": list(manifest.tools),
    }

build_policies(task, integration_branch, default_max_gate_attempts=5)

Build :class:WorkflowPolicies from a task spec and contextual data.

Derives policies from the task's configuration:

• commit_policy: always present (commit_and_push).
• pr_policy: always present; base_branch from integration_branch.
• completion_gate: present if task.completion_gate is set.
• review: present if task.review is set.
• adr: present if task.primary_agent.adr_verbosity is not NONE.
• verification: present for TEST_WRITER and TEST_REVIEWER roles (default pytest --collect-only).

Parameters:

Name	Type	Description	Default
task	Task	Frozen task specification.	required
integration_branch	str	Branch for PR base.	required
default_max_gate_attempts	int	Fallback when task.max_gate_attempts is None but task.completion_gate is set.	5

Returns:

Type	Description
WorkflowPolicies	Composable workflow configuration.

Source code in src/agentrelay/agent_comm_protocol/policies.py

def build_policies(
    task: Task,
    integration_branch: str,
    default_max_gate_attempts: int = 5,
) -> WorkflowPolicies:
    """Build :class:`WorkflowPolicies` from a task spec and contextual data.

    Derives policies from the task's configuration:

    - ``commit_policy``: always present (``commit_and_push``).
    - ``pr_policy``: always present; ``base_branch`` from *integration_branch*.
    - ``completion_gate``: present if ``task.completion_gate`` is set.
    - ``review``: present if ``task.review`` is set.
    - ``adr``: present if ``task.primary_agent.adr_verbosity`` is not ``NONE``.
    - ``verification``: present for ``TEST_WRITER`` and ``TEST_REVIEWER`` roles
      (default ``pytest --collect-only``).

    Args:
        task: Frozen task specification.
        integration_branch: Branch for PR base.
        default_max_gate_attempts: Fallback when ``task.max_gate_attempts``
            is ``None`` but ``task.completion_gate`` is set.

    Returns:
        Composable workflow configuration.
    """
    commit_policy = _CommitPolicy(action=_WorkflowAction.COMMIT_AND_PUSH)

    pr_policy = _PrPolicy(
        action=_WorkflowAction.CREATE_PR,
        base_branch=integration_branch,
        title_template="{task_id}",
        body_sections=(_PrBodySection.SUMMARY, _PrBodySection.FILES_CHANGED),
    )

    completion_gate: Optional[_CompletionGatePolicy] = None
    if task.completion_gate is not None:
        completion_gate = _CompletionGatePolicy(
            command=task.completion_gate,
            max_attempts=task.max_gate_attempts or default_max_gate_attempts,
            output_file="gate_last_output.txt",
        )

    review: Optional[_ReviewPolicy] = None
    if task.review is not None:
        review = _ReviewPolicy(
            model=task.review.agent.model,
            review_on_attempt=task.review.review_on_attempt,
        )

    adr: Optional[_AdrPolicy] = None
    if task.primary_agent.adr_verbosity != AdrVerbosity.NONE:
        adr = _AdrPolicy(verbosity=task.primary_agent.adr_verbosity)

    verification: Optional[_VerificationPolicy] = None
    if task.role in (AgentRole.TEST_WRITER, AgentRole.TEST_REVIEWER):
        verification = _VerificationPolicy(commands=("pytest --collect-only",))

    return WorkflowPolicies(
        schema_version=POLICIES_SCHEMA_VERSION,
        commit_policy=commit_policy,
        pr_policy=pr_policy,
        completion_gate=completion_gate,
        review=review,
        adr=adr,
        verification=verification,
    )

policies_to_dict(policies)

Serialize :class:WorkflowPolicies to a JSON-compatible dict.

The dict structure matches the Layer-3 schema defined in docs/AGENT_COMM_PROTOCOL.md. None-valued policies become JSON null.

Parameters:

Name	Type	Description	Default
policies	WorkflowPolicies	Frozen policies to serialize.	required

Returns:

Type	Description
dict[str, Any]	Dict with schema_version and one key per policy type.

Source code in src/agentrelay/agent_comm_protocol/policies.py

def policies_to_dict(policies: WorkflowPolicies) -> dict[str, Any]:
    """Serialize :class:`WorkflowPolicies` to a JSON-compatible dict.

    The dict structure matches the Layer-3 schema defined in
    ``docs/AGENT_COMM_PROTOCOL.md``.  ``None``-valued policies become
    JSON ``null``.

    Args:
        policies: Frozen policies to serialize.

    Returns:
        Dict with ``schema_version`` and one key per policy type.
    """
    return {
        "schema_version": policies.schema_version,
        "commit_policy": (
            _policy_to_dict(policies.commit_policy)
            if policies.commit_policy is not None
            else None
        ),
        "pr_policy": (
            _policy_to_dict(policies.pr_policy)
            if policies.pr_policy is not None
            else None
        ),
        "completion_gate": (
            _policy_to_dict(policies.completion_gate)
            if policies.completion_gate is not None
            else None
        ),
        "review": (
            _policy_to_dict(policies.review) if policies.review is not None else None
        ),
        "adr": (_policy_to_dict(policies.adr) if policies.adr is not None else None),
        "verification": (
            _policy_to_dict(policies.verification)
            if policies.verification is not None
            else None
        ),
    }

resolve_instructions(role, manifest, adapter_name=None, adr_verbosity=AdrVerbosity.NONE, sandbox_type=None, worktree_path=None, graph_yaml_path=None, signals_base_path=None)

Resolve work instructions by loading and parameterizing a role template.

The assembled document is structured as a work order:

1. Role — who the agent is and what it's doing.
2. Working Directory — where the agent must work (if worktree_path is provided).
3. Tools — environment tools available (if any).
4. What to Do — role-specific steps from the template, or the task description for GENERIC roles.
5. Graph Awareness — graph YAML location, signal directory formula, and guidance on reading upstream artifacts and writing summaries (if graph_yaml_path is provided and manifest.graph_name is set).
6. Previous Attempts — archived artifacts from prior retry attempts (if manifest.attempt_num > 0 and signals_base_path is provided).
7. Architecture Decision Record — ADR writing instructions (if adr_verbosity is not NONE).
8. Isolation Boundary — what the agent can/cannot access and what exists beyond its boundary (if sandbox_type is OCI).
9. Submitting Your Work — how to commit, create a PR, and signal the orchestrator.
10. Task Details — the task author's description (non-generic only, when present).

Template variables ($var syntax via :class:string.Template):

• $description — task description
• $paths_by_category — paths grouped by category as a bullet list, or "(none specified)" when empty
• $task_id — task identifier

Parameters:

Name	Type	Description	Default
role	AgentRole	Agent role determining which template to load.	required
manifest	TaskManifest	Task manifest providing substitution values.	required
adapter_name	Optional[str]	Optional adapter name for adapter-specific override.	None
adr_verbosity	AdrVerbosity	ADR detail level. NONE (default) omits the section.	NONE
sandbox_type	Optional[SandboxType]	Sandbox type. When OCI, an isolation boundary section is included describing what the agent can and cannot access.	None
worktree_path	Optional[Path]	Absolute path to the git worktree for this task. When provided, a Working Directory section is included instructing the agent to stay within the worktree.	None
graph_yaml_path	Optional[Path]	Absolute path to the copied graph YAML at .workflow/<graph>/graph.yaml. When provided (and manifest.graph_name is set), a Graph Awareness section is included.	None
signals_base_path	Optional[Path]	Absolute path to .workflow/<graph>/signals/. Used in the Graph Awareness section so agents know where peer signal directories live.	None

Returns:

Type	Description
str	Resolved markdown instruction text.

Raises:

Type	Description
FileNotFoundError	If no template found for a non-GENERIC role.
ValueError	If GENERIC role has no description.

Source code in src/agentrelay/agent_comm_protocol/templates.py

def resolve_instructions(
    role: AgentRole,
    manifest: TaskManifest,
    adapter_name: Optional[str] = None,
    adr_verbosity: AdrVerbosity = AdrVerbosity.NONE,
    sandbox_type: Optional[SandboxType] = None,
    worktree_path: Optional[Path] = None,
    graph_yaml_path: Optional[Path] = None,
    signals_base_path: Optional[Path] = None,
) -> str:
    """Resolve work instructions by loading and parameterizing a role template.

    The assembled document is structured as a work order:

    1. **Role** — who the agent is and what it's doing.
    2. **Working Directory** — where the agent must work (if
       ``worktree_path`` is provided).
    3. **Tools** — environment tools available (if any).
    4. **What to Do** — role-specific steps from the template, or the
       task description for GENERIC roles.
    5. **Graph Awareness** — graph YAML location, signal directory formula,
       and guidance on reading upstream artifacts and writing summaries
       (if ``graph_yaml_path`` is provided and ``manifest.graph_name``
       is set).
    6. **Previous Attempts** — archived artifacts from prior retry
       attempts (if ``manifest.attempt_num > 0`` and
       ``signals_base_path`` is provided).
    7. **Architecture Decision Record** — ADR writing instructions (if
       ``adr_verbosity`` is not ``NONE``).
    8. **Isolation Boundary** — what the agent can/cannot access and
       what exists beyond its boundary (if ``sandbox_type`` is ``OCI``).
    9. **Submitting Your Work** — how to commit, create a PR, and signal
       the orchestrator.
    10. **Task Details** — the task author's description (non-generic only,
        when present).

    Template variables (``$var`` syntax via :class:`string.Template`):

    - ``$description`` — task description
    - ``$paths_by_category`` — paths grouped by category as a bullet list,
      or ``"(none specified)"`` when empty
    - ``$task_id`` — task identifier

    Args:
        role: Agent role determining which template to load.
        manifest: Task manifest providing substitution values.
        adapter_name: Optional adapter name for adapter-specific override.
        adr_verbosity: ADR detail level. ``NONE`` (default) omits the section.
        sandbox_type: Sandbox type. When ``OCI``, an isolation boundary
            section is included describing what the agent can and cannot
            access.
        worktree_path: Absolute path to the git worktree for this task.
            When provided, a Working Directory section is included
            instructing the agent to stay within the worktree.
        graph_yaml_path: Absolute path to the copied graph YAML at
            ``.workflow/<graph>/graph.yaml``.  When provided (and
            ``manifest.graph_name`` is set), a Graph Awareness section
            is included.
        signals_base_path: Absolute path to ``.workflow/<graph>/signals/``.
            Used in the Graph Awareness section so agents know where
            peer signal directories live.

    Returns:
        Resolved markdown instruction text.

    Raises:
        FileNotFoundError: If no template found for a non-GENERIC role.
        ValueError: If GENERIC role has no description.
    """
    parts = [
        f"# Instructions for Task {manifest.task_id}",
        (
            f"## Role\n\n{_ROLE_SENTENCES[role]} "
            "Follow the instructions below to complete the task."
        ),
    ]

    # Working Directory (conditional, early context).
    wd_text = _working_directory_section(worktree_path)
    if wd_text:
        parts.append(wd_text)

    # Tools (flat H2, only if declared).
    tools_text = tool_guidance(manifest.tools)
    if tools_text:
        parts.append("## Tools\n\n" + tools_text.strip())

    # What to Do.
    if role == AgentRole.GENERIC:
        if manifest.description is None:
            raise ValueError(
                f"GENERIC role requires a task description, but task "
                f"'{manifest.task_id}' has description=None."
            )
        parts.append(
            "## What to Do\n\n" + manifest.description + "\n\n" + _concerns_note()
        )
    else:
        template_text = _load_template(role.value, adapter_name)
        substitutions = {
            "description": manifest.description or "",
            "paths_by_category": _format_paths_by_category(manifest.tagged_paths),
            "task_id": manifest.task_id,
            "concerns_note": _concerns_note(),
        }
        resolved = Template(template_text).substitute(substitutions)
        parts.append("## What to Do\n\n" + resolved.strip())

    # Input Files (conditional, only when inputs_from resolved files).
    input_files_text = _input_files_section(manifest)
    if input_files_text:
        parts.append(input_files_text)

    # Graph Awareness (conditional, cross-cutting).
    graph_text = _graph_awareness_section(manifest, graph_yaml_path, signals_base_path)
    if graph_text:
        parts.append(graph_text)

    # Previous Attempts (conditional, cross-cutting).
    attempts_text = _previous_attempts_section(manifest, signals_base_path)
    if attempts_text:
        parts.append(attempts_text)

    # Architecture Decision Record (conditional, cross-cutting).
    adr_text = _adr_section(adr_verbosity, manifest.task_id)
    if adr_text:
        parts.append(adr_text)

    # Isolation Boundary (conditional, cross-cutting).
    isolation_text = _isolation_section(sandbox_type)
    if isolation_text:
        parts.append(isolation_text)

    parts.append(_submission_section(manifest))

    # Task Details (non-generic, when description exists).
    if role != AgentRole.GENERIC and manifest.description:
        parts.append("## Task Details\n\n" + manifest.description)

    return "\n\n".join(parts) + "\n"