codex-mmo/libexec/mmo_guidance.py

#!/usr/bin/env python3
"""Deterministic Codex guidance compiled from one resolved MMO profile."""

from __future__ import annotations

from collections.abc import Mapping
from typing import Any

PROFILE_SKILL_NAME = "mmo-profile-orchestration"
PROFILE_SKILL_RELATIVE_PATH = f"guidance/skills/{PROFILE_SKILL_NAME}/SKILL.md"


def agent_guidance_relative_path(agent_id: str) -> str:
    return f"guidance/agents/{agent_id}.md"


def coordination_capable_agents(resolved: Mapping[str, Any]) -> list[str]:
    return sorted(
        agent_id
        for agent_id, agent in resolved["agents"].items()
        if agent.get("can_spawn") or agent.get("controls")
    )


def _mechanical_limit_lines(resolved: Mapping[str, Any]) -> list[str]:
    coordination = resolved["coordination"]
    orchestration = coordination["orchestration"]
    lines = [f"- Orchestration: `{orchestration}`."]
    if orchestration in {"mcp", "hybrid"}:
        lines.extend(
            [
                "- Maximum simultaneously admitted Agent MCP jobs plus the root: "
                f"{coordination['max_active_agents']}.",
                "- Terminal, suspended, and cold-paused MCP jobs release admission capacity; "
                "sequential delegation is not limited by a lifetime spawn counter.",
                f"- Maximum MCP delegation depth: {coordination['max_depth']}.",
                f"- Maximum active MCP writers: {coordination['max_active_writers']}.",
                "- MCP write-scope conflicts are rejected mechanically.",
                f"- MCP result visibility: {coordination['result_visibility']}.",
            ]
        )
    if orchestration in {"native", "hybrid"}:
        lines.append(
            "- Maximum concurrent native spawned threads, excluding the root: "
            f"{coordination['native_max_concurrent_threads']}."
        )
    lines.append(f"- Contradictions are resolved by `{coordination['contradiction_policy']}`.")
    return lines


def _profile_coordination_text(resolved: Mapping[str, Any]) -> str:
    coordination = resolved["coordination"]
    root = resolved["profile"]["root"]
    orchestration = coordination["orchestration"]
    if orchestration == "mcp":
        backend_text = """Delegation is handled by the `mmo_mesh` Agent MCP supervisor. Use its
`agent_spawn` or `agents_spawn` tools. Each child is an isolated asynchronous
Codex app-server worker with a durable thread, event trace, partial evidence,
bounded lineage, resource admission, write-scope leasing, lifecycle control,
and output-contract validation."""
    elif orchestration == "native":
        backend_text = """Delegation is handled by Codex native subagents. Use the native agent tools
and the generated custom roles. Native threads integrate with `/agent` and have
lower launch overhead, but MMO cannot mechanically enforce per-child write
scopes, output contracts, or every spawn-graph edge. Keep writes disjoint and
validate all material results."""
    else:
        backend_text = """Both delegation paths are available.

- Use Codex native subagents for low-latency, read-heavy, tightly coupled work
  where the `/agent` UI and shared live context are valuable.
- Use the `mmo_mesh` Agent MCP supervisor for durable asynchronous jobs, strict
  role/model pinning, nested bounded delegation, output contracts, explicit
  write scopes, live steering, detach-safe continuation, auditability, or
  low-trust participants.

Do not launch the same assignment through both paths unless independent
redundancy is intentional. Native and MCP results are both evidence that the
caller must reconcile."""
    limits = "\n".join(_mechanical_limit_lines(resolved))
    return f"""## Codex MMO coordination contract

This session is pinned to immutable profile `{resolved["profile"]["id"]}` and logical profile
`{resolved["logical_hash"][:16]}`. The root role is `{root}`.
Orchestration backend: `{orchestration}`.

{backend_text}

Runtime policy:
{limits}

Keep the immediate critical path. Launch independent side work early, then
continue non-overlapping work. Do not wait merely because a child exists. Wait
only when the next required action depends on unfinished output. Treat every
child result as evidence, not authority. Resolve disagreement from source,
commands, tests, specifications, or other reproducible primary evidence; never
decide by model vote.
"""


