399 lines
19 KiB
Python
399 lines
19 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import sqlite3
|
||
import uuid
|
||
|
||
import pytest
|
||
|
||
from codexpi.agent_protocol import parse_decision_request, parse_specialist_call, record_agent_outcome
|
||
from codexpi.codex_service import CodexManager
|
||
from codexpi.coordination_service import CoordinationError, CoordinationService
|
||
from codexpi.database import Database, utcnow
|
||
from codexpi.session_worker import _background_window_open, concurrent_agent_limit
|
||
|
||
|
||
def test_existing_queue_schema_is_migrated_before_priority_index(tmp_path):
|
||
path = tmp_path / "legacy.db"
|
||
database = Database(path)
|
||
database.initialize()
|
||
with sqlite3.connect(path) as conn:
|
||
conn.execute("DROP INDEX idx_prompt_queue_priority")
|
||
conn.execute("ALTER TABLE prompt_queue DROP COLUMN slot_acquired_at")
|
||
conn.execute("ALTER TABLE prompt_queue DROP COLUMN origin_kind")
|
||
conn.execute("ALTER TABLE prompt_queue DROP COLUMN delegation_depth")
|
||
conn.execute("ALTER TABLE prompt_queue DROP COLUMN priority")
|
||
|
||
database.initialize()
|
||
|
||
with sqlite3.connect(path) as conn:
|
||
columns = {row[1] for row in conn.execute("PRAGMA table_info(prompt_queue)")}
|
||
indexes = {row[1] for row in conn.execute("PRAGMA index_list(prompt_queue)")}
|
||
assert {"priority", "delegation_depth", "origin_kind", "slot_acquired_at"} <= columns
|
||
assert "idx_prompt_queue_priority" in indexes
|
||
|
||
|
||
def environment(tmp_path):
|
||
database = Database(tmp_path / "codexpi.db")
|
||
database.initialize()
|
||
owner = "owner-1"
|
||
project_path = tmp_path / "project"
|
||
self_path = tmp_path / "self"
|
||
project_path.mkdir()
|
||
self_path.mkdir()
|
||
project_id = str(uuid.uuid4())
|
||
self_id = str(uuid.uuid4())
|
||
database.execute(
|
||
"INSERT INTO projects(id,name,path,owner_id,full_access,created_at) VALUES(?,?,?,?,?,?)",
|
||
(project_id, "Projet source", str(project_path), owner, 0, utcnow()),
|
||
)
|
||
database.execute(
|
||
"INSERT INTO projects(id,name,path,owner_id,full_access,execution_group,created_at) VALUES(?,?,?,?,?,?,?)",
|
||
(self_id, "CodexPi (Self)", str(self_path), owner, 1, "__global_agent__", utcnow()),
|
||
)
|
||
codex = CodexManager(database, tmp_path / "data", "/bin/true")
|
||
codex._spawn = lambda *args, **kwargs: 4242
|
||
service = CoordinationService(database, codex, tmp_path / "data")
|
||
return database, codex, service, owner, project_id
|
||
|
||
|
||
def completed_output(database: Database, conversation_id: str, queue_id: str, content: str) -> None:
|
||
database.execute(
|
||
"UPDATE prompt_queue SET status='done',finished_at=? WHERE id=?",
|
||
(utcnow(), queue_id),
|
||
)
|
||
database.execute(
|
||
"UPDATE conversations SET status='completed',worker_pid=NULL,updated_at=? WHERE id=?",
|
||
(utcnow(), conversation_id),
|
||
)
|
||
database.execute(
|
||
"INSERT INTO messages(id,conversation_id,role,content,metadata,created_at) VALUES(?,?,?,?,?,?)",
|
||
(str(uuid.uuid4()), conversation_id, "assistant", content, json.dumps({"queue_id": queue_id}), utcnow()),
|
||
)
|
||
|
||
|
||
def test_coordination_markers_are_strict():
|
||
decision = parse_decision_request(
|
||
'CODEXPI_DECISION {"title":"Déploiement","question":"Quelle cible ?","options":['
|
||
'{"id":"staging","label":"Staging"},{"id":"production","label":"Production"}],'
|
||
'"allow_multiple":false,"allow_custom":true}'
|
||
)
|
||
assert decision and decision["question"] == "Quelle cible ?"
|
||
assert len(decision["options"]) == 2
|
||
assert parse_decision_request('CODEXPI_DECISION {"question":"Sans choix","options":[]}') is None
|
||
assert parse_specialist_call(
|
||
'CODEXPI_SPECIALIST_CALL {"capability":"design.interface","task":"Auditer les écrans mobiles", "context":"PWA"}'
|
||
) == {"capability": "design.interface", "task": "Auditer les écrans mobiles", "context": "PWA"}
|
||
|
||
|
||
def test_coordination_http_endpoints_are_owner_scoped_and_csrf_protected(logged_in, monkeypatch):
|
||
client, app, _, _, headers = logged_in
|
||
monkeypatch.setattr(app.state.codex, "_spawn", lambda *args, **kwargs: 4242)
|
||
monkeypatch.setattr(app.state.codex, "detect", lambda: "/bin/true")
|
||
|
||
assert client.put(
|
||
"/api/coordination/runtime-policies",
|
||
json={
|
||
"max_concurrent": 2, "load_pause_percent": 85, "memory_pause_percent": 88,
|
||
"background_start_hour": 22, "background_end_hour": 7,
|
||
},
|
||
).status_code == 403
|
||
saved = client.put(
|
||
"/api/coordination/runtime-policies", headers=headers,
|
||
json={
|
||
"max_concurrent": 2, "load_pause_percent": 85, "memory_pause_percent": 88,
|
||
"background_start_hour": 22, "background_end_hour": 7,
|
||
},
|
||
)
|
||
assert saved.status_code == 200
|
||
assert client.get("/api/coordination/runtime-policies").json()["global"]["max_concurrent"] == 2
|
||
overview = client.get("/api/coordination/overview")
|
||
assert overview.status_code == 200
|
||
assert set(overview.json()) == {"decisions", "incidents", "runtime_policies", "specialists"}
|
||
|
||
preferences = client.put(
|
||
"/api/notifications/preferences", headers=headers,
|
||
json={"task_completion": False, "decisions": True, "incidents": True},
|
||
)
|
||
assert preferences.status_code == 200
|
||
assert preferences.json() == {"task_completion": False, "decisions": True, "incidents": True}
|
||
|
||
created = client.post(
|
||
"/api/coordination/specialists", headers=headers,
|
||
json={
|
||
"name": "Studio Interface",
|
||
"description": "Conçoit des interfaces PWA accessibles et responsive pour les projets.",
|
||
"capabilities": ["design.interface", "design.ux"],
|
||
},
|
||
)
|
||
assert created.status_code == 201
|
||
assert created.json()["status"] == "BUILDING"
|
||
assert created.json()["full_access"] is False
|
||
assert len(client.get("/api/coordination/specialists").json()) == 1
|
||
|
||
|
||
def test_decision_answer_resumes_the_same_conversation(tmp_path):
|
||
database, codex, service, owner, project_id = environment(tmp_path)
|
||
project = database.one("SELECT * FROM projects WHERE id=?", (project_id,))
|
||
conversation = codex.create_conversation(project_id, "Travail", user_id=owner)
|
||
execution = codex.enqueue(conversation, project, "Prépare le changement", requested_by=owner)
|
||
completed_output(database, conversation["id"], execution["id"], "Réponse provisoire")
|
||
marker = (
|
||
'CODEXPI_DECISION {"title":"Choix","question":"Quelle option ?","options":['
|
||
'{"id":"a","label":"Option A"},{"id":"b","label":"Option B"}],'
|
||
'"allow_multiple":false,"allow_custom":true}'
|
||
)
|
||
with database.connect() as conn:
|
||
record_agent_outcome(conn, conversation["id"], execution["id"], "done", marker, None)
|
||
decision = service.decisions(owner)[0]
|
||
|
||
answered = service.answer_decision(owner, decision["id"], ["a"], "")
|
||
|
||
assert answered["status"] == "ANSWERED"
|
||
response = database.one("SELECT * FROM prompt_queue WHERE id=?", (answered["response_queue_id"],))
|
||
assert response["conversation_id"] == conversation["id"]
|
||
assert response["origin_kind"] == "decision_answer"
|
||
assert "Option A" in response["content"]
|
||
with pytest.raises(CoordinationError, match="déjà"):
|
||
service.answer_decision(owner, decision["id"], ["b"], "")
|
||
|
||
|
||
def test_failed_agent_becomes_a_self_repairable_incident(tmp_path):
|
||
database, codex, service, owner, project_id = environment(tmp_path)
|
||
project = database.one("SELECT * FROM projects WHERE id=?", (project_id,))
|
||
conversation = codex.create_conversation(project_id, "Travail", user_id=owner)
|
||
execution = codex.enqueue(conversation, project, "Exécute", requested_by=owner)
|
||
with database.connect() as conn:
|
||
record_agent_outcome(
|
||
conn, conversation["id"], execution["id"], "failed", None,
|
||
"Le worker s’est arrêté avec une exception inattendue",
|
||
)
|
||
|
||
incident = service.incidents(owner)[0]
|
||
assert incident["status"] == "OPEN"
|
||
assert incident["auto_repair"] is True
|
||
service.tick()
|
||
repairing = service.incidents(owner)[0]
|
||
assert repairing["status"] == "REPAIRING"
|
||
assert repairing["repair_project_name"] == "CodexPi (Self)"
|
||
|
||
|
||
def test_specialist_builds_itself_and_receives_a_delegation(tmp_path):
|
||
database, codex, service, owner, project_id = environment(tmp_path)
|
||
specialist = service.create_specialist(
|
||
owner, "Studio Interface", "Conçoit et audite les interfaces mobiles PWA des projets.", ["design.interface"],
|
||
)
|
||
assert specialist["status"] == "BUILDING"
|
||
assert specialist["full_access"] is False
|
||
assert (service.data_dir / "specialists").is_dir()
|
||
completed_output(
|
||
database, specialist["bootstrap_conversation_id"], specialist["bootstrap_queue_id"],
|
||
'CODEXPI_SPECIALIST_READY {"summary":"Prêt","capabilities":["design.ux"]}',
|
||
)
|
||
service.tick()
|
||
specialist = service.specialist(owner, specialist["id"])
|
||
assert specialist["status"] == "ACTIVE"
|
||
assert "design.interface" in specialist["capabilities"]
|
||
assert "design.ux" in specialist["capabilities"]
|
||
|
||
source_project = database.one("SELECT * FROM projects WHERE id=?", (project_id,))
|
||
source_conversation = codex.create_conversation(project_id, "Refonte", user_id=owner)
|
||
source_execution = codex.enqueue(source_conversation, source_project, "Refonds la page", requested_by=owner)
|
||
completed_output(
|
||
database, source_conversation["id"], source_execution["id"],
|
||
'CODEXPI_SPECIALIST_CALL {"capability":"design.interface","task":"Préparer la maquette mobile",'
|
||
'"context":"Respecter le thème crème"}',
|
||
)
|
||
with database.connect() as conn:
|
||
record_agent_outcome(
|
||
conn, source_conversation["id"], source_execution["id"], "done",
|
||
'CODEXPI_SPECIALIST_CALL {"capability":"design.interface","task":"Préparer la maquette mobile",'
|
||
'"context":"Respecter le thème crème"}', None,
|
||
)
|
||
service.tick()
|
||
delegation = database.one("SELECT * FROM specialist_delegations WHERE source_queue_id=?", (source_execution["id"],))
|
||
assert delegation["status"] == "RUNNING"
|
||
completed_output(
|
||
database, delegation["specialist_conversation_id"], delegation["queue_id"],
|
||
"Maquette prête dans output/mobile-interface.svg",
|
||
)
|
||
service.tick()
|
||
delegation = database.one("SELECT * FROM specialist_delegations WHERE id=?", (delegation["id"],))
|
||
assert delegation["status"] == "COMPLETED"
|
||
follow_up = database.one(
|
||
"SELECT * FROM prompt_queue WHERE conversation_id=? AND origin_kind='specialist_result' ORDER BY created_at DESC LIMIT 1",
|
||
(source_conversation["id"],),
|
||
)
|
||
assert follow_up and "mobile-interface.svg" in follow_up["content"]
|
||
|
||
|
||
def test_failed_specialist_can_rebuild_without_replacing_its_workspace(tmp_path):
|
||
database, _, service, owner, _ = environment(tmp_path)
|
||
specialist = service.create_specialist(
|
||
owner, "Studio Logo", "Conçoit des identités visuelles et des logos cohérents pour les projets.", ["design.logo"],
|
||
)
|
||
workspace = database.one("SELECT path FROM projects WHERE id=?", (specialist["project_id"],))["path"]
|
||
database.execute(
|
||
"UPDATE specialist_agents SET status='FAILED',error='incident de construction' WHERE id=?",
|
||
(specialist["id"],),
|
||
)
|
||
|
||
rebuilt = service.rebuild_specialist(owner, specialist["id"])
|
||
|
||
assert rebuilt["status"] == "BUILDING"
|
||
assert rebuilt["error"] == ""
|
||
assert rebuilt["bootstrap_conversation_id"] != specialist["bootstrap_conversation_id"]
|
||
assert database.one("SELECT path FROM projects WHERE id=?", (specialist["project_id"],))["path"] == workspace
|
||
|
||
|
||
def test_two_specialists_can_collaborate_and_return_to_source(tmp_path):
|
||
database, codex, service, owner, project_id = environment(tmp_path)
|
||
interface = service.create_specialist(
|
||
owner, "Studio Interface", "Conçoit les interfaces et coordonne leur identité visuelle.", ["design.interface"],
|
||
)
|
||
logo = service.create_specialist(
|
||
owner, "Studio Logo", "Conçoit les logos et systèmes d’identité pour les interfaces.", ["design.logo"],
|
||
)
|
||
for specialist, capability in ((interface, "design.interface"), (logo, "design.logo")):
|
||
completed_output(
|
||
database, specialist["bootstrap_conversation_id"], specialist["bootstrap_queue_id"],
|
||
f'CODEXPI_SPECIALIST_READY {{"summary":"Prêt","capabilities":["{capability}"]}}',
|
||
)
|
||
service.tick()
|
||
|
||
project = database.one("SELECT * FROM projects WHERE id=?", (project_id,))
|
||
source_conversation = codex.create_conversation(project_id, "Identité", user_id=owner)
|
||
source_queue = codex.enqueue(source_conversation, project, "Refonds la marque", requested_by=owner)
|
||
first_call = (
|
||
'CODEXPI_SPECIALIST_CALL {"capability":"design.interface","task":"Préparer la direction UI",'
|
||
'"context":"Coordonner le logo"}'
|
||
)
|
||
completed_output(database, source_conversation["id"], source_queue["id"], first_call)
|
||
with database.connect() as conn:
|
||
record_agent_outcome(conn, source_conversation["id"], source_queue["id"], "done", first_call, None)
|
||
service.tick()
|
||
parent = database.one("SELECT * FROM specialist_delegations WHERE source_queue_id=?", (source_queue["id"],))
|
||
|
||
second_call = (
|
||
'CODEXPI_SPECIALIST_CALL {"capability":"design.logo","task":"Produire le symbole de marque",'
|
||
'"context":"Doit fonctionner avec la direction UI"}'
|
||
)
|
||
completed_output(database, parent["specialist_conversation_id"], parent["queue_id"], second_call)
|
||
with database.connect() as conn:
|
||
record_agent_outcome(
|
||
conn, parent["specialist_conversation_id"], parent["queue_id"], "done", second_call, None,
|
||
)
|
||
assert database.one("SELECT status FROM specialist_delegations WHERE id=?", (parent["id"],))["status"] == "WAITING_CHILD"
|
||
service.tick()
|
||
child = database.one("SELECT * FROM specialist_delegations WHERE parent_delegation_id=?", (parent["id"],))
|
||
assert child["status"] == "RUNNING"
|
||
|
||
completed_output(
|
||
database, child["specialist_conversation_id"], child["queue_id"],
|
||
"Logo prêt dans output/logo.svg",
|
||
)
|
||
service.tick()
|
||
resumed_parent = database.one("SELECT * FROM specialist_delegations WHERE id=?", (parent["id"],))
|
||
assert resumed_parent["status"] == "RUNNING"
|
||
completed_output(
|
||
database, parent["specialist_conversation_id"], resumed_parent["queue_id"],
|
||
"Interface et logo coordonnés dans output/ui-kit.svg",
|
||
)
|
||
service.tick()
|
||
assert database.one("SELECT status FROM specialist_delegations WHERE id=?", (parent["id"],))["status"] == "COMPLETED"
|
||
delivered = database.one(
|
||
"SELECT content FROM prompt_queue WHERE conversation_id=? AND origin_kind='specialist_result' ORDER BY created_at DESC LIMIT 1",
|
||
(source_conversation["id"],),
|
||
)
|
||
assert delivered and "ui-kit.svg" in delivered["content"]
|
||
|
||
|
||
def test_decision_inside_specialist_resumes_delegation_chain(tmp_path):
|
||
database, codex, service, owner, project_id = environment(tmp_path)
|
||
specialist = service.create_specialist(
|
||
owner, "Studio Interface", "Conçoit les interfaces mobiles et leurs règles d’intégration.", ["design.interface"],
|
||
)
|
||
completed_output(
|
||
database, specialist["bootstrap_conversation_id"], specialist["bootstrap_queue_id"],
|
||
'CODEXPI_SPECIALIST_READY {"summary":"Prêt","capabilities":["design.interface"]}',
|
||
)
|
||
service.tick()
|
||
project = database.one("SELECT * FROM projects WHERE id=?", (project_id,))
|
||
conversation = codex.create_conversation(project_id, "Refonte", user_id=owner)
|
||
source = codex.enqueue(conversation, project, "Refonds la page", requested_by=owner)
|
||
call = (
|
||
'CODEXPI_SPECIALIST_CALL {"capability":"design.interface","task":"Préparer deux directions visuelles",'
|
||
'"context":"PWA mobile"}'
|
||
)
|
||
completed_output(database, conversation["id"], source["id"], call)
|
||
with database.connect() as conn:
|
||
record_agent_outcome(conn, conversation["id"], source["id"], "done", call, None)
|
||
service.tick()
|
||
delegation = database.one("SELECT * FROM specialist_delegations WHERE source_queue_id=?", (source["id"],))
|
||
decision_marker = (
|
||
'CODEXPI_DECISION {"title":"Direction","question":"Quelle direction retenir ?","options":['
|
||
'{"id":"calme","label":"Calme"},{"id":"dense","label":"Dense"}],'
|
||
'"allow_multiple":false,"allow_custom":true}'
|
||
)
|
||
completed_output(database, delegation["specialist_conversation_id"], delegation["queue_id"], decision_marker)
|
||
with database.connect() as conn:
|
||
record_agent_outcome(
|
||
conn, delegation["specialist_conversation_id"], delegation["queue_id"], "done", decision_marker, None,
|
||
)
|
||
waiting = database.one("SELECT * FROM specialist_delegations WHERE id=?", (delegation["id"],))
|
||
assert waiting["status"] == "WAITING_CHILD"
|
||
|
||
decision = service.decisions(owner)[0]
|
||
answered = service.answer_decision(owner, decision["id"], ["calme"], "")
|
||
resumed = database.one("SELECT * FROM specialist_delegations WHERE id=?", (delegation["id"],))
|
||
assert resumed["status"] == "RUNNING"
|
||
assert resumed["queue_id"] == answered["response_queue_id"]
|
||
|
||
completed_output(
|
||
database, delegation["specialist_conversation_id"], answered["response_queue_id"],
|
||
"Direction calme préparée dans output/interface-calme.svg",
|
||
)
|
||
service.tick()
|
||
completed = database.one("SELECT * FROM specialist_delegations WHERE id=?", (delegation["id"],))
|
||
assert completed["status"] == "COMPLETED"
|
||
follow_up = database.one(
|
||
"SELECT content FROM prompt_queue WHERE conversation_id=? AND origin_kind='specialist_result' ORDER BY created_at DESC LIMIT 1",
|
||
(conversation["id"],),
|
||
)
|
||
assert follow_up and "interface-calme.svg" in follow_up["content"]
|
||
|
||
|
||
def test_runtime_policy_is_dynamic_and_background_window_supports_midnight(tmp_path):
|
||
database, codex, service, owner, project_id = environment(tmp_path)
|
||
service.update_runtime_policy(owner, {
|
||
"max_concurrent": 2, "load_pause_percent": 85, "memory_pause_percent": 88,
|
||
"background_start_hour": 22, "background_end_hour": 7,
|
||
})
|
||
assert concurrent_agent_limit(str(database.path), project_id) == 2
|
||
legacy_id = str(uuid.uuid4())
|
||
legacy_path = tmp_path / "legacy-project"
|
||
legacy_path.mkdir()
|
||
database.execute(
|
||
"INSERT INTO projects(id,name,path,owner_id,created_at) VALUES(?,?,?,?,?)",
|
||
(legacy_id, "Projet historique", str(legacy_path), None, utcnow()),
|
||
)
|
||
legacy_project = database.one("SELECT * FROM projects WHERE id=?", (legacy_id,))
|
||
legacy_conversation = codex.create_conversation(legacy_id, "Travail", user_id=owner)
|
||
legacy_queue = codex.enqueue(legacy_conversation, legacy_project, "Analyse", requested_by=owner)
|
||
assert concurrent_agent_limit(str(database.path), legacy_id, legacy_queue["id"]) == 2
|
||
assert any(item["id"] == legacy_id for item in service.runtime_policies(owner)["projects"])
|
||
assert _background_window_open(22, 7, 23) is True
|
||
assert _background_window_open(22, 7, 5) is True
|
||
assert _background_window_open(22, 7, 12) is False
|
||
|
||
|
||
def test_coordination_lease_prevents_double_processing_during_blue_green(tmp_path):
|
||
database, codex, service, owner, _ = environment(tmp_path)
|
||
standby = CoordinationService(database, codex, service.data_dir)
|
||
assert service._acquire_lease() is True
|
||
try:
|
||
assert standby.tick() == {"state": "standby"}
|
||
finally:
|
||
service._release_lease()
|
||
assert standby.tick() == {"state": "processed"}
|