runtime.py

from __future__ import annotations

import hashlib
import json
import os
import shutil
import signal
import subprocess
import sys
import tempfile
import time
import uuid
from datetime import UTC, datetime
from pathlib import Path
from typing import Any


STATE_ENV_VAR = "CLAUDE_CODE_MCP_STATE_DIR"
CODEX_PLUGIN_DATA_ENV_VAR = "CODEX_PLUGIN_DATA"
DEFAULT_TIMEOUT_SECONDS = 900
DEFAULT_MAX_JOBS = 8
DEFAULT_MODEL = "claude-fable-5"
DEFAULT_EFFORT = "high"
DEFAULT_WRITE = True
CONTEXT_WARNING_RATIO = 0.75
COST_WARNING_USD = 0.5
LAST_SESSION_BY_WORKSPACE: dict[str, str] = {}
SENSITIVE_KEY_PARTS = (
    "account",
    "api_key",
    "auth",
    "credential",
    "email",
    "id_token",
    "key",
    "login",
    "name",
    "org",
    "organization",
    "secret",
    "session",
    "subscription",
    "token",
    "user",
    "workspace",
)
READ_ONLY_DISALLOWED_TOOLS = "Edit,MultiEdit,Write,NotebookEdit"


class ClaudeRuntimeError(ValueError):
    """Raised when a user-provided runtime request is invalid."""


def now_iso() -> str:
    return datetime.now(UTC).isoformat().replace("+00:00", "Z")


def default_workspace_path() -> Path:
    cwd = Path.cwd().resolve()
    git_root = run_command(["git", "rev-parse", "--show-toplevel"], cwd=cwd)
    if git_root["returncode"] == 0 and git_root["stdout"].strip():
        return Path(git_root["stdout"].strip()).resolve()
    return cwd


def resolve_workspace_path(workspace_path: str | None = None) -> Path:
    workspace = Path(workspace_path).expanduser() if workspace_path else default_workspace_path()
    if not workspace.is_absolute():
        raise ClaudeRuntimeError("workspace_path must be an absolute path.")
    if not workspace.exists():
        raise ClaudeRuntimeError(f"workspace_path does not exist: {workspace}")
    if not workspace.is_dir():
        raise ClaudeRuntimeError(f"workspace_path is not a directory: {workspace}")
    return workspace.resolve()


def state_root_dir() -> Path:
    if os.environ.get(STATE_ENV_VAR):
        return Path(os.environ[STATE_ENV_VAR]).expanduser().resolve()
    if os.environ.get(CODEX_PLUGIN_DATA_ENV_VAR):
        return Path(os.environ[CODEX_PLUGIN_DATA_ENV_VAR]).expanduser().resolve() / "claude-code-state"
    return Path(tempfile.gettempdir()) / "claude-code-codex"


def workspace_state_dir(workspace_path: str | Path) -> Path:
    workspace = Path(workspace_path).resolve()
    slug = "".join(character if character.isalnum() or character in "._-" else "-" for character in workspace.name)
    slug = slug.strip("-") or "workspace"
    digest = hashlib.sha256(str(workspace).encode("utf-8")).hexdigest()[:16]
    return state_root_dir() / f"{slug}-{digest}"


def jobs_dir(workspace_path: str | Path) -> Path:
    directory = workspace_state_dir(workspace_path) / "jobs"
    directory.mkdir(parents=True, exist_ok=True)
    return directory


def job_file(workspace_path: str | Path, job_id: str) -> Path:
    return jobs_dir(workspace_path) / f"{job_id}.json"


def generate_job_id() -> str:
    return f"claude-{int(time.time() * 1000):x}-{uuid.uuid4().hex[:8]}"


def write_job(job: dict[str, Any]) -> dict[str, Any]:
    timestamp = now_iso()
    next_job = {**job, "updated_at": timestamp}
    if "created_at" not in next_job:
        next_job["created_at"] = timestamp
    path = job_file(next_job["workspace_path"], next_job["job_id"])
    path.write_text(f"{json.dumps(next_job, indent=2, sort_keys=True)}\n", encoding="utf-8")
    return next_job


def read_job(workspace_path: str | Path, job_id: str) -> dict[str, Any]:
    path = job_file(workspace_path, job_id)
    if not path.is_file():
        raise ClaudeRuntimeError(f"No Claude Code job found for {job_id}.")
    return json.loads(path.read_text(encoding="utf-8"))


