tests/test_adam_manager_foundation.py

from __future__ import annotations

from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.error import HTTPError
from urllib.request import Request, urlopen
import json
import socket

from PIL import Image

from adam.assets import AssetRegistry
from adam.dataset_lab import scan_dataset
from adam.model_profiles import ModelProfileRegistry
from adam.recommendations import recommend_for_profile
from adam.registry import ToolRegistry
from adam.remote_access import (
    REMOTE_MODE_DISABLED,
    REMOTE_MODE_TAILSCALE,
    RemoteAccessService,
    inspect_tailscale,
    remote_scope,
)
from adam.models import ExecutionPlan, Job, JobStatus, PlanStep, SystemSnapshot
from adam.transcript_dataset import clean_transcript, split_transcript, transcript_videos_to_dataset


def test_model_profiles_expose_manifest_metadata() -> None:
    profiles = {profile.id: profile for profile in ModelProfileRegistry(ToolRegistry(Path.cwd()).model_plugins).all()}

    assert profiles["ddpm"].hardware["recommended_vram_gb"] >= 6
    assert "resolution" in profiles["flow"].training
    assert "safetensors" in profiles["lora"].output_formats


def test_profile_recommendation_uses_schema_and_warns_on_small_vram() -> None:
    profile = ModelProfileRegistry(ToolRegistry(Path.cwd()).model_plugins).get("ddpm")

    result = recommend_for_profile(profile, dataset_items=1_000, resolution=128)  # type: ignore[arg-type]

    assert result.epochs == 180
    assert result.settings["batch_size"] == 12
    assert result.estimated_vram_gb is not None


def test_dataset_lab_scans_images_captions_and_duplicates(tmp_path: Path) -> None:
    image = Image.new("RGB", (16, 24), (10, 20, 30))
    first = tmp_path / "first.png"
    second = tmp_path / "second.png"
    image.save(first)
    image.save(second)
    first.with_suffix(".txt").write_text("caption", encoding="utf-8")

    report = scan_dataset(tmp_path)

    assert report.image_count == 2
    assert report.caption_count == 1
    assert report.missing_caption_count == 1
    assert report.duplicate_groups == 1
    assert report.dimensions["16x24"] == 2


def test_transcript_cleaning_and_splitting() -> None:
    text = "Hello   world .\nThis is ADAM! " * 20

    samples = split_transcript(clean_transcript(text), max_chars=80)

    assert len(samples) > 1
    assert all(len(sample) <= 90 for sample in samples)


def test_remote_access_defaults_to_disabled(tmp_path: Path) -> None:
    class Config:
        values = {}

        def get(self, key, default=None):
            return self.values.get(key, default)

        def update(self, values):
            self.values.update(values)

    service = RemoteAccessService(Config(), jobs=None, monitor=None)

    assert service.settings()["enabled"] is False
    assert service.settings()["bind_address"] == "127.0.0.1"
    assert service.settings()["remote_mode"] == "local_wifi"
    assert service.phone_test_url() == ""


def test_remote_access_requires_token_and_reports_safe_permissions() -> None:
    class Config:
        def __init__(self) -> None:
            self.values = {}

        def get(self, key, default=None):
            return self.values.get(key, default)

        def update(self, values):
            self.values.update(values)

    class Jobs:
        active_job = None
        jobs = []

    class Monitor:
        @staticmethod
        def snapshot():
            return SystemSnapshot(cpu_percent=12, memory_percent=34)

    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        port = sock.getsockname()[1]
    config = Config()
    service = RemoteAccessService(config, Jobs(), Monitor())
    settings = service.settings()
    service.save_settings({"enabled": True, "port": port, "token": settings["token"]})
    try:
        url = service.start()
        try:
            urlopen(url, timeout=3)
        except HTTPError as exc:
            assert exc.code == 401
        else:
            raise AssertionError("Remote status URL should require a token")
        local_payload = urlopen(service.local_test_url(), timeout=3).read().decode("utf-8")
        assert "<title>ADAM Remote</title>" in local_payload
        request = Request(url, headers={"Authorization": f"Bearer {settings['token']}"})
        payload = urlopen(request, timeout=3).read().decode("utf-8")
        assert '"dangerous_actions": false' in payload
        assert '"cpu_percent": 12' in payload
    finally:
        service.stop()


