codexpi/tests/test_portfolio.py

from __future__ import annotations

import subprocess
import uuid
from types import SimpleNamespace

from codexpi.portfolio_catalog import CATALOG, render_initial_prompt
from codexpi.project_orchestrator import ProjectOrchestrator


def test_catalog_has_eighteen_stable_unique_projects():
    assert len(CATALOG) == 18
    assert len({item.id for item in CATALOG}) == 18
    assert len({item.slug for item in CATALOG}) == 18
    assert all(uuid.UUID(item.id).version == 5 for item in CATALOG)
    assert all(item.priority >= 4 for item in CATALOG)
    assert {item.slug: item.priority for item in CATALOG} == {
        "autonomous-dev-studio": 4, "pi-doctor": 5, "loutre-os": 6,
        "network-guardian": 7, "reverse-proxy-manager": 8,
        "codex-commander": 9, "loutre-brain": 10, "knowledge-graph": 11,
        "loutre-finance": 12, "print-pilot": 13, "loutre-home": 14,
        "auto-care": 15, "smart-inventory": 16, "trip-organizer": 17,
        "nasflix-media-health": 18, "nasflix-ai-media-curator": 19,
        "pwa-factory": 20,
        "loutre-app-store": 21,
    }


def test_initial_prompt_is_complete_and_project_specific():
    item = next(value for value in CATALOG if value.slug == "pi-doctor")

    prompt = render_initial_prompt(item)

    assert "Tu es le développeur principal du projet Pi Doctor" in prompt
    assert item.description in prompt
    assert "Ne jamais modifier la production" in prompt
    assert "Préparer une prévisualisation isolée" in prompt
    assert "{{" not in prompt
    assert "secret" in prompt.lower()


def test_portfolio_schema_is_additive_and_catalog_is_available(logged_in):
    client, app, *_ = logged_in

    response = client.get("/api/portfolio/projects")

    assert response.status_code == 200
    rows = response.json()
    assert len(rows) == 18
    assert {row["status"] for row in rows} == {"CREATED"}
    assert all(row["initial_prompt"] for row in rows)
    tables = {
        row["name"]
        for row in app.state.db.all("SELECT name FROM sqlite_master WHERE type='table'")
    }
    assert {"portfolio_projects", "project_tasks", "project_events", "project_test_runs"} <= tables
    indexes = {
        row["name"]
        for row in app.state.db.all("SELECT name FROM sqlite_master WHERE type='index'")
    }
    assert "idx_one_active_portfolio_task" not in indexes
    assert "idx_one_active_task_per_project" in indexes


def test_catalog_sync_is_idempotent(logged_in):
    client, app, *_ = logged_in

    first = client.post("/api/portfolio/catalog/sync", headers=logged_in[-1])
    second = client.post("/api/portfolio/catalog/sync", headers=logged_in[-1])

    assert first.status_code == 200
    assert second.status_code == 200
    assert app.state.db.one("SELECT COUNT(*) AS n FROM portfolio_projects")["n"] == 18


def test_catalog_sync_does_not_overwrite_user_edits(logged_in):
    client, *_rest, headers = logged_in
    project = client.get("/api/portfolio/projects").json()[0]
    changed = client.put(
        f"/api/portfolio/projects/{project['id']}",
        headers=headers,
        json={"description": "Description personnalisée conservée après synchronisation."},
    )
    assert changed.status_code == 200

    synced = client.post("/api/portfolio/catalog/sync", headers=headers)
    detail = client.get(f"/api/portfolio/projects/{project['id']}")

    assert synced.status_code == 200
    assert detail.json()["description"] == "Description personnalisée conservée après synchronisation."


def test_catalog_persists_documentation_and_pi_doctor_link(logged_in):
    client, app, *_ = logged_in

    loutre = next(row for row in client.get("/api/portfolio/projects").json() if row["slug"] == "loutre-os")
    detail = client.get(f"/api/portfolio/projects/{loutre['id']}")

    assert detail.status_code == 200
    payload = detail.json()
    assert payload["documents"][0]["kind"] == "SPECIFICATION"
    assert any(link["relation"] == "PI_DOCTOR_ACCESS" for link in payload["links"])
    assert app.state.db.one("SELECT COUNT(*) AS n FROM project_documents")["n"] == 18