def agent_instructions_text(resolved: Mapping[str, Any], agent_id: str, *, is_root: bool) -> str:
    agent = resolved["agents"][agent_id]
    model = resolved["models"][agent["model"]]
    route = resolved["routes"][model["route"]]
    role = "root integration authority" if is_root else "delegated participant"
    native_children = [
        child for child in agent["can_spawn"] if "native" in resolved["agents"][child]["backends"]
    ]
    mcp_children = [
        child for child in agent["can_spawn"] if "mcp" in resolved["agents"][child]["backends"]
    ]
    child_lines: list[str] = []
    for child in agent["can_spawn"]:
        child_agent = resolved["agents"][child]
        backends = "/".join(child_agent["backends"])
        child_lines.append(
            f"- `{child}` via {backends}: {child_agent.get('description') or 'profile participant'}"
        )
    children_text = "\n".join(child_lines) or "- none"
    if agent["execution_mode"] == "goal":
        execution_text = f"""Execution mode: durable Codex `goal`. The initial token budget is
{agent["goal_token_budget"]} and the profile ceiling is {agent["max_goal_token_budget"]}.
Silence for {agent["stall_warning_seconds"]} seconds produces an operator warning only; it does
not interrupt the model. A terminal schema turn may use up to
{agent["finalization_grace_seconds"]} seconds after investigation is complete. The host owns
token accounting and lifecycle state. Never estimate elapsed time or emit checkpoint prose merely
to prove liveness. A final assistant message does not finish an active goal. When the objective is
actually achieved and no required work remains, call `update_goal` with `status="complete"` in the
terminal turn, then provide the final result. Use `blocked` only under the tool-defined repeated-
impasse rule; do not use it for ordinary uncertainty, slow work, or a nearly exhausted budget."""
    else:
        execution_text = f"""Execution mode: one durable Codex `turn`, with no profile wall-clock
task deadline. Silence for {agent["stall_warning_seconds"]} seconds produces an operator warning
only. A strict terminal repair may use up to {agent["finalization_grace_seconds"]} seconds. Never
estimate elapsed time or emit checkpoint prose merely to prove liveness."""
    text = f"""# Codex MMO agent: {agent_id}

You are the `{agent_id}` {role} in profile `{resolved["profile"]["id"]}`.

Model binding: `{model["upstream_id"]}` through route `{model["route"]}` ({route["name"]}).
Trust policy: `{agent["trust"]}`.
Verification policy: `{agent["verification"]}`.
Maximum permissions: `{agent["permissions"]}`.
{execution_text}

Permitted child roles and execution paths:
{children_text}

{_profile_coordination_text(resolved)}
"""
    if not is_root:
        text += """
The parent and root retain ownership of integration and the final user-facing
answer. Stay inside the delegated objective. Do not broaden scope or make an
unstated architectural or product decision. Preserve unrelated changes and
return precise evidence, validation, risks, and blockers.
"""
    if agent["can_spawn"]:
        text += """
## Required delegation checkpoint

Use `$mmo-profile-orchestration` before prolonged work. Within at most three
substantive task calls, or before starting a second independent workstream,
identify work retained here and eligible independent work for a direct child.
A substantive call performs repository discovery, shell execution, external
lookup, analysis, or implementation; loading guidance and managing an existing
child do not count.

For a nontrivial task, launch at least one eligible independent branch early.
When two independent branches are eligible and capacity permits, batch-launch
them. If no branch is launched, state the concrete reason before continuing:
the task is atomic, no direct child is independently useful, a required route
is unavailable, a graph or resource limit is exhausted, the user prohibited
delegation, or an authority/tool boundary prevents a safe handoff. Do not use a
generic claim that solo work is easier.
"""
    if native_children:
        text += """
## Native delegation

Use the generated Codex native roles for fast read-heavy parallelism and
closely coupled work. Native agents share the Codex process and workspace; do
not assume MMO write-scope or output-contract enforcement applies to them.
"""
    if mcp_children or agent.get("controls"):
        text += """
## Agent MCP delegation

Use `mmo_mesh` for durable asynchronous jobs, strict model pinning, disjoint
write scopes, contract validation, nested bounded delegation, cancellation, or
low-trust roles. Continue separate work after spawning when this role has spawn
authority. Start with `agent_list`: controls use opaque `agent_run_ref` values,
never guessed job IDs or model-supplied identities. Inspect or trace an
authorized long-running agent when its state matters; silence produces a
warning, not an automatic failure. Use only the actions granted for the target:
steer, answer pending input, change an allowed effort, interrupt the current
turn, pause continued work, continue the same thread (or extend a recoverable
goal within its compiled token ceiling), detach without stopping work, compact,
fork, request terminal serialization, or fully stop while retaining evidence.
Use the revision from the latest inspection for every mutating control.

Use the `progress_revision` values from `agent_status` or a prior
`agents_wait` call as the exact `after_revision` map for every requested job;
this returns compact state on the first
durable change instead of repeatedly injecting unchanged job metadata. Result
previews are opt-in and bounded. Read each material terminal result through
`agent_result`, beginning at cursor 0 and following every returned
`next_cursor` until it is null. Never read MMO supervisor state, job result
files, event logs, stderr files, or sockets directly from the state directory;
use `agent_result`, `agent_inspect`, `agent_trace`, and `agent_trace_record`.
Page a truncated trace summary through its `record_cursor`. When this role
has lineage disposition authority, explicitly accept or reject a successfully
completed result before relying on it and integrate a writable result only
after acceptance and review.
A lost client or app-server transport does not imply lost work. After recovery,
use `agent_list`, `agent_status`, and `agent_inspect` to find retained runs before
spawning any replacement. Continue the same suspended run when its original
objective remains useful. A replacement turn-mode host first settles an
orphaned active turn and starts at most one continuation; a terminal result
that completed during that race remains authoritative.
A provider limit, transport failure, malformed tool call, or failed terminal
turn remains typed on the same run with raw error and partial evidence. Treat
provider reset text without a timezone as provider-local/unspecified; do not
invent a timezone or replace the affected role with an undeclared route.
A detached job retains its live host, thread, and evidence. A cold-paused MCP
job retains its persisted thread and evidence while retiring its host and
releasing execution capacity; continuation starts one replacement host for
that same thread after fresh admission. Native pause remains logical because
native threads share the root host. A suspended job also retains its persisted
thread and evidence. Failed, stopped, or cancelled jobs retain inspectable
evidence but cannot be dispositioned as successful results.
"""
    if agent["can_spawn"]:
        text += """
You remain responsible for consuming and reconciling every material descendant
result before reporting upward.
"""
    elif agent.get("controls"):
        text += """
You have control authority but no spawn authority. Do not create agents. Use
the exact control graph only to unblock, correct, preserve, or conclude work
that another authorized role already admitted.
"""
    else:
        text += "\nYou are a true leaf participant with no spawn or control authority.\n"
    if agent["trust"] == "low":
        text += """
## Low-trust evidence boundary

Perform only bounded, literal, directly verifiable work. Do not infer intent,
architecture, correctness, causality, or recommended action unless the task and
contract explicitly permit it. Report conflicts without choosing a winner. The
parent must independently verify every material claim.
"""
    if agent["verification"] == "root_adjudication":
        text += (
            "\nYour conclusions are adversarial input for root adjudication, not final decisions.\n"
        )
    profile_text = agent.get("instructions_text", "").strip()
    if profile_text:
        text += "\n## Profile-specific role instructions\n\n" + profile_text + "\n"
    return text