def test_remote_access_scope_labels() -> None:
    assert remote_scope("127.0.0.1") == "local-device only"
    assert remote_scope("172.16.0.5") == "local network"
    assert remote_scope("172.15.0.5") == "custom bind address"
    assert remote_scope("0.0.0.0") == "all network interfaces"


def test_remote_access_phone_url_uses_network_bind_address() -> None:
    class Config:
        def __init__(self) -> None:
            self.values = {}

        def get(self, key, default=None):
            return self.values.get(key, default)

        def update(self, values):
            self.values.update(values)

    service = RemoteAccessService(Config(), jobs=None, monitor=None)
    token = service.settings()["token"]
    service.save_settings({"enabled": True, "bind_address": "192.168.1.25", "port": 8765, "token": token})

    assert service.phone_test_url() == f"http://192.168.1.25:8765/?token={token}"


def test_remote_mode_disabled_refuses_to_start() -> None:
    class Config:
        def __init__(self) -> None:
            self.values = {}

        def get(self, key, default=None):
            return self.values.get(key, default)

        def update(self, values):
            self.values.update(values)

    service = RemoteAccessService(Config(), jobs=None, monitor=None)
    service.save_settings({"enabled": True, "remote_mode": REMOTE_MODE_DISABLED})

    try:
        service.start()
    except RuntimeError as exc:
        assert "Remote Mode" in str(exc)
    else:
        raise AssertionError("Disabled remote mode should not start")


def test_tailscale_unavailable_and_available_states() -> None:
    assert inspect_tailscale(which=lambda _name: None).installed is False

    class Result:
        def __init__(self, returncode=0, stdout="", stderr=""):
            self.returncode = returncode
            self.stdout = stdout
            self.stderr = stderr

    disconnected = inspect_tailscale(
        which=lambda _name: "tailscale",
        runner=lambda _command: Result(1, stderr="not logged in"),
    )
    assert disconnected.installed is True
    assert disconnected.connected is False

    def runner(command):
        if command[1:3] == ["status", "--json"]:
            return Result(
                stdout=json.dumps(
                    {
                        "BackendState": "Running",
                        "Self": {
                            "HostName": "adam-pc",
                            "DNSName": "adam-pc.tailnet.ts.net.",
                            "TailscaleIPs": ["100.64.0.12", "fd7a:115c:a1e0::12"],
                        },
                    }
                )
            )
        return Result(stdout='{"TCP":{}}')

    connected = inspect_tailscale(which=lambda _name: "tailscale", runner=runner)
    assert connected.installed is True
    assert connected.connected is True
    assert connected.device_name == "adam-pc"
    assert connected.dns_name == "adam-pc.tailnet.ts.net"
    assert connected.tailscale_ip == "100.64.0.12"


