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")
IN_REVIEW_BUCKET_ID = os.environ.get("IN_REVIEW_BUCKET_ID", "")

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 = {}
    try:
        resp = httpx.get(
            f"{OPENBAO_ADDR}/v1/secret/data/autojanet/{role}/litellm-key",
            headers={"X-Vault-Token": bao_token},
            timeout=10,
        )
        resp.raise_for_status()
        secrets["litellm-key"] = resp.json()["data"]["data"]["key"]
        log.info("Fetched litellm-key")
    except Exception as e:
        log.warning("Could not fetch litellm-key: %s", e)
    return secrets


def write_opencode_config(secrets: dict, role: str) -> None:
    """Write opencode config and set secrets as env vars for opencode to pick up."""
    CONFIG_DIR.mkdir(parents=True, exist_ok=True)

    litellm_key = secrets.get("litellm-key", "")

    config = {
        "$schema": "https://opencode.ai/config.json",
        "model": "litellm/copilot/claude-sonnet-4.6",
        "provider": {
            "litellm": {
                "npm": "@ai-sdk/openai-compatible",
                "name": "LiteLLM",
                "options": {
                    "baseURL": f"{LITELLM_BASE_URL}/v1",
                    "apiKey": litellm_key,
                },
                "models": {
                    "copilot/claude-sonnet-4.6": {"name": "copilot/claude-sonnet-4.6"},
                },
            }
        },
        "mcp": {
            "vikunja": {
                "type": "remote",
                "url": f"{LITELLM_BASE_URL}/mcp/vikunja",
                "headers": {
                    "Authorization": f"Bearer {litellm_key}",
                },
                "enabled": True,
            },
            "forgejo": {
                "type": "remote",
                "url": f"{LITELLM_BASE_URL}/mcp/forgejo",
                "headers": {
                    "Authorization": f"Bearer {litellm_key}",
                },
                "enabled": True,
            },
        }
    }

    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. Fetch full task details from Vikunja if needed.
2. Complete the task using the tools available to you.
3. If you wrote code, open a PR on Forgejo — do not merge it yourself.
4. Leave a comment on the Vikunja task summarising what you did and linking any PR.
5. Do not mark the task as done — the entrypoint will move it to In Review when you finish.
6. Do not ask for confirmation — act autonomously within your role 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 move_task_to_in_review(vikunja_token: str) -> None:
    """Move the task bucket to In Review after work is complete."""
    if not IN_REVIEW_BUCKET_ID:
        log.warning("IN_REVIEW_BUCKET_ID not set, skipping bucket move")
        return
    try:
        resp = httpx.post(
            f"{VIKUNJA_BASE_URL}/api/v1/tasks/{TASK_ID}",
            headers={"Authorization": f"Bearer {vikunja_token}", "Content-Type": "application/json"},
            json={"bucket_id": int(IN_REVIEW_BUCKET_ID)},
            timeout=10,
        )
        resp.raise_for_status()
        log.info("Moved task %s → In Review (bucket %s)", TASK_ID, IN_REVIEW_BUCKET_ID)
    except Exception as e:
        log.warning("Failed to move task to In Review: %s", e)


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)

    # Fetch vikunja token for bucket moves
    vikunja_token = ""
    try:
        vikunja_token = get_secret(bao_token, f"autojanet/{AGENT_ROLE}/vikunja-token", "token")
        log.info("Fetched vikunja-token")
    except Exception as e:
        log.warning("Could not fetch vikunja-token: %s", e)

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

    # Move to In Review regardless of exit code — work was attempted
    if vikunja_token:
        move_task_to_in_review(vikunja_token)

    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()