def _cell(value: Any) -> str:
    return str(value).replace("|", "\\|").replace("\n", " ").strip()


def profile_skill_text(resolved: Mapping[str, Any]) -> str:
    profile_id = str(resolved["profile"]["id"])
    coordination = resolved["coordination"]
    role_rows: list[str] = []
    for agent_id in sorted(resolved["agents"]):
        agent = resolved["agents"][agent_id]
        backends = "/".join(agent.get("backends", [])) or "root"
        task_kinds = ", ".join(agent.get("allowed_task_kinds", [])) or "root-owned"
        children = ", ".join(agent.get("can_spawn", [])) or "none"
        role_rows.append(
            "| "
            + " | ".join(
                _cell(value)
                for value in (
                    f"`{agent_id}`",
                    backends,
                    task_kinds,
                    agent["permissions"],
                    f"{agent['trust']}/{agent['verification']}",
                    children,
                )
            )
            + " |"
        )
    rows = "\n".join(role_rows)
    control_rows = "\n".join(
        f"- `{agent_id}` -> `{target}`: " + ", ".join(f"`{action}`" for action in grant["actions"])
        for agent_id, agent in sorted(resolved["agents"].items())
        for target, grant in sorted(agent.get("controls", {}).items())
    )
    orchestration = coordination["orchestration"]
    if orchestration == "native":
        path_description = "Codex native agents"
        execution_paths = """- Use the generated roles through Codex native agent tools. They are
  appropriate for fast, read-heavy, tightly coupled work. Keep writes disjoint
  because MMO does not enforce native write scopes or result contracts
  mechanically.
- `mmo_mesh` Agent MCP delegation is not available in this profile."""
    elif orchestration == "mcp":
        path_description = "mmo_mesh"
        execution_paths = """- Use MCP roles through `mmo_mesh`. Tool MCP servers such as IDA or
  Firecrawl provide capabilities; they do not launch agents. `mmo_mesh` owns
  agent lineage, admission, job state, cancellation, contracts, and result
  decisions.
- Codex native agent delegation is not available in this profile."""
    else:
        path_description = "native agents or mmo_mesh"
        execution_paths = """- Use native roles through Codex native agent tools. They are appropriate for
  fast, read-heavy, tightly coupled work. Keep writes disjoint because MMO does
  not enforce native write scopes or result contracts mechanically.
- Use MCP roles through `mmo_mesh`. Tool MCP servers such as IDA or Firecrawl
  provide capabilities; they do not launch agents. `mmo_mesh` owns agent
  lineage, admission, job state, cancellation, contracts, and result decisions.
- Do not send the same assignment through both paths unless independent
  reproduction or adversarial diversity is the explicit objective."""
    has_mcp_lifecycle = orchestration in {"mcp", "hybrid"} and any(
        agent.get("controls")
        or any(
            "mcp" in resolved["agents"][child].get("backends", [])
            for child in agent.get("can_spawn", [])
        )
        for agent in resolved["agents"].values()
    )
    mcp_lifecycle = ""
    if has_mcp_lifecycle:
        mcp_lifecycle = """
## Use the MCP lifecycle

For each MCP task, provide an objective, necessary context and paths, non-goals,
the required deliverable or result contract, and validation evidence. Use
`agents_spawn` for independent batches and `agent_spawn` for one branch.

Remain productive while jobs run. Check status only when useful. Call
`agents_wait` only at a genuine dependency barrier. Pass the latest exact
per-job `progress_revision` map for every requested job as `after_revision` so unchanged work does not bloat
the caller context; the call returns when durable state changes or its bounded
wait expires. Result previews are opt-in and never the complete result. Read
every terminal result with
`agent_result`: begin at cursor 0 and keep calling it with each `next_cursor`
until `next_cursor` is null. Concatenate text pages in cursor order without
overlap; a complete strict structured result may instead arrive once as JSON.
Never bypass this lifecycle by opening MMO job result files, event logs, stderr
files, sockets, or other supervisor state directly. Use `agent_result`,
`agent_inspect`, and `agent_trace`; when a trace summary is truncated, use its
`record_cursor` with `agent_trace_record` and page through every `next_cursor`.
Same-user filesystem access is not an authorization boundary.
Use `agent_list` to discover root, native, and MCP runs plus their opaque refs;
use `agent_status` for a known supervised job, then use `agent_inspect` and
`agent_trace` instead of polling blindly. After any client or transport
recovery, discover and inspect retained work before spawning replacements.
Continue the same suspended run when its objective remains useful. Mutating
controls are compare-and-swap operations: inspect first and pass the returned
revision. Use `agent_steer` to add direction to an active turn without replacing
its existing task. `agent_interrupt` stops only the current turn and an active
goal may continue; `agent_pause` first pauses the goal, interrupts it, retains
partial evidence, and cold-retires a supervised MCP host.
`agent_detach` removes the client while work continues, `agent_continue`
reactivates recoverable work (and may raise its token budget only within the
compiled ceiling), and `agent_stop` pauses, interrupts, retains evidence, and
retires a supervised MCP host or terminates a native run without retiring its
shared root host. Use `agent_respond` for pending requests with the exact
method-specific response shape, `agent_finalize` for strict terminal
serialization, `agent_compact` for thread compaction, `agent_set_effort` within
the role grant. `agent_fork` normally admits an independent MCP job; a native
fork instead inherits its role, cwd, and sandbox in the shared root host, obeys
the native-thread limit, and cannot accept MCP write-scope or attachment overrides.

For a successfully completed job, use `agent_result_accept` or
`agent_result_reject` with a concrete reason when those lineage-authority tools
are exposed to the current role. A detached job keeps its live host. A
cold-paused supervised MCP job keeps its thread, trace, partial evidence, and
artifacts while releasing host capacity; a paused native thread remains in its
shared root host. A suspended job keeps the same durable evidence;
continuation starts or reattaches exactly one host for that thread after fresh
admission. Failed, stopped, and
cancelled jobs cannot be dispositioned; preserve their status and uncertainty.
Integrate an accepted writable patch only after reviewing its scope and tests.
Use `agent_cancel` only for work that is stale, superseded, unsafe, or no longer
worth its cost; cancellation is immediate and distinct from evidence-preserving
finalization.

Every root and supervised MCP worker owns one Unix app-server host and durable
thread; native agents are durable child threads inside their parent root host.
Interactive clients attach to the current root generation of the immutable MMO
session and run. An intentional stock-TUI fresh-context action may create a new
top-level Codex root generation only while the verified attached root client is
idle; it does not create another MMO session or run. Resume by the MMO session
ID or any predecessor root-thread ID attaches to the current generation. This
is host-owned lifecycle: do not simulate it by spawning a replacement root or
starting another MMO session. A replacement turn-mode worker host first settles
an orphaned active turn before starting at most one same-thread continuation;
a terminal result completed during that race is preserved. Goal roles are
bounded by Codex token accounting; turn roles have no task wall clock. Stall
intervals are warning-only and provider/model slowness does not erase work.
Do not tell a model to watch a clock or emit periodic checkpoint prose. Strict
contracts apply to the
explicit terminal serialization turn, followed by at most one same-thread
repair.

Resolve conflicts with primary evidence, not voting or model reputation. A
worker failure, budget suspension, unavailable route, or malformed contract is explicit
uncertainty; it is not permission to silently substitute another route.
"""
        if control_rows:
            mcp_lifecycle += f"""
## Control graph

{control_rows}
"""
    limits = "\n".join(_mechanical_limit_lines(resolved))
    return f"""---
name: {PROFILE_SKILL_NAME}
description: "Coordinate the resolved Codex MMO profile {profile_id}. Use for nontrivial work when the current MMO role can delegate or control durable work through {path_description}, especially for parallel investigation, specialist work, independent verification, or bounded nested delegation."
---

# Orchestrate the profile

Identify the current role from `AGENTS.md`. Spawn only its direct children and
obey the exact profile graph and limits below. Keep ownership of the critical
path and final integration.

## Delegation checkpoint

This checkpoint applies only when the current role's `Direct children` cell is
not `none`. A control-only role must not spawn; it should use the MCP lifecycle
and exact control graph below only at a real dependency, correction, or risk
boundary.

Within at most three substantive task calls, or before entering a second
independent workstream:

1. State the critical-path work retained in the current role.
2. Identify independent work matched to a permitted child.
3. Launch at least one useful branch for a nontrivial decomposable task.
4. Batch-launch two or more independent branches when capacity permits.
5. Continue non-overlapping work immediately after launch.

If delegation is skipped, state one concrete reason before continuing: atomic
task, no independently useful direct child, unavailable route, exhausted graph
or resource limit, explicit user prohibition, or an authority/tool boundary.
Do not treat unfamiliarity with the orchestration tools as a skip reason.

## Choose the execution path

{execution_paths}
{mcp_lifecycle}

## Profile graph

| Role | Backend | Task kinds | Permissions | Trust/verification | Direct children |
|---|---|---|---|---|---|
{rows}

## Mechanical limits

{limits}
"""


def compiled_guidance(resolved: Mapping[str, Any]) -> dict[str, str]:
    root = resolved["profile"]["root"]
    payload = {
        agent_guidance_relative_path(agent_id): agent_instructions_text(
            resolved, agent_id, is_root=agent_id == root
        )
        for agent_id in sorted(resolved["agents"])
    }
    if coordination_capable_agents(resolved):
        payload[PROFILE_SKILL_RELATIVE_PATH] = profile_skill_text(resolved)
    return dict(sorted(payload.items()))