"""Tests for the Phase 9 Commit A interaction capture loop.""" import time import pytest from fastapi.testclient import TestClient from atocore.interactions.service import ( get_interaction, list_interactions, record_interaction, ) from atocore.main import app from atocore.models.database import init_db # --- Service-level tests -------------------------------------------------- def test_record_interaction_persists_all_fields(tmp_data_dir): init_db() interaction = record_interaction( prompt="What is the lateral support material for p05?", response="The current lateral support uses GF-PTFE pads per Decision D-024.", response_summary="lateral support: GF-PTFE per D-024", project="p05-interferometer", client="claude-code", session_id="sess-001", memories_used=["mem-aaa", "mem-bbb"], chunks_used=["chunk-111", "chunk-222", "chunk-333"], context_pack={"budget": 3000, "chunks": 3}, ) assert interaction.id assert interaction.created_at fetched = get_interaction(interaction.id) assert fetched is not None assert fetched.prompt.startswith("What is the lateral support") assert fetched.response.startswith("The current lateral support") assert fetched.response_summary == "lateral support: GF-PTFE per D-024" assert fetched.project == "p05-interferometer" assert fetched.client == "claude-code" assert fetched.session_id == "sess-001" assert fetched.memories_used == ["mem-aaa", "mem-bbb"] assert fetched.chunks_used == ["chunk-111", "chunk-222", "chunk-333"] assert fetched.context_pack == {"budget": 3000, "chunks": 3} def test_record_interaction_minimum_fields(tmp_data_dir): init_db() interaction = record_interaction(prompt="ping") assert interaction.id assert interaction.prompt == "ping" assert interaction.response == "" assert interaction.memories_used == [] assert interaction.chunks_used == [] def test_record_interaction_rejects_empty_prompt(tmp_data_dir): init_db() with pytest.raises(ValueError): record_interaction(prompt="") with pytest.raises(ValueError): record_interaction(prompt=" ") def test_get_interaction_returns_none_for_unknown_id(tmp_data_dir): init_db() assert get_interaction("does-not-exist") is None assert get_interaction("") is None def test_list_interactions_filters_by_project(tmp_data_dir): init_db() record_interaction(prompt="p04 question", project="p04-gigabit") record_interaction(prompt="p05 question", project="p05-interferometer") record_interaction(prompt="another p05", project="p05-interferometer") p05 = list_interactions(project="p05-interferometer") p04 = list_interactions(project="p04-gigabit") assert len(p05) == 2 assert len(p04) == 1 assert all(i.project == "p05-interferometer" for i in p05) assert p04[0].prompt == "p04 question" def test_list_interactions_filters_by_session_and_client(tmp_data_dir): init_db() record_interaction(prompt="a", session_id="sess-A", client="openclaw") record_interaction(prompt="b", session_id="sess-A", client="claude-code") record_interaction(prompt="c", session_id="sess-B", client="openclaw") sess_a = list_interactions(session_id="sess-A") openclaw = list_interactions(client="openclaw") assert len(sess_a) == 2 assert len(openclaw) == 2 assert {i.client for i in sess_a} == {"openclaw", "claude-code"} def test_list_interactions_orders_newest_first_and_respects_limit(tmp_data_dir): init_db() # created_at has 1-second resolution; sleep enough to keep ordering # deterministic regardless of insert speed. for index in range(5): record_interaction(prompt=f"prompt-{index}") time.sleep(1.05) items = list_interactions(limit=3) assert len(items) == 3 # Newest first: prompt-4, prompt-3, prompt-2 assert items[0].prompt == "prompt-4" assert items[1].prompt == "prompt-3" assert items[2].prompt == "prompt-2" def test_list_interactions_respects_since_filter(tmp_data_dir): init_db() first = record_interaction(prompt="early") time.sleep(1.05) second = record_interaction(prompt="late") after_first = list_interactions(since=first.created_at) ids_after_first = {item.id for item in after_first} assert second.id in ids_after_first assert first.id in ids_after_first # cutoff is inclusive after_second = list_interactions(since=second.created_at) ids_after_second = {item.id for item in after_second} assert second.id in ids_after_second assert first.id not in ids_after_second def test_list_interactions_zero_limit_returns_empty(tmp_data_dir): init_db() record_interaction(prompt="ping") assert list_interactions(limit=0) == [] # --- API-level tests ------------------------------------------------------ def test_post_interactions_endpoint_records_interaction(tmp_data_dir): init_db() client = TestClient(app) response = client.post( "/interactions", json={ "prompt": "What changed in p06 this week?", "response": "Polisher kinematic frame parameters updated to v0.3.", "response_summary": "p06 frame parameters bumped to v0.3", "project": "p06-polisher", "client": "claude-code", "session_id": "sess-xyz", "memories_used": ["mem-1"], "chunks_used": ["chunk-a", "chunk-b"], "context_pack": {"chunks": 2}, }, ) assert response.status_code == 200 body = response.json() assert body["status"] == "recorded" interaction_id = body["id"] # Round-trip via the GET endpoint fetched = client.get(f"/interactions/{interaction_id}") assert fetched.status_code == 200 fetched_body = fetched.json() assert fetched_body["prompt"].startswith("What changed in p06") assert fetched_body["response"].startswith("Polisher kinematic frame") assert fetched_body["project"] == "p06-polisher" assert fetched_body["chunks_used"] == ["chunk-a", "chunk-b"] assert fetched_body["context_pack"] == {"chunks": 2} def test_post_interactions_rejects_empty_prompt(tmp_data_dir): init_db() client = TestClient(app) response = client.post("/interactions", json={"prompt": ""}) assert response.status_code == 400 def test_get_unknown_interaction_returns_404(tmp_data_dir): init_db() client = TestClient(app) response = client.get("/interactions/does-not-exist") assert response.status_code == 404 def test_list_interactions_endpoint_returns_summaries(tmp_data_dir): init_db() client = TestClient(app) client.post( "/interactions", json={"prompt": "alpha", "project": "p04-gigabit", "response": "x" * 10}, ) client.post( "/interactions", json={"prompt": "beta", "project": "p05-interferometer", "response": "y" * 50}, ) response = client.get("/interactions", params={"project": "p05-interferometer"}) assert response.status_code == 200 body = response.json() assert body["count"] == 1 assert body["interactions"][0]["prompt"] == "beta" assert body["interactions"][0]["response_chars"] == 50 # The list endpoint never includes the full response body assert "response" not in body["interactions"][0]