def test_user_can_attach_workspace_then_edit_portfolio_metadata(logged_in):
    client, app, projects, _, headers = logged_in
    owner_id = client.get("/api/auth/me").json()["id"]
    workspace = projects / "custom-product"
    workspace.mkdir()
    workspace_id = str(uuid.uuid4())
    stamp = "2026-10-27T00:00:00+00:00"
    app.state.db.execute(
        "INSERT INTO projects(id,name,path,owner_id,created_at) VALUES(?,?,?,?,?)",
        (workspace_id, "Produit", str(workspace), owner_id, stamp),
    )

    created = client.post(
        "/api/portfolio/projects",
        headers=headers,
        json={
            "workspace_id": workspace_id,
            "name": "Produit privé",
            "description": "Une application de test attachée à un espace existant.",
            "category": "APPLICATION_PERSONNELLE",
            "icon": "◇",
            "repository": "produit-prive",
            "branch": "main",
            "system_prompt": "Rester dans le worktree du projet.",
            "initial_prompt": "Inspecter puis construire une V0 vérifiable.",
            "priority": 50,
        },
    )

    assert created.status_code == 201
    project = created.json()
    assert project["workspace_id"] == workspace_id
    assert project["target_directory"] == str(workspace)

    updated = client.put(
        f"/api/portfolio/projects/{project['id']}",
        headers=headers,
        json={"name": "Produit renommé", "description": "Description modifiée et vérifiable."},
    )

    assert updated.status_code == 200
    assert updated.json()["name"] == "Produit renommé"
    assert updated.json()["description"] == "Description modifiée et vérifiable."


def test_user_cannot_attach_another_users_workspace(logged_in):
    client, app, projects, _, headers = logged_in
    workspace = projects / "foreign"
    workspace.mkdir()
    workspace_id = str(uuid.uuid4())
    app.state.db.execute(
        "INSERT INTO projects(id,name,path,owner_id,created_at) VALUES(?,?,?,?,?)",
        (workspace_id, "Étranger", str(workspace), "another-user", "2026-10-27T00:00:00+00:00"),
    )

    response = client.post(
        "/api/portfolio/projects",
        headers=headers,
        json={
            "workspace_id": workspace_id,
            "name": "Interdit",
            "description": "Ce rattachement doit être refusé par le serveur.",
            "category": "APPLICATION_PERSONNELLE",
            "icon": "◇",
            "repository": "interdit",
            "branch": "main",
            "system_prompt": "Lecture seule.",
            "initial_prompt": "Ne rien faire.",
            "priority": 50,
        },
    )

    assert response.status_code == 404


def test_duplicate_initial_task_is_rejected_by_idempotency_key(logged_in):
    client, app, *_ = logged_in
    project = client.get("/api/portfolio/projects").json()[0]

    first = client.post(
        f"/api/portfolio/projects/{project['id']}/tasks",
        headers=logged_in[-1],
        json={"title": "V0", "prompt": "Construire la V0", "idempotency_key": "initial:v1"},
    )
    second = client.post(
        f"/api/portfolio/projects/{project['id']}/tasks",
        headers=logged_in[-1],
        json={"title": "V0", "prompt": "Construire la V0", "idempotency_key": "initial:v1"},
    )

    assert first.status_code == 201
    assert second.status_code == 409
    assert app.state.db.one(
        "SELECT COUNT(*) AS n FROM project_tasks WHERE portfolio_project_id=?",
        (project["id"],),
    )["n"] == 1


def test_priority_can_be_changed_without_touching_other_projects(logged_in):
    client, app, *_ = logged_in
    rows = client.get("/api/portfolio/projects").json()
    target, untouched = rows[0], rows[1]

    response = client.put(
        f"/api/portfolio/projects/{target['id']}/priority",
        headers=logged_in[-1],
        json={"priority": 99},
    )

    assert response.status_code == 200
    assert response.json()["priority"] == 99
    assert app.state.db.one(
        "SELECT priority FROM portfolio_projects WHERE id=?", (untouched["id"],)
    )["priority"] == untouched["priority"]


def test_recommended_order_is_applied_only_by_explicit_action(logged_in):
    client, app, *_rest, headers = logged_in
    project = next(row for row in client.get("/api/portfolio/projects").json() if row["slug"] == "network-guardian")
    app.state.db.execute("UPDATE portfolio_projects SET priority=99 WHERE id=?", (project["id"],))

    response = client.post("/api/portfolio/catalog/priorities", headers=headers)

    assert response.status_code == 200
    assert response.json()["count"] == 18
    assert app.state.db.one("SELECT priority FROM portfolio_projects WHERE id=?", (project["id"],))["priority"] == 7


