562 lines
24 KiB
Python
562 lines
24 KiB
Python
|
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"
|