Spaces:
Sleeping
Sleeping
| """Agent orchestrator (stub/scripted path — CI never needs a model).""" | |
| import base64 | |
| import json | |
| from calendar_out.ics import events_to_ics | |
| from server.orchestrator import run_orchestrator | |
| from server.schema import Event | |
| THREAD = ( | |
| "Room parent: picture day is Thursday June 18 at 9am, wear green!\n" | |
| "Me: adding it now" | |
| ) | |
| def _cal_b64() -> str: | |
| ics = events_to_ics([ | |
| Event(title="Standup", start="2026-06-18T09:00:00", end="2026-06-18T09:30:00"), | |
| ]) | |
| return base64.b64encode(ics).decode("ascii") | |
| def test_orchestrator_scripted_end_to_end(): | |
| steps = list(run_orchestrator(THREAD)) | |
| kinds = [s["kind"] for s in steps] | |
| assert kinds[0] == "plan" | |
| assert "tool_call" in kinds and "tool_result" in kinds | |
| assert kinds[-1] == "final" | |
| final = steps[-1] | |
| assert final["summary"]["events"] >= 1 | |
| assert final["plan"]["events"], "extracted at least one event" | |
| # the agent rendered a real .ics | |
| assert final["ics_base64"] | |
| assert b"BEGIN:VCALENDAR" in base64.b64decode(final["ics_base64"]) | |
| # every step must survive the JSON boundary (UI/trace serialisation) | |
| json.dumps(steps) | |
| def test_orchestrator_checks_conflicts_when_calendar_given(): | |
| steps = list(run_orchestrator(THREAD, ics_b64=_cal_b64())) | |
| tools = [s.get("tool") for s in steps if s["kind"] == "tool_call"] | |
| assert "check_conflicts" in tools | |
| assert steps[-1]["kind"] == "final" | |
| def test_orchestrator_skips_conflicts_without_calendar(): | |
| steps = list(run_orchestrator(THREAD)) | |
| tools = [s.get("tool") for s in steps if s["kind"] == "tool_call"] | |
| assert "check_conflicts" not in tools | |
| def test_orchestrator_handles_eventless_thread(): | |
| steps = list(run_orchestrator("ok thanks!! see you")) | |
| final = steps[-1] | |
| assert final["kind"] == "final" | |
| assert final["summary"]["events"] == 0 | |
| assert final["ics_base64"] is None | |
| def test_extract_events_tool_accepts_memory(): | |
| from server.mcp_tools import extract_events | |
| plan = extract_events(THREAD, memory="Dana is the soccer coach") | |
| assert isinstance(plan, dict) and "events" in plan | |
| def test_homepage_analyze_runs_the_agentic_workflow(): | |
| """The homepage's _on_analyze is orchestrator-driven: the thinking panel | |
| carries the agent's step trace, results keep the legacy 5-tuple contract.""" | |
| from ui.blocks import _on_analyze | |
| *_, last = list(_on_analyze(THREAD, None, None)) | |
| trace, rows, plan_md, reply, status = last | |
| assert "ag-final" in trace and "ag-tool_call" in trace | |
| assert rows and rows[0][0] | |
| assert status.startswith("Found ") | |