def test_remote_access_mobile_dashboard_prompt_and_preview(tmp_path: Path) -> None:
    class Config:
        def __init__(self) -> None:
            self.values = {}

        def get(self, key, default=None):
            return self.values.get(key, default)

        def update(self, values):
            self.values.update(values)

    class Planner:
        def plan(self, prompt: str) -> ExecutionPlan:
            return ExecutionPlan(
                request=prompt,
                summary="Generate one image.",
                steps=[PlanStep("preview_generator", "Generate preview", "Generate one preview.", {})],
                project_name="Phone Prompt",
            )

    class Jobs:
        def __init__(self) -> None:
            preview = tmp_path / "preview.png"
            Image.new("RGB", (8, 8), (10, 120, 240)).save(preview)
            plan = ExecutionPlan(
                "train",
                "Training.",
                [],
                project_name="Phone Preview",
                orion_review={"estimated_high_minutes": 20},
            )
            self.active_job = Job(
                plan=plan,
                status=JobStatus.RUNNING,
                progress=55,
                started_at=(datetime.now(timezone.utc) - timedelta(minutes=5)).isoformat(),
                preview_path=str(preview),
                preview_epoch=3,
                preview_kind="training",
            )
            self.jobs = [self.active_job]
            self.submitted: list[ExecutionPlan] = []

        def submit(self, plan: ExecutionPlan) -> Job:
            self.submitted.append(plan)
            job = Job(plan=plan, status=JobStatus.QUEUED)
            self.jobs.insert(0, job)
            return job

    class Monitor:
        @staticmethod
        def snapshot():
            return SystemSnapshot(cpu_percent=12, memory_percent=34, gpu_percent=56, vram_percent=78)

    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        port = sock.getsockname()[1]
    jobs = Jobs()
    config = Config()
    service = RemoteAccessService(config, jobs, Monitor(), Planner())
    token = service.settings()["token"]
    service.save_settings({"enabled": True, "port": port, "token": token})
    try:
        service.start()
        html = urlopen(service.local_test_url(), timeout=3).read().decode("utf-8")
        assert "Prompt ADAM" in html
        assert "Create Model" in html
        assert "Collect Dataset" in html
        assert "Quick" in html
        assert "Auto-approve remote training" in html
        assert "Keep screen updated" in html
        assert "Time Left" in html
        assert "Live Preview" in html
        active_script = html.split("<script>", 1)[1].split("</script>", 1)[0]
        assert "XMLHttpRequest" in active_script
        assert "window.addEventListener(\"error\"" in active_script
        assert "fetch(" not in active_script
        assert "async " not in active_script
        assert "=>" not in active_script
        assert "?." not in active_script
        assert "??" not in active_script

        status = json.loads(urlopen(service.url() + f"?token={token}", timeout=3).read().decode("utf-8"))
        assert status["preview"]["available"] is True
        assert status["permissions"]["prompt"] is True
        assert status["permissions"]["auto_approve_training"] is False
        assert status["active_job"]["timing"]["remaining_seconds"] is not None
        assert status["active_job"]["timing"]["estimate_label"]

        settings_request = Request(
            f"http://127.0.0.1:{port}/api/remote-settings?token={token}",
            data=json.dumps({"auto_approve_training": True}).encode("utf-8"),
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        # The desktop owner must grant control before a phone may enable approval.
        service.save_settings({"allow_job_control": True})
        settings_response = json.loads(urlopen(settings_request, timeout=3).read().decode("utf-8"))
        assert settings_response["ok"] is True
        status = json.loads(urlopen(service.url() + f"?token={token}", timeout=3).read().decode("utf-8"))
        assert status["permissions"]["auto_approve_training"] is True

        preview = urlopen(f"http://127.0.0.1:{port}/api/preview?token={token}", timeout=3).read()
        assert preview.startswith(b"\x89PNG")

        request = Request(
            f"http://127.0.0.1:{port}/api/prompt?token={token}",
            data=json.dumps({"prompt": "generate one preview"}).encode("utf-8"),
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        response = json.loads(urlopen(request, timeout=3).read().decode("utf-8"))
        assert response["ok"] is True
        assert response["job_id"]
        assert jobs.submitted[0].project_name == "Phone Prompt"

        message_request = Request(
            f"http://127.0.0.1:{port}/api/prompt?token={token}",
            data=json.dumps({"message": "generate one preview"}).encode("utf-8"),
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        message_response = json.loads(urlopen(message_request, timeout=3).read().decode("utf-8"))
        assert message_response["ok"] is True
        assert jobs.submitted[0].project_name == "Phone Prompt"
    finally:
        service.stop()


def test_remote_job_actions_include_existing_controls() -> None:
    class Config:
        def __init__(self) -> None:
            self.values = {}

        def get(self, key, default=None):
            return self.values.get(key, default)

        def update(self, values):
            self.values.update(values)

    plan = ExecutionPlan(
        request="train",
        summary="Training.",
        steps=[PlanStep("ddpm_trainer", "Train model", "Run training")],
        requires_confirmation=True,
        project_name="Phone Controls",
    )
    job = Job(plan=plan, status=JobStatus.AWAITING_CONFIRMATION, current_step=0)

    class Jobs:
        active_job = job
        jobs = [job]

        def __init__(self) -> None:
            self.actions: list[tuple[str, str]] = []

        def get(self, job_id: str) -> Job:
            assert job_id == job.id
            return job

        def confirm(self, job_id: str) -> None:
            self.actions.append(("confirm", job_id))
            job.status = JobStatus.QUEUED

        def pause(self, job_id: str) -> None:
            self.actions.append(("pause", job_id))
            job.status = JobStatus.PAUSED

        def resume(self, job_id: str) -> None:
            self.actions.append(("resume", job_id))
            job.status = JobStatus.RUNNING

        def cancel(self, job_id: str) -> None:
            self.actions.append(("cancel", job_id))
            job.status = JobStatus.CANCELLED

        def retry(self, job_id: str) -> Job:
            self.actions.append(("retry", job_id))
            return Job(plan=plan, status=JobStatus.AWAITING_CONFIRMATION)

        def end_task(self, job_id: str) -> bool:
            self.actions.append(("end", job_id))
            return True

    jobs = Jobs()
    service = RemoteAccessService(Config(), jobs, monitor=None)

    summary = service._job_summary(job)
    assert summary["current_step_title"] == "Train model"

    assert service.job_action(job.id, "confirm", True)["ok"] is True
    assert service.job_action(job.id, "pause", True)["ok"] is True
    assert service.job_action(job.id, "resume", True)["ok"] is True
    assert service.job_action(job.id, "retry", True)["ok"] is True
    assert service.job_action(job.id, "end", True)["ok"] is True
    assert [action for action, _job_id in jobs.actions] == [
        "confirm",
        "pause",
        "resume",
        "retry",
        "end",
    ]
    assert service.job_action(job.id, "pause", False)["ok"] is False


def test_remote_prompt_queues_image_generation_before_conversation(tmp_path: Path) -> None:
    class Config:
        def __init__(self) -> None:
            self.values = {"tool_folders": {}, "generation_settings": {}}

        def get(self, key, default=None):
            return self.values.get(key, default)

        def update(self, values):
            self.values.update(values)

    class Jobs:
        def __init__(self) -> None:
            self.active_job = None
            self.jobs = []
            self.submitted: list[ExecutionPlan] = []

        def submit(self, plan: ExecutionPlan) -> Job:
            self.submitted.append(plan)
            job = Job(plan=plan, status=JobStatus.QUEUED)
            self.jobs.insert(0, job)
            return job

    class Planner:
        def __init__(self, assets: AssetRegistry, registry: ToolRegistry) -> None:
            self.assets = assets
            self.registry = registry

        def plan(self, _prompt: str) -> ExecutionPlan:
            raise AssertionError("Image generation should not fall back to conversation planning.")

    model = tmp_path / "ddpm" / "output" / "Minecraft"
    model.mkdir(parents=True)
    (model / "model_index.json").write_text("{}", encoding="utf-8")
    assets = AssetRegistry(tmp_path)
    assets.register(kind="model", name="Minecraft", path=str(model), trainer="ddpm")
    jobs = Jobs()
    service = RemoteAccessService(
        Config(),
        jobs=jobs,
        monitor=None,
        planner=Planner(assets, ToolRegistry(Path.cwd())),
    )

    response = service.submit_prompt("Generate an image of Minecraft")

    assert response["ok"] is True
    assert response["job_id"]
    assert jobs.submitted[0].steps[0].tool_id == "ddpm_generator"
    assert jobs.submitted[0].steps[0].arguments["model_name"] == "Minecraft"
    assert jobs.submitted[0].steps[0].arguments["prompt"] == "Minecraft"


def test_remote_prompt_can_auto_approve_training_plan(tmp_path: Path) -> None:
    class Config:
        def __init__(self) -> None:
            self.values = {"remote_access": {"auto_approve_training": True}}

        def get(self, key, default=None):
            return self.values.get(key, default)

        def update(self, values):
            self.values.update(values)

    class Planner:
        def plan(self, prompt: str) -> ExecutionPlan:
            return ExecutionPlan(
                request=prompt,
                summary="Train DDPM.",
                steps=[PlanStep("ddpm_trainer", "Train", "Train a model.", {})],
                requires_confirmation=True,
                confirmation_reason="This starts training.",
                project_name="Remote Training",
            )

    class Jobs:
        def __init__(self) -> None:
            self.active_job = None
            self.jobs = []
            self.confirmed: list[str] = []

        def submit(self, plan: ExecutionPlan) -> Job:
            job = Job(plan=plan, status=JobStatus.AWAITING_CONFIRMATION)
            self.jobs.insert(0, job)
            return job

        def confirm(self, job_id: str) -> None:
            self.confirmed.append(job_id)
            self.jobs[0].status = JobStatus.QUEUED

    jobs = Jobs()
    service = RemoteAccessService(Config(), jobs=jobs, monitor=None, planner=Planner())

    response = service.submit_prompt("train a DDPM model")

    assert response["ok"] is True
    assert response["auto_approved"] is True
    assert response["requires_approval"] is False
    assert jobs.confirmed == [response["job_id"]]


def test_remote_prompt_selects_flow_match_generation_model(tmp_path: Path) -> None:
    class Config:
        def __init__(self) -> None:
            self.values = {"tool_folders": {}, "generation_settings": {}}

        def get(self, key, default=None):
            return self.values.get(key, default)

        def update(self, values):
            self.values.update(values)

    class Jobs:
        def __init__(self) -> None:
            self.active_job = None
            self.jobs = []
            self.submitted: list[ExecutionPlan] = []

        def submit(self, plan: ExecutionPlan) -> Job:
            self.submitted.append(plan)
            job = Job(plan=plan, status=JobStatus.QUEUED)
            self.jobs.insert(0, job)
            return job

    class Planner:
        def __init__(self, assets: AssetRegistry, registry: ToolRegistry) -> None:
            self.assets = assets
            self.registry = registry

        def plan(self, _prompt: str) -> ExecutionPlan:
            raise AssertionError("Flow image generation should not fall back to conversation planning.")

    model = tmp_path / "flow" / "output_flow_models" / "Minecraft Flow Match"
    (model / "unet").mkdir(parents=True)
    (model / "flow_model_info.json").write_text("{}", encoding="utf-8")
    (model / "unet" / "config.json").write_text("{}", encoding="utf-8")
    assets = AssetRegistry(tmp_path)
    assets.register(kind="model", name="Minecraft Flow Match", path=str(model), trainer="flow")
    jobs = Jobs()
    service = RemoteAccessService(
        Config(),
        jobs=jobs,
        monitor=None,
        planner=Planner(assets, ToolRegistry(Path.cwd())),
    )

    response = service.submit_prompt("Generate an image of Minecraft Flow Match")

    assert response["ok"] is True
    assert jobs.submitted[0].steps[0].tool_id == "flow_generator"
    assert jobs.submitted[0].steps[0].arguments["model_name"] == "Minecraft Flow Match"
    assert jobs.submitted[0].steps[0].arguments["sampler"] == "Heun"


def test_remote_status_serves_latest_generation_image(tmp_path: Path) -> None:
    class Config:
        def __init__(self) -> None:
            self.values = {}

        def get(self, key, default=None):
            return self.values.get(key, default)

        def update(self, values):
            self.values.update(values)

    class Planner:
        root = tmp_path

        def __init__(self) -> None:
            self.registry = ToolRegistry(Path.cwd())

    class Jobs:
        active_job = None
        jobs = []

    folder = tmp_path / "data" / "generations" / "flow_generator" / "Minecraft Flow Match"
    folder.mkdir(parents=True)
    image = folder / "image_001.png"
    second_image = folder / "image_002.png"
    Image.new("RGB", (8, 8), (20, 160, 80)).save(image)
    Image.new("RGB", (8, 8), (140, 60, 220)).save(second_image)
    (folder / "generation_20260830_TEST.json").write_text(
        json.dumps(
            {
                "provider_id": "flow_generator",
                "provider_name": "Flow Matching Generator",
                "model_name": "Minecraft Flow Match",
                "model_path": "D:/Flow/Minecraft Flow Match",
                "prompt": "Minecraft Flow Match",
                "seed": 0,
                "steps": 20,
                "sampler": "Heun",
                "aspect_ratio": "1:1 (Square)",
                "images": [str(image), str(second_image)],
                "created_at": "2026-08-30T00:00:00+00:00",
            }
        ),
        encoding="utf-8",
    )

    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        port = sock.getsockname()[1]
    config = Config()
    service = RemoteAccessService(config, Jobs(), monitor=None, planner=Planner())
    token = service.settings()["token"]
    service.save_settings({"enabled": True, "port": port, "token": token})
    try:
        service.start()
        status = json.loads(urlopen(service.url() + f"?token={token}", timeout=3).read().decode("utf-8"))
        assert status["latest_generation"]["available"] is True
        assert status["latest_generation"]["model_name"] == "Minecraft Flow Match"
        assert status["latest_generation"]["image_count"] == 2
        assert status["latest_generation"]["images"][1]["url"].endswith("image=1")

        generated = urlopen(
            f"http://127.0.0.1:{port}/api/generation-image?record=0&image=0&token={token}",
            timeout=3,
        ).read()
        assert generated.startswith(b"\x89PNG")
        generated_second = urlopen(
            f"http://127.0.0.1:{port}/api/generation-image?record=0&image=1&token={token}",
            timeout=3,
        ).read()
        assert generated_second.startswith(b"\x89PNG")
    finally:
        service.stop()


def test_remote_job_controls_are_gated(tmp_path: Path) -> None:
    class Config:
        def __init__(self) -> None:
            self.values = {}

        def get(self, key, default=None):
            return self.values.get(key, default)

        def update(self, values):
            self.values.update(values)

    class Jobs:
        def __init__(self) -> None:
            self.job = Job(
                plan=ExecutionPlan("run", "Run.", [], project_name="Remote Job"),
                status=JobStatus.QUEUED,
            )
            self.jobs = [self.job]
            self.active_job = None
            self.cancelled: list[str] = []

        def get(self, job_id: str):
            return self.job if job_id == self.job.id else None

        def cancel(self, job_id: str) -> None:
            self.cancelled.append(job_id)

        def retry(self, job_id: str) -> Job:
            return Job(plan=self.job.plan, status=JobStatus.QUEUED)

    jobs = Jobs()
    service = RemoteAccessService(Config(), jobs=jobs, monitor=None)
    assert service.job_action(jobs.job.id, "cancel", False)["ok"] is False
    assert service.job_action(jobs.job.id, "cancel", True)["ok"] is True
    assert jobs.cancelled == [jobs.job.id]


def test_transcript_export_reports_missing_ffmpeg(monkeypatch, tmp_path: Path) -> None:
    monkeypatch.setattr("adam.transcript_dataset.shutil.which", lambda _name: None)

    result = transcript_videos_to_dataset([tmp_path / "video.mp4"], tmp_path / "out")

    assert result.available is False
    assert "FFmpeg" in result.message