autojanet/container/entrypoint.py

#!/usr/bin/env python3
"""
AutoJanet Agent Entrypoint

Bootstraps an agent container:
1. Authenticates to OpenBao via AppRole (OPENBAO_ROLE_ID + OPENBAO_SECRET_ID)
2. Fetches all secrets for AGENT_ROLE from OpenBao
3. Writes an opencode-compatible ~/.config/opencode/ environment
4. Runs opencode non-interactively with the task as the prompt

Environment variables (injected by dispatcher Job):
  AGENT_ROLE        — e.g. "coder"
  TASK_ID           — Vikunja task ID
  TASK_TITLE        — Vikunja task title (used as initial prompt)
  OPENBAO_ADDR      — e.g. "http://openbao.openbao.svc.cluster.local:8200"
  OPENBAO_ROLE_ID   — AppRole role_id
  OPENBAO_SECRET_ID — AppRole secret_id
  LITELLM_BASE_URL  — e.g. "https://llm.ctz.fyi"
  VIKUNJA_BASE_URL  — e.g. "https://tasks.ctz.fyi"
  FORGEJO_BASE_URL  — e.g. "https://git.ctz.fyi"
"""

import json
import logging
import os
import subprocess
import sys
import tempfile
from pathlib import Path

import httpx

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
    stream=sys.stdout,
)
log = logging.getLogger("entrypoint")

OPENBAO_ADDR = os.environ["OPENBAO_ADDR"]
OPENBAO_ROLE_ID = os.environ["OPENBAO_ROLE_ID"]
OPENBAO_SECRET_ID = os.environ["OPENBAO_SECRET_ID"]
AGENT_ROLE = os.environ["AGENT_ROLE"]
TASK_ID = os.environ["TASK_ID"]
TASK_TITLE = os.environ.get("TASK_TITLE", f"Task {TASK_ID}")
LITELLM_BASE_URL = os.environ.get("LITELLM_BASE_URL", "https://llm.ctz.fyi")
VIKUNJA_BASE_URL = os.environ.get("VIKUNJA_BASE_URL", "https://tasks.ctz.fyi")
FORGEJO_BASE_URL = os.environ.get("FORGEJO_BASE_URL", "https://git.ctz.fyi")

HOME = Path(os.environ.get("HOME", "/home/agent"))
CONFIG_DIR = HOME / ".config" / "opencode"


def get_openbao_token() -> str:
    resp = httpx.post(
        f"{OPENBAO_ADDR}/v1/auth/approle/login",
        json={"role_id": OPENBAO_ROLE_ID, "secret_id": OPENBAO_SECRET_ID},
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json()["auth"]["client_token"]


def get_secret(bao_token: str, path: str, key: str) -> str:
    resp = httpx.get(
        f"{OPENBAO_ADDR}/v1/secret/data/{path}",
        headers={"X-Vault-Token": bao_token},
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json()["data"]["data"][key]


def fetch_role_secrets(bao_token: str, role: str) -> dict:
    """Fetch all secrets for a role. Returns dict of secret_name -> value."""
    secrets = {}
    secret_names = ["litellm-key", "vikunja-token", "forgejo-token", "argocd-token"]
    for name in secret_names:
        try:
            key = "token" if name != "litellm-key" else "key"
            secrets[name] = get_secret(bao_token, f"autojanet/{role}/{name}", key)
            log.info("Fetched secret: %s", name)
        except Exception as e:
            log.warning("Could not fetch %s: %s", name, e)
    return secrets


def write_opencode_config(secrets: dict, role: str) -> None:
    """Write opencode config with the agent's secrets and MCP server tokens."""
    CONFIG_DIR.mkdir(parents=True, exist_ok=True)

    litellm_key = secrets.get("litellm-key", "")
    vikunja_token = secrets.get("vikunja-token", "")
    forgejo_token = secrets.get("forgejo-token", "")

    config = {
        "model": f"litellm/copilot/claude-sonnet-4.6",
        "providers": {
            "litellm": {
                "apiKey": litellm_key,
                "baseURL": f"{LITELLM_BASE_URL}/v1",
            }
        },
        "mcp": {
            "vikunja": {
                "type": "sse",
                "url": f"{LITELLM_BASE_URL}/mcp/vikunja",
                "headers": {
                    "x-vikunja-token": vikunja_token,
                }
            },
            "forgejo": {
                "type": "sse",
                "url": f"{LITELLM_BASE_URL}/mcp/forgejo",
                "headers": {
                    "x-forgejo-token": forgejo_token,
                }
            },
        }
    }

    config_path = CONFIG_DIR / "config.json"
    config_path.write_text(json.dumps(config, indent=2))
    log.info("Wrote opencode config to %s", config_path)

    # Write AGENTS.md with role-specific instructions
    agent_md_src = Path(f"/app/agents/{role}.agent.md")
    agents_md_dst = CONFIG_DIR / "AGENTS.md"
    if agent_md_src.exists():
        agents_md_dst.write_text(agent_md_src.read_text())
        log.info("Loaded agent instructions from %s", agent_md_src)
    else:
        log.warning("No agent file found at %s", agent_md_src)


def build_prompt(task_id: str, task_title: str) -> str:
    return f"""You are the AutoJanet agent for role: {AGENT_ROLE}

Your current task (Vikunja task #{task_id}):
{task_title}

Instructions:
1. Read the task carefully.
2. Fetch full task details from Vikunja if needed.
3. Complete the task using the tools available to you.
4. Move the task to Done in Vikunja when complete.
5. Open a PR if code was written.
6. Do not ask for confirmation — act autonomously within your constraints.
"""


def run_opencode(prompt: str) -> int:
    """Run opencode non-interactively with the given prompt."""
    cmd = ["opencode", "run", prompt]
    log.info("Running: %s", " ".join(cmd))

    result = subprocess.run(cmd, check=False)
    return result.returncode


def main() -> None:
    log.info("Agent entrypoint: role=%s task=%s", AGENT_ROLE, TASK_ID)

    bao_token = get_openbao_token()
    log.info("OpenBao authenticated")

    secrets = fetch_role_secrets(bao_token, AGENT_ROLE)
    write_opencode_config(secrets, AGENT_ROLE)

    prompt = build_prompt(TASK_ID, TASK_TITLE)
    rc = run_opencode(prompt)

    if rc != 0:
        log.error("opencode exited with code %d", rc)
        sys.exit(rc)

    log.info("Agent completed task %s successfully", TASK_ID)


if __name__ == "__main__":
    main()