def read_all_jobs(workspace_path: str | Path) -> list[dict[str, Any]]:
    records: list[dict[str, Any]] = []
    for path in sorted(jobs_dir(workspace_path).glob("*.json")):
        try:
            records.append(json.loads(path.read_text(encoding="utf-8")))
        except (OSError, json.JSONDecodeError):
            continue
    return sorted(records, key=lambda item: str(item.get("updated_at", "")), reverse=True)


def latest_finished_job(workspace_path: str | Path) -> dict[str, Any] | None:
    for job in read_all_jobs(workspace_path):
        if job.get("status") in {"completed", "failed", "cancelled"}:
            return job
    return None


def workspace_session_key(workspace: Path) -> str:
    return str(workspace.resolve())


def get_last_session_id(workspace: Path) -> str | None:
    return LAST_SESSION_BY_WORKSPACE.get(workspace_session_key(workspace))


def set_last_session_id(workspace: Path, session_id: str | None) -> None:
    if session_id:
        LAST_SESSION_BY_WORKSPACE[workspace_session_key(workspace)] = session_id


def remember_job_session(workspace: Path, job: dict[str, Any]) -> None:
    if job.get("status") != "completed":
        return
    session = job.get("session")
    if not isinstance(session, dict):
        return
    session_id = session.get("session_id")
    if isinstance(session_id, str) and session_id.strip():
        set_last_session_id(workspace, session_id.strip())


def prompt_requests_new_conversation(prompt: str) -> bool:
    lowered_prompt = prompt.lower()
    new_conversation_terms = (
        "新开",
        "新对话",
        "新的对话",
        "不要继续",
        "别继续",
        "从头开始",
        "重新开始",
        "fresh",
        "new conversation",
        "new session",
        "start over",
        "do not resume",
        "don't resume",
    )
    return any(term in lowered_prompt for term in new_conversation_terms)


def extract_session_id(stdout_json: Any) -> str | None:
    if isinstance(stdout_json, dict):
        session_id = stdout_json.get("session_id")
        if isinstance(session_id, str) and session_id.strip():
            return session_id.strip()
    return None


def conversation_warning(stdout_json: Any) -> dict[str, Any] | None:
    if not isinstance(stdout_json, dict):
        return None

    reasons: list[str] = []
    usage = stdout_json.get("usage") if isinstance(stdout_json.get("usage"), dict) else {}
    model_usage = stdout_json.get("modelUsage") if isinstance(stdout_json.get("modelUsage"), dict) else {}

    total_tokens = 0
    for key in ("input_tokens", "output_tokens", "cache_read_input_tokens", "cache_creation_input_tokens"):
        try:
            total_tokens += int(usage.get(key) or 0)
        except (TypeError, ValueError):
            continue

    context_window = None
    for model_data in model_usage.values():
        if not isinstance(model_data, dict):
            continue
        try:
            candidate = int(model_data.get("contextWindow") or 0)
        except (TypeError, ValueError):
            continue
        if candidate > 0:
            context_window = max(context_window or 0, candidate)

    context_ratio = None
    if context_window and total_tokens:
        context_ratio = total_tokens / context_window
        if context_ratio >= CONTEXT_WARNING_RATIO:
            reasons.append(
                f"Claude context used about {context_ratio:.0%} of {context_window} tokens."
            )

    try:
        total_cost_usd = float(stdout_json.get("total_cost_usd") or 0)
    except (TypeError, ValueError):
        total_cost_usd = 0.0
    if total_cost_usd >= COST_WARNING_USD:
        reasons.append(f"Claude reported total_cost_usd={total_cost_usd:.4f}.")

    if not reasons:
        return None
    return {
        "recommend_new_conversation": True,
        "reasons": reasons,
        "start_new_hint": "下次调用时设置 new_conversation=true，或直接说“新开一个 Claude 对话”。",
        "total_tokens": total_tokens,
        "context_window": context_window,
        "context_ratio": context_ratio,
        "total_cost_usd": total_cost_usd,
    }


def sanitize_auth_payload(value: Any) -> Any:
    if isinstance(value, dict):
        sanitized: dict[str, Any] = {}
        for key, item in value.items():
            lowered_key = str(key).lower()
            if any(part in lowered_key for part in SENSITIVE_KEY_PARTS):
                continue
            sanitized[key] = sanitize_auth_payload(item)
        return sanitized
    if isinstance(value, list):
        return [sanitize_auth_payload(item) for item in value]
    return value