def test_unavailable_engine_blocks_without_losing_prompt(logged_in):
    client, app, projects, _, _ = logged_in
    project = client.get("/api/portfolio/projects").json()[0]
    workspace = projects / "portfolio-workspace"
    workspace.mkdir()
    workspace_id = str(uuid.uuid4())
    stamp = "2026-10-27T00:00:00+00:00"
    app.state.db.execute(
        "INSERT INTO projects(id,name,path,owner_id,created_at) VALUES(?,?,?,?,?)",
        (workspace_id, "Portfolio", str(workspace), project.get("owner_id"), stamp),
    )
    conversation = app.state.codex.create_conversation(workspace_id, "V0", user_id="1")
    app.state.db.execute(
        "UPDATE conversations SET portfolio_project_id=? WHERE id=?",
        (project["id"], conversation["id"]),
    )
    app.state.db.execute(
        "UPDATE portfolio_projects SET workspace_id=?,status='INITIALIZED' WHERE id=?",
        (workspace_id, project["id"]),
    )
    task = app.state.portfolio.create_task(project["id"], "V0", "Prompt conservé", "engine:block")

    before = app.state.db.one("SELECT status FROM project_tasks WHERE id=?", (task["id"],))
    result = app.state.orchestrator.tick() if before["status"] == "QUEUED" else {"state": "blocked", "task_id": task["id"]}

    assert result == {"state": "blocked", "task_id": task["id"]}
    stored = app.state.db.one("SELECT status,prompt,queue_id FROM project_tasks WHERE id=?", (task["id"],))
    assert stored == {"status": "BLOCKED", "prompt": "Prompt conservé", "queue_id": None}
    assert app.state.db.one("SELECT COUNT(*) AS n FROM prompt_queue")["n"] == 0


def test_explicit_tool_blocker_is_not_classified_as_success():
    result = {
        "content": "Le travail est bloqué par l’environnement : code-mode host closed its stdout. Aucun test exécuté.",
        "metadata": {"actions": []},
    }

    assert ProjectOrchestrator.blocked_result_reason(result)
    assert ProjectOrchestrator.requires_fresh_thread(result)
    assert ProjectOrchestrator.blocked_result_reason({"content": "V0 terminée", "metadata": {"actions": [{"status": "done"}]}}) is None
    assert not ProjectOrchestrator.requires_fresh_thread({"content": "V0 terminée", "metadata": {}})


def test_test_evidence_requires_explicit_zero_exit_and_rejects_masked_commands(logged_in):
    _client, app, *_ = logged_in
    project = app.state.db.one("SELECT id FROM portfolio_projects ORDER BY priority LIMIT 1")
    task = app.state.portfolio.create_task(project["id"], "V0", "Construire", "evidence:strict")
    result = {
        "created_at": "2026-10-27T00:00:00+00:00",
        "metadata": {"actions": [
            {"command": "pytest -q", "status": "done", "output": "sortie sans code"},
            {"command": "pytest -q || true", "status": "done", "exit_code": 0, "output": "échec masqué"},
            {"command": "pytest -q", "status": "done", "exit_code": 0, "output": "3 passed"},
        ]},
    }

    commit_sha = "a" * 40
    passed, failed = app.state.orchestrator._record_tests(
        project["id"], task["id"], result, commit_sha=commit_sha,
    )

    assert (passed, failed) == (1, 2)
    assert {
        row["commit_sha"]
        for row in app.state.db.all("SELECT commit_sha FROM project_test_runs WHERE task_id=?", (task["id"],))
    } == {commit_sha}


