car-tracker/ops/ssh/host_run.py

"""Forced SSH command: accept only a fixed verb, UUID, revision and image digest."""

import fcntl
import json
import math
import os
from pathlib import Path
import re
import subprocess
import sys

from reconcile import attempt, command, compose_environment, validate_request


def limit_reasons(manifest, pipeline):
    """Recognize only verified intentional limits, never generic partial runs."""
    if (manifest.get("status") != "partial" or "error_type" in manifest or
            pipeline.get("import_status") != "succeeded" or
            pipeline.get("collection_status") != "partial" or
            "error_type" in pipeline):
        return []
    summary = pipeline.get("import_summary", {})
    if summary.get("rejected") != 0 or summary.get("run_id") != manifest.get("run_id"):
        return []
    searches = manifest.get("searches", [])
    if not searches or len(searches) != len(manifest.get("planned_searches", [])):
        return []
    reasons = set()
    for search in searches:
        if search.get("status") == "complete":
            continue
        if search.get("status") != "partial" or search.get("reason") != "page_limit":
            return []
        reasons.add("page_limit")
    details = manifest.get("details", {})
    if details.get("status") not in ("complete", "partial", "disabled"):
        return []
    for item in details.get("listings", {}).values():
        if item.get("status") == "complete":
            continue
        if item.get("status") != "not_attempted" or item.get("reason") != "detail_limit":
            return []
        reasons.add("detail_limit")
    return sorted(reasons)


def actions_outcome(code, request, state, raw_root):
    if code != 1:
        return code
    try:
        paths = []
        with (state / f"{request['request_id']}.log").open() as log:
            for line in log:
                if line.startswith("Manifest: "):
                    paths.append(line.removeprefix("Manifest: ").strip())
        if len(paths) != 1 or not re.fullmatch(
                r"/data/[0-9]{4}-[0-9]{2}-[0-9]{2}/[0-9a-f-]{36}/manifest\.json", paths[0]):
            return code
        raw_root = raw_root.resolve()
        path = (raw_root / paths[0].removeprefix("/data/")).resolve()
        if not path.is_relative_to(raw_root):
            return code
        manifest = json.loads(path.read_text())
        pipeline = json.loads(path.with_name("pipeline.json").read_text())
        reasons = limit_reasons(manifest, pipeline)
        if not reasons:
            return code
        explanations = {"page_limit": "the configured page limit left additional search pages unfetched",
                        "detail_limit": "the configured detail limit left some discovered ads without detail attempts"}
        message = "; ".join(explanations[reason] for reason in reasons)
        print(f"::warning::Import succeeded with no rejected records, but coverage is intentionally partial: {message}. "
              "This is not full source coverage. Host state retains pipeline exit 1.", flush=True)
        return 0
    except (OSError, ValueError, KeyError, TypeError, AttributeError):
        # Missing/malformed evidence must never turn a failure green.
        return code


def collection_command(settings):
    family = settings.get("COLLECT_FAMILY", "saab-900")
    pages = settings.get("COLLECT_MAX_PAGES", "1")
    details = settings.get("COLLECT_MAX_DETAILS", "1")
    delay = settings.get("COLLECT_DELAY", "5")
    if not isinstance(family, str) or not family.strip():
        raise ValueError("A collection family or all is required")
    if not isinstance(pages, str) or not pages.isdecimal() or int(pages) < 1:
        raise ValueError("Page limit must be a positive integer")
    if not isinstance(details, str) or (details != "all" and
                                      (not details.isdecimal() or int(details) < 1)):
        raise ValueError("Detail limit must be all or a positive integer")
    if not isinstance(delay, str) or not math.isfinite(float(delay)) or float(delay) < 1:
        raise ValueError("Delay must be finite and at least one second")
    args = ["car-tracker-pipeline", "--headed", "--config", "/app/searches.toml",
            "--output", "/data", "--max-pages", pages, "--delay", delay]
    if family != "all":
        args.extend(["--family", family])
    if details != "all":
        args.extend(["--max-details", details])
    return args


def parse_command(value):
    parts = value.split()
    if len(parts) != 4 or parts[0] != "run":
        raise ValueError("Only the fixed run command is allowed")
    return validate_request({"schema_version": 1, "request_id": parts[1],
                             "revision": parts[2], "image": parts[3]})


def run(root, request):
    state = root / "state/requests"
    state.mkdir(parents=True, mode=0o700, exist_ok=True)
    with (root / "state/agent.lock").open("a") as lock:
        try:
            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            raise ValueError("Another job is active")
        for path in state.glob("*.json"):
            if json.loads(path.read_text()).get("status") == "attempted":
                raise ValueError("Unknown previous outcome requires investigation")
        config = json.loads(command(
            "docker", "compose", "--env-file", str(root / "app.env"),
            "-f", str(root / "compose.runtime.yaml"), "config", "--format", "json",
            env=compose_environment(request["image"])))
        args = collection_command(config["services"]["app"]["environment"])
        code = attempt(request, state, root / "compose.runtime.yaml", root / "app.env",
                       private_log=True, app_command=args)
        raw_mounts = [volume["source"] for volume in config["services"]["app"].get("volumes", [])
                      if volume.get("target") == "/data" and volume.get("type") == "bind"]
        if len(raw_mounts) != 1:
            return code
        return actions_outcome(code, request, state, Path(raw_mounts[0]))


def main():
    os.umask(0o077)
    try:
        # Fixed location: the SSH caller cannot override the stack or environment.
        root = Path.home() / "containers/car-tracker"
        request = parse_command(os.environ.get("SSH_ORIGINAL_COMMAND", ""))
        return run(root, request)
    except (OSError, ValueError, KeyError, TypeError, subprocess.SubprocessError):
        print("Run refused or failed before completion. Inspect private host state/logs; do not automatically retry.")
        return 70


if __name__ == "__main__":
    sys.exit(main())