def parse_json_or_text(text: str) -> Any:
    normalized = text.strip()
    if not normalized:
        return None
    try:
        return json.loads(normalized)
    except json.JSONDecodeError:
        return normalized


def run_command(command: list[str], cwd: Path | None = None, timeout_seconds: float = 30) -> dict[str, Any]:
    try:
        completed = subprocess.run(
            command,
            cwd=str(cwd) if cwd else None,
            text=True,
            capture_output=True,
            timeout=timeout_seconds,
            check=False,
        )
    except FileNotFoundError:
        return {
            "returncode": None,
            "stdout": "",
            "stderr": f"{command[0]} was not found.",
            "error": "not_found",
        }
    except subprocess.TimeoutExpired as error:
        return {
            "returncode": None,
            "stdout": error.stdout or "",
            "stderr": error.stderr or "",
            "error": "timeout",
        }
    return {
        "returncode": completed.returncode,
        "stdout": completed.stdout,
        "stderr": completed.stderr,
        "error": None,
    }


def claude_setup(workspace_path: str | None = None) -> dict[str, Any]:
    workspace = resolve_workspace_path(workspace_path)
    claude_path = shutil.which("claude")
    if not claude_path:
        return {
            "ready": False,
            "claude": {
                "available": False,
                "path": None,
                "version": None,
            },
            "auth": {
                "checked": False,
                "logged_in": False,
                "status": None,
            },
            "next_steps": ["Install Claude Code and make sure `claude` is on PATH."],
        }

    version_result = run_command([claude_path, "--version"], cwd=workspace)
    auth_result = run_command([claude_path, "auth", "status", "--json"], cwd=workspace)
    auth_payload = parse_json_or_text(auth_result["stdout"])
    sanitized_auth_payload = sanitize_auth_payload(auth_payload)
    logged_in = auth_result["returncode"] == 0
    next_steps = [] if logged_in else ["Run `claude auth login` before using Claude Code from Codex."]

    return {
        "ready": version_result["returncode"] == 0 and logged_in,
        "claude": {
            "available": version_result["returncode"] == 0,
            "path": claude_path,
            "version": version_result["stdout"].strip() or None,
            "stderr": version_result["stderr"].strip() or None,
        },
        "auth": {
            "checked": True,
            "logged_in": logged_in,
            "status": sanitized_auth_payload,
            "stderr": auth_result["stderr"].strip() or None,
        },
        "next_steps": next_steps,
    }


def normalize_timeout(timeout_seconds: int | float | None) -> float:
    if timeout_seconds is None:
        return float(DEFAULT_TIMEOUT_SECONDS)
    timeout = float(timeout_seconds)
    if timeout <= 0:
        raise ClaudeRuntimeError("timeout_seconds must be greater than 0.")
    return timeout


def validate_task_prompt(prompt: str) -> str:
    normalized = str(prompt or "").strip()
    if not normalized:
        raise ClaudeRuntimeError("prompt must be a non-empty string.")
    return normalized


def build_claude_command(
    prompt: str,
    *,
    write: bool,
    model: str | None = None,
    effort: str | None = None,
    max_budget_usd: int | float | None = None,
    resume_session_id: str | None = None,
) -> list[str]:
    command = [
        "claude",
        "--print",
        "--output-format",
        "json",
        "--permission-mode",
        "acceptEdits" if write else "plan",
    ]
    if not write:
        command.extend(["--disallowedTools", READ_ONLY_DISALLOWED_TOOLS])
    if model:
        command.extend(["--model", str(model)])
    if effort:
        command.extend(["--effort", str(effort)])
    if max_budget_usd is not None:
        budget = float(max_budget_usd)
        if budget <= 0:
            raise ClaudeRuntimeError("max_budget_usd must be greater than 0.")
        command.extend(["--max-budget-usd", str(max_budget_usd)])
    if resume_session_id:
        command.extend(["--resume", resume_session_id])
    command.append(prompt)
    return command


def git_status_snapshot(workspace: Path) -> bytes | None:
    inside_git = subprocess.run(
        ["git", "rev-parse", "--is-inside-work-tree"],
        cwd=str(workspace),
        text=True,
        capture_output=True,
        check=False,
    )
    if inside_git.returncode != 0:
        return None
    status = subprocess.run(
        ["git", "status", "--porcelain=v1", "-z"],
        cwd=str(workspace),
        capture_output=True,
        check=False,
    )
    if status.returncode != 0:
        return None
    return status.stdout