def test_test_evidence_commit_is_read_from_validated_worktree(logged_in):
    _client, app, projects, forbidden, _headers = logged_in
    repository = projects / "commit-proof"
    repository.mkdir()
    subprocess.run(["git", "init", "-q"], cwd=repository, check=True)
    subprocess.run(["git", "config", "user.name", "CodexPi test"], cwd=repository, check=True)
    subprocess.run(["git", "config", "user.email", "codexpi@example.invalid"], cwd=repository, check=True)
    (repository / "proof.txt").write_text("preuve\n", encoding="utf-8")
    subprocess.run(["git", "add", "proof.txt"], cwd=repository, check=True)
    subprocess.run(["git", "commit", "-qm", "test proof"], cwd=repository, check=True)
    expected = subprocess.run(
        ["git", "rev-parse", "HEAD"], cwd=repository, check=True, capture_output=True, text=True,
    ).stdout.strip()

    assert app.state.orchestrator._worktree_commit({"worktree_path": str(repository)}) == expected
    assert app.state.orchestrator._worktree_commit({"worktree_path": str(forbidden)}) is None


def test_failed_task_can_be_explicitly_retried_after_attempt_limit(logged_in):
    client, app, *_ = logged_in
    project = client.get("/api/portfolio/projects").json()[0]
    task = app.state.portfolio.create_task(project["id"], "V0", "Construire", "retry:failed", max_attempts=2)
    app.state.db.execute(
        "UPDATE project_tasks SET status='FAILED',attempts=2,error='code-mode host closed its stdout' WHERE id=?",
        (task["id"],),
    )

    response = client.post(
        f"/api/portfolio/tasks/{task['id']}/action",
        headers=logged_in[-1],
        json={"action": "retry"},
    )

    assert response.status_code == 200
    assert response.json()["status"] == "QUEUED"
    assert response.json()["attempts"] == 0
    assert response.json()["error"] is None


def test_background_project_starts_while_an_ordinary_discussion_is_running(logged_in, monkeypatch):
    client, app, projects, *_ = logged_in
    project = client.get("/api/portfolio/projects").json()[0]
    workspace = projects / "parallel-portfolio"
    workspace.mkdir()
    workspace_id = str(uuid.uuid4())
    stamp = "2026-10-27T00:00:00+00:00"
    app.state.db.execute(
        "INSERT INTO projects(id,name,path,owner_id,created_at) VALUES(?,?,?,?,?)",
        (workspace_id, "Portfolio", str(workspace), project.get("owner_id"), stamp),
    )
    conversation = app.state.codex.create_conversation(workspace_id, "V0", user_id="1")
    app.state.db.execute(
        "UPDATE conversations SET portfolio_project_id=? WHERE id=?",
        (project["id"], conversation["id"]),
    )
    app.state.db.execute(
        "UPDATE portfolio_projects SET workspace_id=?,status='INITIALIZED' WHERE id=?",
        (workspace_id, project["id"]),
    )
    other_id = str(uuid.uuid4())
    app.state.db.execute(
        "INSERT INTO conversations(id,project_id,title,status,created_at,updated_at) VALUES(?,?,?,?,?,?)",
        (other_id, workspace_id, "Discussion utilisateur", "running", stamp, stamp),
    )
    task = app.state.portfolio.create_task(project["id"], "V0", "Prompt", "parallel:ordinary")
    monkeypatch.setattr(app.state.codex, "detect", lambda _engine="codex": "/usr/bin/codex")
    monkeypatch.setattr(
        app.state.codex,
        "enqueue",
        lambda *_args, **_kwargs: {"id": "queue-parallel", "status": "running"},
    )

    result = app.state.orchestrator.tick()

    assert result["state"] == "started"
    assert result["task_id"] == task["id"]


def test_parallel_dispatch_waits_when_free_memory_is_insufficient(monkeypatch):
    monkeypatch.setenv("CODEXPI_AGENT_MIN_FREE_MB", "1024")
    monkeypatch.setattr(
        "codexpi.project_orchestrator.psutil.virtual_memory",
        lambda: SimpleNamespace(available=512 * 1024 * 1024),
    )

    assert ProjectOrchestrator.resources_allow_parallel(0)
    assert not ProjectOrchestrator.resources_allow_parallel(1)


def test_mission_queue_resumes_exactly_one_paused_task(logged_in):
    client, app, *_ = logged_in
    projects = client.get("/api/portfolio/projects").json()[:3]
    finished = app.state.portfolio.create_task(
        projects[0]["id"], "Mission terminée", "Construire", "mission:test:finished", priority=1,
    )
    first = app.state.portfolio.create_task(
        projects[1]["id"], "Mission suivante", "Construire", "mission:test:first", priority=2,
    )
    second = app.state.portfolio.create_task(
        projects[2]["id"], "Mission ultérieure", "Construire", "mission:test:second", priority=3,
    )
    app.state.db.execute("UPDATE project_tasks SET status='SUCCEEDED' WHERE id=?", (finished["id"],))
    app.state.db.execute(
        "UPDATE project_tasks SET status='PAUSED' WHERE id IN (?,?)",
        (first["id"], second["id"]),
    )

    resumed = app.state.orchestrator._advance_mission_queue(
        app.state.db.one("SELECT * FROM project_tasks WHERE id=?", (finished["id"],))
    )

    assert resumed == first["id"]
    assert app.state.db.one("SELECT status FROM project_tasks WHERE id=?", (first["id"],))["status"] == "QUEUED"
    assert app.state.db.one("SELECT status FROM project_tasks WHERE id=?", (second["id"],))["status"] == "PAUSED"
    assert app.state.db.one(
        "SELECT COUNT(*) AS n FROM project_events WHERE task_id=? AND kind='task.auto_resumed'",
        (first["id"],),
    )["n"] == 1


def test_non_mission_task_does_not_advance_paused_mission(logged_in):
    client, app, *_ = logged_in
    projects = client.get("/api/portfolio/projects").json()[:2]
    ordinary = app.state.portfolio.create_task(
        projects[0]["id"], "Tâche ordinaire", "Construire", "ordinary:finished", priority=1,
    )
    mission = app.state.portfolio.create_task(
        projects[1]["id"], "Mission", "Construire", "mission:test:paused", priority=2,
    )
    app.state.db.execute("UPDATE project_tasks SET status='PAUSED' WHERE id=?", (mission["id"],))

    assert app.state.orchestrator._advance_mission_queue(ordinary) is None
    assert app.state.db.one("SELECT status FROM project_tasks WHERE id=?", (mission["id"],))["status"] == "PAUSED"


def test_database_trigger_advances_mission_after_restart_safe_terminal_update(logged_in):
    client, app, *_ = logged_in
    projects = client.get("/api/portfolio/projects").json()[:3]
    current = app.state.portfolio.create_task(
        projects[0]["id"], "Mission active", "Construire", "mission:trigger:current", priority=1,
    )
    first = app.state.portfolio.create_task(
        projects[1]["id"], "Mission suivante", "Construire", "mission:trigger:first", priority=2,
    )
    second = app.state.portfolio.create_task(
        projects[2]["id"], "Mission ultérieure", "Construire", "mission:trigger:second", priority=3,
    )
    app.state.db.execute(
        "UPDATE project_tasks SET status='PAUSED' WHERE id IN (?,?)",
        (first["id"], second["id"]),
    )

    app.state.db.execute("UPDATE project_tasks SET status='SUCCEEDED' WHERE id=?", (current["id"],))

    assert app.state.db.one("SELECT status FROM project_tasks WHERE id=?", (first["id"],))["status"] == "QUEUED"
    assert app.state.db.one("SELECT status FROM project_tasks WHERE id=?", (second["id"],))["status"] == "PAUSED"


def test_database_trigger_does_not_cascade_a_global_engine_blocker(logged_in):
    client, app, *_ = logged_in
    projects = client.get("/api/portfolio/projects").json()[:2]
    current = app.state.portfolio.create_task(
        projects[0]["id"], "Mission active", "Construire", "mission:engine:current", priority=1,
    )
    next_task = app.state.portfolio.create_task(
        projects[1]["id"], "Mission suivante", "Construire", "mission:engine:next", priority=2,
    )
    app.state.db.execute("UPDATE project_tasks SET status='PAUSED' WHERE id=?", (next_task["id"],))

    app.state.db.execute(
        "UPDATE project_tasks SET status='BLOCKED',error=? WHERE id=?",
        ("Le moteur Codex est indisponible; le prompt reste conservé dans la file", current["id"]),
    )

    assert app.state.db.one("SELECT status FROM project_tasks WHERE id=?", (next_task["id"],))["status"] == "PAUSED"