def git_status_text(workspace: Path) -> str:
    status = subprocess.run(
        ["git", "status", "--porcelain=v1"],
        cwd=str(workspace),
        text=True,
        capture_output=True,
        check=False,
    )
    if status.returncode != 0:
        return ""
    return status.stdout.strip()


def terminate_process_group(pid: int) -> dict[str, Any]:
    if sys.platform == "win32":
        completed = subprocess.run(
            ["taskkill", "/PID", str(pid), "/T", "/F"],
            text=True,
            capture_output=True,
            check=False,
        )
        return {
            "attempted": True,
            "delivered": completed.returncode == 0,
            "method": "taskkill",
            "stderr": completed.stderr.strip(),
        }

    try:
        os.killpg(pid, signal.SIGTERM)
        return {"attempted": True, "delivered": True, "method": "process-group"}
    except ProcessLookupError:
        return {"attempted": True, "delivered": False, "method": "process-group"}
    except PermissionError as error:
        return {"attempted": True, "delivered": False, "method": "process-group", "error": str(error)}


def run_process(
    command: list[str],
    *,
    workspace: Path,
    timeout_seconds: float,
    child_new_session: bool,
) -> dict[str, Any]:
    process = subprocess.Popen(
        command,
        cwd=str(workspace),
        text=True,
        stdin=subprocess.DEVNULL,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        start_new_session=child_new_session,
    )
    try:
        stdout, stderr = process.communicate(timeout=timeout_seconds)
        return {
            "returncode": process.returncode,
            "stdout": stdout,
            "stderr": stderr,
            "timed_out": False,
        }
    except subprocess.TimeoutExpired:
        if child_new_session:
            terminate_process_group(process.pid)
        else:
            process.terminate()
        stdout, stderr = process.communicate(timeout=5)
        return {
            "returncode": None,
            "stdout": stdout,
            "stderr": stderr,
            "timed_out": True,
        }


def execute_job(
    job: dict[str, Any],
    *,
    active_pid: int | None,
    child_new_session: bool,
) -> dict[str, Any]:
    workspace = Path(job["workspace_path"])
    request = job["request"]
    command = request["command"]
    write = bool(request["write"])
    timeout_seconds = float(request["timeout_seconds"])

    running_job = write_job(
        {
            **job,
            "status": "running",
            "pid": active_pid,
            "started_at": now_iso(),
        }
    )
    before_status = None if write else git_status_snapshot(workspace)

    try:
        result = run_process(
            command,
            workspace=workspace,
            timeout_seconds=timeout_seconds,
            child_new_session=child_new_session,
        )
        after_status = None if write else git_status_snapshot(workspace)
        changed_during_read_only = before_status is not None and after_status is not None and before_status != after_status
        stdout_json = parse_json_or_text(result["stdout"])
        session_id = extract_session_id(stdout_json)
        warning = conversation_warning(stdout_json)
        if result["timed_out"]:
            status = "failed"
            error_message = f"Claude Code timed out after {timeout_seconds:g} seconds."
        elif changed_during_read_only:
            status = "failed"
            error_message = "Claude Code changed files while write=false."
        elif result["returncode"] == 0:
            status = "completed"
            error_message = None
        else:
            status = "failed"
            error_message = f"Claude Code exited with status {result['returncode']}."

        completed_job = write_job(
            {
                **running_job,
                "status": status,
                "pid": None,
                "completed_at": now_iso(),
                "returncode": result["returncode"],
                "stdout": result["stdout"],
                "stdout_json": stdout_json,
                "stderr": result["stderr"],
                "error_message": error_message,
                "read_only_changed_files": git_status_text(workspace) if changed_during_read_only else "",
                "session": {
                    "resumed": bool(request.get("resume_session_id")),
                    "resume_session_id": request.get("resume_session_id"),
                    "new_conversation": bool(request.get("new_conversation")),
                    "session_id": session_id,
                },
                "conversation_warning": warning,
            }
        )
        if status == "completed":
            set_last_session_id(workspace, session_id)
        return completed_job
    except FileNotFoundError:
        failed_job = write_job(
            {
                **running_job,
                "status": "failed",
                "pid": None,
                "completed_at": now_iso(),
                "returncode": None,
                "stdout": "",
                "stdout_json": None,
                "stderr": "claude was not found.",
                "error_message": "Claude Code CLI is not installed or is not on PATH.",
            }
        )
        return failed_job