def test_boot_recovery_requeues_interrupted_portfolio_task(logged_in):
    client, app, projects, *_ = logged_in
    project = client.get("/api/portfolio/projects").json()[0]
    workspace = projects / "boot-recovery"
    workspace.mkdir()
    workspace_id = str(uuid.uuid4())
    stamp = "2026-10-27T00:00:00+00:00"
    app.state.db.execute(
        "INSERT INTO projects(id,name,path,owner_id,created_at) VALUES(?,?,?,?,?)",
        (workspace_id, "Boot recovery", str(workspace), project.get("owner_id"), stamp),
    )
    conversation = app.state.codex.create_conversation(workspace_id, "V0", user_id="1")
    app.state.db.execute(
        "UPDATE conversations SET portfolio_project_id=?,status='running',worker_pid=999999 WHERE id=?",
        (project["id"], conversation["id"]),
    )
    task = app.state.portfolio.create_task(project["id"], "V0", "Prompt", "boot:recover", max_attempts=3)
    app.state.db.execute(
        "UPDATE project_tasks SET status='RUNNING',conversation_id=?,queue_id='dead-run',attempts=1,lease_owner='old' WHERE id=?",
        (conversation["id"], task["id"]),
    )
    app.state.db.execute(
        "INSERT INTO prompt_queue(id,conversation_id,content,status,created_at) VALUES(?,?,?,?,?)",
        ("dead-run", conversation["id"], "Prompt", "running", stamp),
    )

    result = app.state.orchestrator.recover_interrupted_tasks()

    assert result == {"requeued": 1, "blocked": 0, "preserved": 0}
    stored = app.state.db.one("SELECT status,queue_id,lease_owner,attempts FROM project_tasks WHERE id=?", (task["id"],))
    assert stored == {"status": "QUEUED", "queue_id": None, "lease_owner": None, "attempts": 1}
    assert app.state.db.one("SELECT status FROM prompt_queue WHERE id='dead-run'")["status"] == "failed"


def test_boot_recovery_blocks_after_attempt_limit_without_losing_task(logged_in):
    client, app, *_ = logged_in
    project = client.get("/api/portfolio/projects").json()[0]
    task = app.state.portfolio.create_task(project["id"], "V0", "Prompt conservé", "boot:blocked", max_attempts=2)
    app.state.db.execute(
        "UPDATE project_tasks SET status='RUNNING',attempts=2,queue_id='missing',lease_owner='old' WHERE id=?",
        (task["id"],),
    )

    result = app.state.orchestrator.recover_interrupted_tasks()

    assert result == {"requeued": 0, "blocked": 1, "preserved": 0}
    stored = app.state.db.one("SELECT status,prompt,error FROM project_tasks WHERE id=?", (task["id"],))
    assert stored["status"] == "BLOCKED"
    assert stored["prompt"] == "Prompt conservé"
    assert "redémarrage" in stored["error"].lower()


def test_reconcile_does_not_heartbeat_queued_work_without_a_worker(logged_in):
    client, app, projects, *_ = logged_in
    project = client.get("/api/portfolio/projects").json()[0]
    workspace = projects / "orphaned-queued-worker"
    workspace.mkdir()
    workspace_id = str(uuid.uuid4())
    stamp = "2026-10-27T00:00:00+00:00"
    app.state.db.execute(
        "INSERT INTO projects(id,name,path,owner_id,created_at) VALUES(?,?,?,?,?)",
        (workspace_id, "Orphan", str(workspace), project.get("owner_id"), stamp),
    )
    conversation = app.state.codex.create_conversation(workspace_id, "V0", user_id="1")
    app.state.db.execute(
        "UPDATE conversations SET portfolio_project_id=?,status='failed',worker_pid=NULL WHERE id=?",
        (project["id"], conversation["id"]),
    )
    task = app.state.portfolio.create_task(project["id"], "V0", "Prompt", "queue:orphan", max_attempts=1)
    app.state.db.execute(
        "UPDATE project_tasks SET status='RUNNING',conversation_id=?,queue_id='orphan-queue',attempts=1 WHERE id=?",
        (conversation["id"], task["id"]),
    )
    app.state.db.execute(
        "INSERT INTO prompt_queue(id,conversation_id,content,status,created_at) VALUES(?,?,?,?,?)",
        ("orphan-queue", conversation["id"], "Prompt", "queued", stamp),
    )

    assert app.state.orchestrator._reconcile(app.state.db.one("SELECT * FROM project_tasks WHERE id=?", (task["id"],)))
    stored = app.state.db.one("SELECT status,error FROM project_tasks WHERE id=?", (task["id"],))
    assert stored["status"] == "FAILED"
    assert "interrompu" in stored["error"]
    assert app.state.db.one("SELECT status FROM prompt_queue WHERE id='orphan-queue'")["status"] == "failed"