def create_task_job(
    workspace: Path,
    prompt: str,
    *,
    write: bool,
    background: bool,
    model: str | None,
    effort: str | None,
    max_budget_usd: int | float | None,
    timeout_seconds: float,
    resume_session_id: str | None,
    new_conversation: bool,
) -> dict[str, Any]:
    command = build_claude_command(
        prompt,
        write=write,
        model=model,
        effort=effort,
        max_budget_usd=max_budget_usd,
        resume_session_id=resume_session_id,
    )
    job_id = generate_job_id()
    return write_job(
        {
            "job_id": job_id,
            "status": "queued" if background else "created",
            "workspace_path": str(workspace),
            "pid": None,
            "command": command,
            "request": {
                "prompt": prompt,
                "write": write,
                "background": background,
                "model": model,
                "effort": effort,
                "max_budget_usd": max_budget_usd,
                "timeout_seconds": timeout_seconds,
                "resume_session_id": resume_session_id,
                "new_conversation": new_conversation,
                "command": command,
            },
            "stdout": "",
            "stdout_json": None,
            "stderr": "",
            "returncode": None,
            "error_message": None,
        }
    )


def spawn_background_worker(workspace: Path, job_id: str) -> subprocess.Popen[str]:
    return subprocess.Popen(
        [
            sys.executable,
            "-m",
            "claude_codex_mcp.worker",
            "--workspace-path",
            str(workspace),
            "--job-id",
            job_id,
        ],
        cwd=str(workspace),
        text=True,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        start_new_session=True,
    )


def claude_task(
    prompt: str,
    workspace_path: str | None = None,
    write: bool = DEFAULT_WRITE,
    background: bool = False,
    model: str | None = DEFAULT_MODEL,
    effort: str | None = DEFAULT_EFFORT,
    max_budget_usd: int | float | None = None,
    timeout_seconds: int | float | None = DEFAULT_TIMEOUT_SECONDS,
    new_conversation: bool = False,
    resume_session_id: str | None = None,
) -> dict[str, Any]:
    workspace = resolve_workspace_path(workspace_path)
    normalized_prompt = validate_task_prompt(prompt)
    timeout = normalize_timeout(timeout_seconds)
    start_new = bool(new_conversation or prompt_requests_new_conversation(normalized_prompt))
    resolved_resume_session_id = None if start_new else resume_session_id or get_last_session_id(workspace)
    job = create_task_job(
        workspace,
        normalized_prompt,
        write=bool(write),
        background=bool(background),
        model=model,
        effort=effort,
        max_budget_usd=max_budget_usd,
        timeout_seconds=timeout,
        resume_session_id=resolved_resume_session_id,
        new_conversation=start_new,
    )

    if background:
        worker = spawn_background_worker(workspace, job["job_id"])
        queued_job = write_job({**job, "status": "queued", "pid": worker.pid})
        return {
            "job_id": queued_job["job_id"],
            "status": queued_job["status"],
            "workspace_path": queued_job["workspace_path"],
            "pid": queued_job["pid"],
            "message": f"Claude Code task started in the background as {queued_job['job_id']}.",
        }

    return execute_job(job, active_pid=None, child_new_session=True)


def claude_ask(
    prompt: str,
    workspace_path: str | None = None,
    quick: bool = False,
    background: bool = False,
    model: str | None = DEFAULT_MODEL,
    effort: str | None = None,
    max_budget_usd: int | float | None = None,
    timeout_seconds: int | float | None = DEFAULT_TIMEOUT_SECONDS,
    new_conversation: bool = False,
) -> dict[str, Any]:
    return claude_task(
        workspace_path=workspace_path,
        prompt=prompt,
        write=False,
        background=background,
        model=model,
        effort="low" if quick else (effort or DEFAULT_EFFORT),
        max_budget_usd=max_budget_usd,
        timeout_seconds=timeout_seconds,
        new_conversation=new_conversation,
    )


def claude_fix(
    prompt: str,
    workspace_path: str | None = None,
    background: bool = False,
    model: str | None = DEFAULT_MODEL,
    effort: str | None = DEFAULT_EFFORT,
    max_budget_usd: int | float | None = None,
    timeout_seconds: int | float | None = DEFAULT_TIMEOUT_SECONDS,
    new_conversation: bool = False,
) -> dict[str, Any]:
    return claude_task(
        workspace_path=workspace_path,
        prompt=prompt,
        write=True,
        background=background,
        model=model,
        effort=effort,
        max_budget_usd=max_budget_usd,
        timeout_seconds=timeout_seconds,
        new_conversation=new_conversation,
    )


def claude_continue(
    prompt: str,
    workspace_path: str | None = None,
    write: bool = DEFAULT_WRITE,
    background: bool = False,
    model: str | None = DEFAULT_MODEL,
    effort: str | None = DEFAULT_EFFORT,
    max_budget_usd: int | float | None = None,
    timeout_seconds: int | float | None = DEFAULT_TIMEOUT_SECONDS,
) -> dict[str, Any]:
    workspace = resolve_workspace_path(workspace_path)
    resume_session_id = get_last_session_id(workspace)
    if not resume_session_id:
        raise ClaudeRuntimeError("No Claude session has been recorded in this Codex session yet.")
    return claude_task(
        workspace_path=str(workspace),
        prompt=prompt,
        write=write,
        background=background,
        model=model,
        effort=effort,
        max_budget_usd=max_budget_usd,
        timeout_seconds=timeout_seconds,
        resume_session_id=resume_session_id,
    )


def claude_status(workspace_path: str | None = None, job_id: str | None = None, all: bool = False) -> dict[str, Any]:
    workspace = resolve_workspace_path(workspace_path)
    if job_id:
        job = read_job(workspace, job_id)
        remember_job_session(workspace, job)
        return {
            "workspace_path": str(workspace),
            "job": job,
        }

    jobs = read_all_jobs(workspace)
    for job in jobs:
        if job.get("status") == "completed":
            remember_job_session(workspace, job)
            break
    visible_jobs = jobs if all else jobs[:DEFAULT_MAX_JOBS]
    running = [job for job in visible_jobs if job.get("status") in {"queued", "running"}]
    recent = [job for job in visible_jobs if job.get("status") not in {"queued", "running"}]
    return {
        "workspace_path": str(workspace),
        "running": running,
        "recent": recent,
        "count": len(visible_jobs),
        "total_count": len(jobs),
    }


def claude_result(workspace_path: str | None = None, job_id: str | None = None) -> dict[str, Any]:
    workspace = resolve_workspace_path(workspace_path)
    if job_id:
        job = read_job(workspace, job_id)
    else:
        job = latest_finished_job(workspace)
        if job is None:
            raise ClaudeRuntimeError("No finished Claude Code jobs were found for this workspace.")
    if job.get("status") not in {"completed", "failed", "cancelled"}:
        raise ClaudeRuntimeError(f"Claude Code job {job['job_id']} is still {job.get('status')}.")
    remember_job_session(workspace, job)
    return {
        "workspace_path": str(workspace),
        "job": job,
    }


def claude_cancel(workspace_path: str | None, job_id: str) -> dict[str, Any]:
    workspace = resolve_workspace_path(workspace_path)
    job = read_job(workspace, job_id)
    if job.get("status") not in {"queued", "running"}:
        return {
            "job_id": job_id,
            "status": job.get("status"),
            "cancelled": False,
            "message": f"Claude Code job {job_id} is not running.",
        }

    pid = job.get("pid")
    if not isinstance(pid, int):
        cancelled_job = write_job(
            {
                **job,
                "status": "cancelled",
                "completed_at": now_iso(),
                "pid": None,
                "error_message": "Cancelled before a process id was recorded.",
            }
        )
        return {
            "job_id": job_id,
            "status": cancelled_job["status"],
            "cancelled": True,
            "termination": {"attempted": False, "delivered": False, "method": None},
        }

    termination = terminate_process_group(pid)
    cancelled_job = write_job(
        {
            **job,
            "status": "cancelled",
            "completed_at": now_iso(),
            "pid": None,
            "error_message": "Cancelled by claude_cancel.",
            "termination": termination,
        }
    )
    return {
        "job_id": job_id,
        "status": cancelled_job["status"],
        "cancelled": True,
        "termination": termination,
    }


def run_worker(workspace_path: str, job_id: str) -> dict[str, Any]:
    workspace = resolve_workspace_path(workspace_path)
    job = read_job(workspace, job_id)
    if job.get("status") == "cancelled":
        return job
    return execute_job(job, active_pid=os.getpid(), child_new_session=False)